Subscription.h

Go to the documentation of this file.

 
#include "RobotRaconteur/Client.h"
 
#include <boost/tuple/tuple_comparison.hpp>
#include <boost/unordered_set.hpp>
 
#pragma once
 
namespace RobotRaconteur
{
namespace detail
{
class Discovery;
class Discovery_nodestorage;
class ServiceInfo2Subscription_client;
class ServiceSubscription_client;
class ServiceSubscription_retrytimer;
class WireSubscription_connection;
class PipeSubscription_connection;
class WireSubscription_send_iterator;
class PipeSubscription_send_iterator;
class ServiceSubscription_custom_member_subscribers;
class ServiceSubscriptionManager_subscription;
} // namespace detail
 
class ROBOTRACONTEUR_CORE_API WireConnectionBase;
template <typename T>
class WireConnection;
class ROBOTRACONTEUR_CORE_API WireSubscriptionBase;
template <typename T>
class WireSubscription;
 
class ROBOTRACONTEUR_CORE_API PipeEndpointBase;
template <typename T>
class PipeEndpoint;
class ROBOTRACONTEUR_CORE_API PipeSubscriptionBase;
template <typename T>
class PipeSubscription;
class ROBOTRACONTEUR_CORE_API SubObjectSubscription;
class ROBOTRACONTEUR_CORE_API ServiceSubscriptionManager;

class ROBOTRACONTEUR_CORE_API ServiceSubscriptionFilterNode
{
  public:
    ::RobotRaconteur::NodeID NodeID;
    std::string NodeName;
    std::string Username;
    RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> > Credentials;
};

class ROBOTRACONTEUR_CORE_API ServiceSubscriptionFilterAttribute
{
  public:
    std::string Name;
    std::string Value;
    boost::regex ValueRegex;
    bool UseRegex;

    ServiceSubscriptionFilterAttribute();
    ServiceSubscriptionFilterAttribute(boost::string_ref value);
    ServiceSubscriptionFilterAttribute(const boost::regex& value_regex);
    ServiceSubscriptionFilterAttribute(boost::string_ref name, boost::string_ref value);

    ServiceSubscriptionFilterAttribute(boost::string_ref name, const boost::regex& value_regex);

    bool IsMatch(boost::string_ref value) const;
    bool IsMatch(boost::string_ref name, boost::string_ref value) const;
    bool IsMatch(const std::vector<std::string>& values) const;
    bool IsMatch(const RR_INTRUSIVE_PTR<RRList<RRValue> >& values) const;
    bool IsMatch(const RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> >& values) const;
    bool IsMatch(const std::map<std::string, std::string>& values) const;
};

ROBOTRACONTEUR_CORE_API ServiceSubscriptionFilterAttribute
CreateServiceSubscriptionFilterAttributeRegex(boost::string_ref regex_value);
ROBOTRACONTEUR_CORE_API ServiceSubscriptionFilterAttribute
CreateServiceSubscriptionFilterAttributeRegex(boost::string_ref name, boost::string_ref regex_value);

ROBOTRACONTEUR_CORE_API
ServiceSubscriptionFilterAttribute CreateServiceSubscriptionFilterAttributeCombinedIdentifier(
    boost::string_ref combined_identifier);
ROBOTRACONTEUR_CORE_API ServiceSubscriptionFilterAttribute
CreateServiceSubscriptionFilterAttributeIdentifier(boost::string_ref identifier_name, boost::string_ref uuid_string);
ROBOTRACONTEUR_CORE_API ServiceSubscriptionFilterAttribute CreateServiceSubscriptionFilterAttributeIdentifier(
    boost::string_ref name, boost::string_ref identifier_name, boost::string_ref uuid_string);

enum ServiceSubscriptionFilterAttributeGroupOperation
{
    ServiceSubscriptionFilterAttributeGroupOperation_OR,
    ServiceSubscriptionFilterAttributeGroupOperation_AND,
    ServiceSubscriptionFilterAttributeGroupOperation_NOR,
    ServiceSubscriptionFilterAttributeGroupOperation_NAND
};

class ROBOTRACONTEUR_CORE_API ServiceSubscriptionFilterAttributeGroup
{
  public:
    std::vector<ServiceSubscriptionFilterAttribute> Attributes;
    std::vector<ServiceSubscriptionFilterAttributeGroup> Groups;
    ServiceSubscriptionFilterAttributeGroupOperation Operation;

    bool SplitStringAttribute;
    char SplitStringDelimiter;

    ServiceSubscriptionFilterAttributeGroup();
    ServiceSubscriptionFilterAttributeGroup(ServiceSubscriptionFilterAttributeGroupOperation operation);
    ServiceSubscriptionFilterAttributeGroup(ServiceSubscriptionFilterAttributeGroupOperation operation,
                                            std::vector<ServiceSubscriptionFilterAttribute> attributes);
    ServiceSubscriptionFilterAttributeGroup(ServiceSubscriptionFilterAttributeGroupOperation operation,
                                            std::vector<ServiceSubscriptionFilterAttributeGroup> groups);

    bool IsMatch(boost::string_ref value) const;

    bool IsMatch(RR_INTRUSIVE_PTR<RRArray<char> >& value) const;

    bool IsMatch(const std::vector<std::string>& values) const;
    bool IsMatch(const RR_INTRUSIVE_PTR<RRList<RRValue> >& values) const;
    bool IsMatch(const RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> >& values) const;
    bool IsMatch(const std::map<std::string, std::string>& values) const;
 
    bool IsMatch(const RR_INTRUSIVE_PTR<RRValue>& value) const;
};

class ROBOTRACONTEUR_CORE_API ServiceSubscriptionFilter
{
  public:
    std::vector<RR_SHARED_PTR<ServiceSubscriptionFilterNode> > Nodes;
    std::vector<std::string> ServiceNames;
    std::vector<std::string> TransportSchemes;
    std::map<std::string, ServiceSubscriptionFilterAttributeGroup> Attributes;
    ServiceSubscriptionFilterAttributeGroupOperation AttributesMatchOperation;
    boost::function<bool(const ServiceInfo2&)> Predicate;
    uint32_t MaxConnections;
 
    ServiceSubscriptionFilter();
};

class ROBOTRACONTEUR_CORE_API ServiceSubscriptionClientID
{
  public:
    ::RobotRaconteur::NodeID NodeID;
    std::string ServiceName;

    ServiceSubscriptionClientID(const ::RobotRaconteur::NodeID& nodeid, boost::string_ref service_name);

    ServiceSubscriptionClientID();

    bool operator==(const ServiceSubscriptionClientID& id2) const;

    bool operator!=(const ServiceSubscriptionClientID& id2) const;

    bool operator<(const ServiceSubscriptionClientID& id2) const;
};
 
size_t hash_value(const ServiceSubscriptionClientID& id);
 
class IServiceSubscription
{
  public:
    friend class detail::Discovery;
 
  protected:
    virtual void Init(const std::vector<std::string>& service_types,
                      const RR_SHARED_PTR<ServiceSubscriptionFilter>& filter) = 0;
    virtual void NodeUpdated(RR_SHARED_PTR<detail::Discovery_nodestorage> storage) = 0;
    virtual void NodeLost(RR_SHARED_PTR<detail::Discovery_nodestorage> storage) = 0;
 
  public:
    virtual void Close() = 0;
    virtual ~IServiceSubscription(){};
};

class ROBOTRACONTEUR_CORE_API ServiceInfo2Subscription : public IServiceSubscription,
                                                         public RR_ENABLE_SHARED_FROM_THIS<ServiceInfo2Subscription>,
                                                         private boost::noncopyable
{
  public:
    friend class detail::Discovery;
    friend class ServiceSubscriptionManager;
 
    typedef boost::signals2::connection event_connection;

    std::map<ServiceSubscriptionClientID, ServiceInfo2> GetDetectedServiceInfo2();

    event_connection AddServiceDetectedListener(
        boost::function<void(const RR_SHARED_PTR<ServiceInfo2Subscription>&, const ServiceSubscriptionClientID&,
                             const ServiceInfo2&)>
            handler);

    event_connection AddServiceLostListener(
        boost::function<void(const RR_SHARED_PTR<ServiceInfo2Subscription>&, const ServiceSubscriptionClientID&,
                             const ServiceInfo2&)>
            handler);

    RR_OVIRTUAL void Close() RR_OVERRIDE;
 
    RR_SHARED_PTR<RobotRaconteurNode> GetNode();
 
  protected:
    boost::mutex this_lock;
 
    bool active;
 
    std::map<ServiceSubscriptionClientID, RR_SHARED_PTR<detail::ServiceInfo2Subscription_client> > clients;
 
    RR_WEAK_PTR<RobotRaconteurNode> node;
    RR_WEAK_PTR<detail::Discovery> parent;
 
    std::vector<std::string> service_types;
    RR_SHARED_PTR<ServiceSubscriptionFilter> filter;
 
    uint32_t retry_delay;
 
    boost::signals2::signal<void(const RR_SHARED_PTR<ServiceInfo2Subscription>&, const ServiceSubscriptionClientID&,
                                 const ServiceInfo2&)>
        detected_listeners;
    boost::signals2::signal<void(const RR_SHARED_PTR<ServiceInfo2Subscription>&, const ServiceSubscriptionClientID&,
                                 const ServiceInfo2&)>
        lost_listeners;
 
  public:
    // Do not call, use RobotRaconteurNode()->SubscribeServiceInfo2()
    ServiceInfo2Subscription(const RR_SHARED_PTR<detail::Discovery>& parent);
 
  protected:
    RR_OVIRTUAL void Init(const std::vector<std::string>& service_types,
                          const RR_SHARED_PTR<ServiceSubscriptionFilter>& filter) RR_OVERRIDE;
    RR_OVIRTUAL void NodeUpdated(RR_SHARED_PTR<detail::Discovery_nodestorage> storage) RR_OVERRIDE;
    RR_OVIRTUAL void NodeLost(RR_SHARED_PTR<detail::Discovery_nodestorage> storage) RR_OVERRIDE;
 
    void fire_ServiceDetectedListener(const ServiceSubscriptionClientID& noden, const ServiceInfo2& info);
    void fire_ServiceLostListener(const ServiceSubscriptionClientID& noden, const ServiceInfo2& info);
};
 
namespace detail
{
template <typename T>
void AsyncGetDefaultClient_handler_adapter(
    const boost::function<void(const RR_SHARED_PTR<T>&, const RR_SHARED_PTR<RobotRaconteurException>&)>& handler,
    const RR_SHARED_PTR<RRObject>& obj, const RR_SHARED_PTR<RobotRaconteurException>& err)
{
    if (err)
    {
        handler(RR_SHARED_PTR<T>(), err);
        return;
    }
 
    RR_SHARED_PTR<T> obj1 = RR_DYNAMIC_POINTER_CAST<T>(obj);
    if (!obj1)
    {
        handler(RR_SHARED_PTR<T>(), RR_MAKE_SHARED<ConnectionException>("Cannot cast client to requested type"));
        return;
    }
 
    handler(obj1, RR_SHARED_PTR<RobotRaconteurException>());
}
} // namespace detail

class ROBOTRACONTEUR_CORE_API ServiceSubscription : public IServiceSubscription,
                                                    public RR_ENABLE_SHARED_FROM_THIS<ServiceSubscription>,
                                                    private boost::noncopyable
{
  public:
    friend class detail::Discovery;
    friend class detail::ServiceSubscription_retrytimer;
    friend class WireSubscriptionBase;
    friend class PipeSubscriptionBase;
    friend class detail::ServiceSubscription_custom_member_subscribers;
    friend class SubObjectSubscription;
    friend class ServiceSubscriptionManager;
 
    typedef boost::signals2::connection event_connection;

    std::map<ServiceSubscriptionClientID, RR_SHARED_PTR<RRObject> > GetConnectedClients();

    event_connection AddClientConnectListener(
        boost::function<void(const RR_SHARED_PTR<ServiceSubscription>&, const ServiceSubscriptionClientID&,
                             const RR_SHARED_PTR<RRObject>&)>
            handler);

    event_connection AddClientDisconnectListener(
        boost::function<void(const RR_SHARED_PTR<ServiceSubscription>&, const ServiceSubscriptionClientID&,
                             const RR_SHARED_PTR<RRObject>&)>
            handler);

    event_connection AddClientConnectFailedListener(
        boost::function<void(const RR_SHARED_PTR<ServiceSubscription>&, const ServiceSubscriptionClientID&,
                             const std::vector<std::string>&, const RR_SHARED_PTR<RobotRaconteurException>&)>
            handler);

    RR_OVIRTUAL void Close() RR_OVERRIDE;

    virtual void ClaimClient(const RR_SHARED_PTR<RRObject>& client);

    virtual void ReleaseClient(const RR_SHARED_PTR<RRObject>& client);

    uint32_t GetConnectRetryDelay();

    void SetConnectRetryDelay(uint32_t delay_milliseconds);

    template <typename T>
    RR_SHARED_PTR<WireSubscription<T> > SubscribeWire(boost::string_ref membername, boost::string_ref servicepath = "")
    {
        RR_SHARED_PTR<WireSubscription<T> > o =
            RR_MAKE_SHARED<WireSubscription<T> >(shared_from_this(), membername, servicepath);
        SubscribeWire1(o);
        return o;
    }

    template <typename T>
    RR_SHARED_PTR<PipeSubscription<T> > SubscribePipe(boost::string_ref membername, boost::string_ref servicepath = "",
                                                      uint32_t max_recv_packets = std::numeric_limits<uint32_t>::max())
    {
        RR_SHARED_PTR<PipeSubscription<T> > o =
            RR_MAKE_SHARED<PipeSubscription<T> >(shared_from_this(), membername, servicepath, max_recv_packets);
        SubscribePipe1(o);
        return o;
    }

    template <typename T>
    RR_SHARED_PTR<T> GetDefaultClient()
    {
        return rr_cast<T>(GetDefaultClientBase());
    }

    template <typename T>
    bool TryGetDefaultClient(RR_SHARED_PTR<T>& client_out)
    {
        RR_SHARED_PTR<RRObject> c;
        if (!TryGetDefaultClientBase(c))
        {
            return false;
        }
        RR_SHARED_PTR<T> c1 = RR_DYNAMIC_POINTER_CAST<T>(c);
        if (!c1)
            return false;
 
        client_out = c1;
        return true;
    }

    template <typename T>
    RR_SHARED_PTR<T> GetDefaultClientWait(int32_t timeout = RR_TIMEOUT_INFINITE)
    {
        return rr_cast<T>(GetDefaultClientWaitBase(timeout));
    }

    template <typename T>
    bool TryGetDefaultClientWait(RR_SHARED_PTR<T>& client_out, int32_t timeout = RR_TIMEOUT_INFINITE)
    {
        RR_SHARED_PTR<RRObject> c;
        if (!TryGetDefaultClientWaitBase(c, timeout))
        {
            return false;
        }
        RR_SHARED_PTR<T> c1 = RR_DYNAMIC_POINTER_CAST<T>(c);
        if (!c1)
            return false;
 
        client_out = c1;
        return true;
    }

    template <typename T>
    void AsyncGetDefaultClient(
        boost::function<void(const RR_SHARED_PTR<T>&, const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
        int32_t timeout = RR_TIMEOUT_INFINITE)
    {
        AsyncGetDefaultClientBase(boost::bind(&detail::AsyncGetDefaultClient_handler_adapter<T>, handler,
                                              RR_BOOST_PLACEHOLDERS(_1), RR_BOOST_PLACEHOLDERS(_2)),
                                  timeout);
    }

    std::vector<std::string> GetServiceURL();

    void UpdateServiceURL(boost::string_ref url, boost::string_ref username = "",
                          const RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> >& credentials =
                              (RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> >()),
                          boost::string_ref object_type = "", bool close_connected = false);

    void UpdateServiceURL(const std::vector<std::string>& url, boost::string_ref username = "",
                          const RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> >& credentials =
                              (RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> >()),
                          boost::string_ref object_type = "", bool close_connected = false);

 
    void UpdateServiceByType(
        const std::vector<std::string>& service_types,
        const RR_SHARED_PTR<ServiceSubscriptionFilter>& filter = RR_SHARED_PTR<ServiceSubscriptionFilter>());

    RR_SHARED_PTR<SubObjectSubscription> SubscribeSubObject(boost::string_ref servicepath,
                                                            boost::string_ref objecttype = "");
 
    RR_SHARED_PTR<RobotRaconteurNode> GetNode();
 
  protected:
    boost::mutex this_lock;
 
    bool active;
 
    std::map<ServiceSubscriptionClientID, RR_SHARED_PTR<detail::ServiceSubscription_client> > clients;
 
    RR_WEAK_PTR<RobotRaconteurNode> node;
    RR_WEAK_PTR<detail::Discovery> parent;
 
    std::vector<std::string> service_types;
    RR_SHARED_PTR<ServiceSubscriptionFilter> filter;
 
    uint32_t retry_delay;
 
    boost::signals2::signal<void(const RR_SHARED_PTR<ServiceSubscription>&, const ServiceSubscriptionClientID&,
                                 const RR_SHARED_PTR<RRObject>&)>
        connect_listeners;
    boost::signals2::signal<void(const RR_SHARED_PTR<ServiceSubscription>&, const ServiceSubscriptionClientID&,
                                 const RR_SHARED_PTR<RRObject>&)>
        disconnect_listeners;
    boost::signals2::signal<void(const RR_SHARED_PTR<ServiceSubscription>&, const ServiceSubscriptionClientID&,
                                 const std::vector<std::string>&, const RR_SHARED_PTR<RobotRaconteurException>&)>
        connect_failed_listeners;
 
    RR_SHARED_PTR<RR_BOOST_ASIO_STRAND> listener_strand;
 
    boost::unordered_set<RR_SHARED_PTR<WireSubscriptionBase> > wire_subscriptions;
    boost::unordered_set<RR_SHARED_PTR<PipeSubscriptionBase> > pipe_subscriptions;
 
    bool use_service_url;
    std::vector<std::string> service_url;
    std::string service_url_username;
    RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> > service_url_credentials;
 
  public:
    // Do not call, use RobotRaconteurNode()->SubscribeService()
    ServiceSubscription(const RR_SHARED_PTR<detail::Discovery>& parent);
 
  protected:
    RR_OVIRTUAL void Init(const std::vector<std::string>& service_types,
                          const RR_SHARED_PTR<ServiceSubscriptionFilter>& filter) RR_OVERRIDE;
    virtual void InitServiceURL(const std::vector<std::string>& url, boost::string_ref username = "",
                                const RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> >& credentials =
                                    (RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> >()),
                                boost::string_ref objecttype = "");
    RR_OVIRTUAL void NodeUpdated(RR_SHARED_PTR<detail::Discovery_nodestorage> storage) RR_OVERRIDE;
    RR_OVIRTUAL void NodeLost(RR_SHARED_PTR<detail::Discovery_nodestorage> storage) RR_OVERRIDE;
 
    void ClientConnected(const RR_SHARED_PTR<RRObject>& c, const RR_SHARED_PTR<RobotRaconteurException>& err,
                         const RR_SHARED_PTR<detail::ServiceSubscription_client>& c2,
                         const std::vector<std::string>& url);
    void ConnectRetry(const RR_SHARED_PTR<detail::ServiceSubscription_client>& c2);
    void ConnectRetry2(const RR_SHARED_PTR<detail::ServiceSubscription_client>& c2);
 
    static void ClientEvent(RR_WEAK_PTR<ServiceSubscription> this_, const RR_SHARED_PTR<ClientContext>& ctx,
                            ClientServiceListenerEventType evt, const RR_SHARED_PTR<void>& p,
                            RR_WEAK_PTR<detail::ServiceSubscription_client> c2);
 
    void fire_ClientConnectListeners(const ServiceSubscriptionClientID& noden, const RR_SHARED_PTR<RRObject>& client);
    void fire_ClientDisconnectListeners(const ServiceSubscriptionClientID& noden,
                                        const RR_SHARED_PTR<RRObject>& client);
    void fire_ClientConnectFailedListeners(const ServiceSubscriptionClientID& noden,
                                           const std::vector<std::string>& url,
                                           const RR_SHARED_PTR<RobotRaconteurException>& err);
 
    void SubscribeWire1(const RR_SHARED_PTR<WireSubscriptionBase>& s);
    void SubscribePipe1(const RR_SHARED_PTR<PipeSubscriptionBase>& s);
 
    void WireSubscriptionClosed(const RR_SHARED_PTR<WireSubscriptionBase>& s);
    void PipeSubscriptionClosed(const RR_SHARED_PTR<PipeSubscriptionBase>& s);
 
    RR_SHARED_PTR<RRObject> GetDefaultClientBase();
    bool TryGetDefaultClientBase(RR_SHARED_PTR<RRObject>& client_out);
 
    RR_SHARED_PTR<RRObject> GetDefaultClientWaitBase(int32_t timeout = RR_TIMEOUT_INFINITE);
    bool TryGetDefaultClientWaitBase(RR_SHARED_PTR<RRObject>& client_out, int32_t timeout = RR_TIMEOUT_INFINITE);
 
    void AsyncGetDefaultClientBase(
        boost::function<void(const RR_SHARED_PTR<RRObject>&, const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
        int32_t timeout = RR_TIMEOUT_INFINITE);
 
    void SoftClose();
};

class ROBOTRACONTEUR_CORE_API WireSubscriptionBase : public RR_ENABLE_SHARED_FROM_THIS<WireSubscriptionBase>,
                                                     private boost::noncopyable
{
  public:
    friend class WireConnectionBase;
    friend class ServiceSubscription;
    friend class detail::WireSubscription_connection;
    friend class detail::WireSubscription_send_iterator;
 
    typedef boost::signals2::connection event_connection;
 
    virtual ~WireSubscriptionBase() {}
 
    RR_INTRUSIVE_PTR<RRValue> GetInValueBase(TimeSpec* time = NULL,
                                             RR_SHARED_PTR<WireConnectionBase>* connection = NULL);
    bool TryGetInValueBase(RR_INTRUSIVE_PTR<RRValue>& val, TimeSpec* time = NULL,
                           RR_SHARED_PTR<WireConnectionBase>* connection = NULL);

    bool WaitInValueValid(int32_t timeout = RR_TIMEOUT_INFINITE);

    size_t GetActiveWireConnectionCount();

    bool GetIgnoreInValue();

    void SetIgnoreInValue(bool ignore);

    int32_t GetInValueLifespan();

    void SetInValueLifespan(int32_t millis);
 
    void SetOutValueAllBase(const RR_INTRUSIVE_PTR<RRValue>& val);

    void Close();
 
    WireSubscriptionBase(const RR_SHARED_PTR<ServiceSubscription>& parent, boost::string_ref membername,
                         boost::string_ref servicepath);
 
    RR_SHARED_PTR<RobotRaconteurNode> GetNode();
 
  protected:
    void ClientConnected(const ServiceSubscriptionClientID& client_id, const RR_SHARED_PTR<RRObject>& client);
    void ClientDisconnected(const ServiceSubscriptionClientID& client_id, const RR_SHARED_PTR<RRObject>& client);
 
    void WireConnectionClosed(const RR_SHARED_PTR<detail::WireSubscription_connection>& wire);
    void WireValueChanged(const RR_SHARED_PTR<detail::WireSubscription_connection>& wire,
                          const RR_INTRUSIVE_PTR<RRValue>& value, const TimeSpec& time);
 
    boost::mutex this_lock;
    boost::unordered_map<ServiceSubscriptionClientID, RR_SHARED_PTR<detail::WireSubscription_connection> > connections;
    boost::initialized<bool> closed;
    RR_WEAK_PTR<RobotRaconteurNode> node;
    RR_WEAK_PTR<ServiceSubscription> parent;
 
    RR_INTRUSIVE_PTR<RRValue> in_value;
    TimeSpec in_value_time;
    boost::initialized<bool> in_value_valid;
    RR_SHARED_PTR<WireConnectionBase> in_value_connection;
    int32_t in_value_lifespan;
    boost::posix_time::ptime in_value_time_local;
 
    boost::condition_variable in_value_wait;
 
    boost::initialized<bool> ignore_in_value;
 
    std::string membername;
    std::string servicepath;
 
    virtual void fire_WireValueChanged(const RR_INTRUSIVE_PTR<RRValue>& value, const TimeSpec& time,
                                       const RR_SHARED_PTR<WireConnectionBase>& connection);
    virtual bool isempty_WireValueChanged();
    RR_SHARED_PTR<detail::async_signal_pool_semaphore> wire_value_changed_semaphore;
};

template <typename T>
class WireSubscription : public WireSubscriptionBase
{
  public:
    WireSubscription(const RR_SHARED_PTR<ServiceSubscription>& parent, boost::string_ref membername,
                     boost::string_ref servicepath)
        : WireSubscriptionBase(parent, membername, servicepath)
    {}

    T GetInValue(TimeSpec* time = NULL, typename RR_SHARED_PTR<WireConnection<T> >* connection = NULL)
    {
        RR_SHARED_PTR<WireConnectionBase> connection1;
        T o = RRPrimUtil<T>::PreUnpack(GetInValueBase(time, &connection1));
        if (connection && connection1)
        {
            *connection = RR_DYNAMIC_POINTER_CAST<WireConnection<T> >(connection1);
        }
        return o;
    }

    bool TryGetInValue(T& val, TimeSpec* time = NULL, typename RR_SHARED_PTR<WireConnection<T> >* connection = NULL)
    {
        RR_INTRUSIVE_PTR<RRValue> o;
        RR_SHARED_PTR<WireConnectionBase> connection1;
        if (!TryGetInValueBase(o, time, &connection1))
            return false;
        val = RRPrimUtil<T>::PreUnpack(o);
        if (connection && connection1)
        {
            *connection = RR_DYNAMIC_POINTER_CAST<WireConnection<T> >(connection1);
        }
        return true;
    }

    void SetOutValueAll(const T& val) { SetOutValueAllBase(RRPrimUtil<T>::PrePack(val)); }

    event_connection AddWireValueChangedListener(
        boost::function<void(const RR_SHARED_PTR<WireSubscription<T> >&, const T&, const TimeSpec&)> f)
    {
        return wire_value_changed.connect(f);
    }
 
  protected:
    boost::signals2::signal<void(const RR_SHARED_PTR<WireSubscription<T> >&, const T&, const TimeSpec&)>
        wire_value_changed;
 
    RR_OVIRTUAL void fire_WireValueChanged(const RR_INTRUSIVE_PTR<RRValue>& value, const TimeSpec& time,
                                           const RR_SHARED_PTR<WireConnectionBase>& connection) RR_OVERRIDE
    {
        RR_UNUSED(connection);
        wire_value_changed(RR_STATIC_POINTER_CAST<WireSubscription<T> >(shared_from_this()),
                           RRPrimUtil<T>::PreUnpack(value), time);
    }
 
    RR_OVIRTUAL bool isempty_WireValueChanged() RR_OVERRIDE { return wire_value_changed.empty(); }
};

class ROBOTRACONTEUR_CORE_API PipeSubscriptionBase : public RR_ENABLE_SHARED_FROM_THIS<PipeSubscriptionBase>,
                                                     private boost::noncopyable
{
  public:
    friend class PipeConnectionBase;
    friend class ServiceSubscription;
    friend class detail::PipeSubscription_connection;
    friend class detail::PipeSubscription_send_iterator;
 
    typedef boost::signals2::connection event_connection;
 
    virtual ~PipeSubscriptionBase() {}
 
    RR_INTRUSIVE_PTR<RRValue> ReceivePacketBase();
    bool TryReceivePacketBase(RR_INTRUSIVE_PTR<RRValue>& packet);
    bool TryReceivePacketBaseWait(RR_INTRUSIVE_PTR<RRValue>& packet, int32_t timeout = RR_TIMEOUT_INFINITE,
                                  bool peek = false, RR_SHARED_PTR<PipeEndpointBase>* ep = NULL);

    size_t Available();

    size_t GetActivePipeEndpointCount();

    bool GetIgnoreReceived();

    void SetIgnoreReceived(bool ignore);
 
    void AsyncSendPacketAllBase(const RR_INTRUSIVE_PTR<RRValue>& packet);

    void Close();
 
    PipeSubscriptionBase(const RR_SHARED_PTR<ServiceSubscription>& parent, boost::string_ref membername,
                         boost::string_ref servicepath = "", int32_t max_recv_packets = -1,
                         int32_t max_send_backlog = 5);
 
    RR_SHARED_PTR<RobotRaconteurNode> GetNode();
 
  protected:
    void ClientConnected(const ServiceSubscriptionClientID& client_id, const RR_SHARED_PTR<RRObject>& client);
    void ClientDisconnected(const ServiceSubscriptionClientID& client_id, const RR_SHARED_PTR<RRObject>& client);
 
    void PipeEndpointClosed(const RR_SHARED_PTR<detail::PipeSubscription_connection>& pipe);
    void PipeEndpointPacketReceived(const RR_SHARED_PTR<detail::PipeSubscription_connection>& pipe,
                                    const RR_INTRUSIVE_PTR<RRValue>& value);
 
    boost::mutex this_lock;
    boost::unordered_map<ServiceSubscriptionClientID, RR_SHARED_PTR<detail::PipeSubscription_connection> > connections;
    boost::initialized<bool> closed;
    RR_WEAK_PTR<ServiceSubscription> parent;
    RR_WEAK_PTR<RobotRaconteurNode> node;
 
    std::deque<boost::tuple<RR_INTRUSIVE_PTR<RRValue>, RR_SHARED_PTR<PipeEndpointBase> > > recv_packets;
    boost::condition_variable recv_packets_wait;
 
    std::string membername;
    std::string servicepath;
 
    boost::initialized<int32_t> max_recv_packets;
    boost::initialized<bool> ignore_incoming_packets;
 
    boost::initialized<int32_t> max_send_backlog;
 
    virtual void fire_PipePacketReceived();
    virtual bool isempty_PipePacketReceived();
    RR_SHARED_PTR<detail::async_signal_pool_semaphore> pipe_packet_received_semaphore;
};

template <typename T>
class PipeSubscription : public PipeSubscriptionBase
{
  public:
    PipeSubscription(const RR_SHARED_PTR<ServiceSubscription>& parent, boost::string_ref membername,
                     boost::string_ref servicepath = "", int32_t max_recv_packets = -1, int32_t max_send_backlog = 5)
        : PipeSubscriptionBase(parent, membername, servicepath, max_recv_packets)
    {
        RR_UNUSED(max_send_backlog);
    }

    T ReceivePacket() { return RRPrimUtil<T>::PreUnpack(ReceivePacketBase()); }

 
    bool TryReceivePacket(T& packet)
    {
        RR_INTRUSIVE_PTR<RRValue> o;
        if (!TryReceivePacketBase(o))
            return false;
        packet = RRPrimUtil<T>::PreUnpack(o);
        return true;
    }

    bool TryReceivePacketWait(T& packet, int32_t timeout = RR_TIMEOUT_INFINITE, bool peek = false,
                              RR_SHARED_PTR<PipeEndpoint<T> >* ep = NULL)
    {
        RR_INTRUSIVE_PTR<RRValue> o;
        RR_SHARED_PTR<PipeEndpointBase> ep1;
        if (!TryReceivePacketBaseWait(o, timeout, peek, &ep1))
            return false;
        packet = RRPrimUtil<T>::PreUnpack(o);
        if (ep1)
        {
            *ep = RR_DYNAMIC_POINTER_CAST<PipeEndpoint<T> >(ep1);
        }
        return true;
    }

    void AsyncSendPacketAll(const T& packet) { AsyncSendPacketAllBase(RRPrimUtil<T>::PrePack(packet)); }

    event_connection AddPipePacketReceivedListener(boost::function<void(const RR_SHARED_PTR<PipeSubscription<T> >&)> f)
    {
        return pipe_packet_received.connect(f);
    }
 
  protected:
    boost::signals2::signal<void(const RR_SHARED_PTR<PipeSubscription<T> >&)> pipe_packet_received;
 
    RR_OVIRTUAL void fire_PipePacketReceived() RR_OVERRIDE
    {
        pipe_packet_received(RR_STATIC_POINTER_CAST<PipeSubscription<T> >(shared_from_this()));
    }
 
    RR_OVIRTUAL bool isempty_PipePacketReceived() RR_OVERRIDE { return pipe_packet_received.empty(); }
};

class ROBOTRACONTEUR_CORE_API SubObjectSubscription : public RR_ENABLE_SHARED_FROM_THIS<SubObjectSubscription>,
                                                      private boost::noncopyable
{
 
  public:
    friend class ServiceSubscription;
 
    virtual ~SubObjectSubscription() {}

    template <typename T>
    RR_SHARED_PTR<T> GetDefaultClient()
    {
        return rr_cast<T>(GetDefaultClientBase());
    }

    template <typename T>
    bool TryGetDefaultClient(RR_SHARED_PTR<T>& client_out)
    {
        RR_SHARED_PTR<RRObject> c;
        if (!TryGetDefaultClientBase(c))
        {
            return false;
        }
        RR_SHARED_PTR<T> c1 = RR_DYNAMIC_POINTER_CAST<T>(c);
        if (!c1)
            return false;
 
        client_out = c1;
        return true;
    }

    template <typename T>
    RR_SHARED_PTR<T> GetDefaultClientWait(int32_t timeout = RR_TIMEOUT_INFINITE)
    {
        return rr_cast<T>(GetDefaultClientWaitBase(timeout));
    }

    template <typename T>
    bool TryGetDefaultClientWait(RR_SHARED_PTR<T>& client_out, int32_t timeout = RR_TIMEOUT_INFINITE)
    {
        RR_SHARED_PTR<RRObject> c;
        if (!TryGetDefaultClientWaitBase(c, timeout))
        {
            return false;
        }
        RR_SHARED_PTR<T> c1 = RR_DYNAMIC_POINTER_CAST<T>(c);
        if (!c1)
            return false;
 
        client_out = c1;
        return true;
    }

    template <typename T>
    void AsyncGetDefaultClient(
        boost::function<void(const RR_SHARED_PTR<T>&, const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
        int32_t timeout = RR_TIMEOUT_INFINITE)
    {
        AsyncGetDefaultClientBase(boost::bind(&detail::AsyncGetDefaultClient_handler_adapter<T>, handler,
                                              RR_BOOST_PLACEHOLDERS(_1), RR_BOOST_PLACEHOLDERS(_2)),
                                  timeout);
    }
 
    void Init();

    void Close();
 
    SubObjectSubscription(const RR_SHARED_PTR<ServiceSubscription>& parent, boost::string_ref servicepath,
                          boost::string_ref objecttype);
 
    RR_SHARED_PTR<RobotRaconteurNode> GetNode();
 
  protected:
    RR_SHARED_PTR<RRObject> GetDefaultClientBase();
    bool TryGetDefaultClientBase(RR_SHARED_PTR<RRObject>& client_out);
 
    RR_SHARED_PTR<RRObject> GetDefaultClientWaitBase(int32_t timeout = RR_TIMEOUT_INFINITE);
    bool TryGetDefaultClientWaitBase(RR_SHARED_PTR<RRObject>& client_out, int32_t timeout = RR_TIMEOUT_INFINITE);
 
    void AsyncGetDefaultClientBase(
        boost::function<void(const RR_SHARED_PTR<RRObject>&, const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
        int32_t timeout = RR_TIMEOUT_INFINITE);
 
    RR_WEAK_PTR<ServiceSubscription> parent;
    RR_WEAK_PTR<RobotRaconteurNode> node;
    std::string servicepath;
    std::string objecttype;
};

enum ServiceSubscriptionManager_CONNECTION_METHOD
{
    ServiceSubscriptionManager_CONNECTION_METHOD_DEFAULT,
    ServiceSubscriptionManager_CONNECTION_METHOD_URL,
    ServiceSubscriptionManager_CONNECTION_METHOD_TYPE
};

struct ROBOTRACONTEUR_CORE_API ServiceSubscriptionManagerDetails
{
    std::string Name;
    ServiceSubscriptionManager_CONNECTION_METHOD ConnectionMethod;
    std::vector<std::string> Urls;
    std::string UrlUsername;
    RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> > UrlCredentials;
    std::vector<std::string> ServiceTypes;
    RR_SHARED_PTR<ServiceSubscriptionFilter> Filter;
    bool Enabled;

    ServiceSubscriptionManagerDetails();

    ServiceSubscriptionManagerDetails(
        const boost::string_ref& Name,
        ServiceSubscriptionManager_CONNECTION_METHOD ConnectionMethod =
            ServiceSubscriptionManager_CONNECTION_METHOD_DEFAULT,
        const std::vector<std::string>& Urls = std::vector<std::string>(), const boost::string_ref& UrlUsername = "",
        const RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> >& UrlCredentials =
            RR_INTRUSIVE_PTR<RRMap<std::string, RRValue> >(),
        const std::vector<std::string>& ServiceTypes = std::vector<std::string>(),
        const RR_SHARED_PTR<ServiceSubscriptionFilter>& Filter = RR_SHARED_PTR<ServiceSubscriptionFilter>(),
        bool Enabled = true);
};

class ROBOTRACONTEUR_CORE_API ServiceSubscriptionManager
{
  public:
    ServiceSubscriptionManager(const RR_SHARED_PTR<RobotRaconteurNode>& node = RobotRaconteurNode::sp());
 
    virtual ~ServiceSubscriptionManager();

    void Init(const std::vector<ServiceSubscriptionManagerDetails>& details);

    void AddSubscription(const ServiceSubscriptionManagerDetails& details);

    void RemoveSubscription(const boost::string_ref& name, bool close = true);

    void EnableSubscription(const boost::string_ref& name);

    void DisableSubscription(const boost::string_ref& name, bool close = true);

    RR_SHARED_PTR<ServiceSubscription> GetSubscription(const boost::string_ref& name, bool force_create = true);

    bool IsConnected(const boost::string_ref& name);

    bool IsEnabled(const boost::string_ref& name);

    void Close(bool close_subscriptions = true);

    std::vector<std::string> GetSubscriptionNames();

    std::vector<ServiceSubscriptionManagerDetails> GetSubscriptionDetails();

    RR_SHARED_PTR<RobotRaconteurNode> GetNode();
 
  protected:
    RR_WEAK_PTR<RobotRaconteurNode> node;
 
    boost::mutex this_lock;
 
    boost::unordered_map<std::string, detail::ServiceSubscriptionManager_subscription> subscriptions;
 
    RR_SHARED_PTR<ServiceSubscription> CreateSubscription(const ServiceSubscriptionManagerDetails& details);
 
    void UpdateSubscription(detail::ServiceSubscriptionManager_subscription& sub,
                            const ServiceSubscriptionManagerDetails& details, bool close = false);
};
 
namespace detail
{
class ROBOTRACONTEUR_CORE_API WireSubscription_send_iterator
{
  protected:
    RR_SHARED_PTR<WireSubscriptionBase> subscription;
    boost::mutex::scoped_lock subscription_lock;
    boost::unordered_map<ServiceSubscriptionClientID, RR_SHARED_PTR<WireSubscription_connection> >::iterator
        connections_iterator;
    boost::unordered_map<ServiceSubscriptionClientID, RR_SHARED_PTR<WireSubscription_connection> >::iterator
        current_connection;
 
  public:
    WireSubscription_send_iterator(const RR_SHARED_PTR<WireSubscriptionBase>& subscription);
    RR_SHARED_PTR<WireConnectionBase> Next();
    void SetOutValue(const RR_INTRUSIVE_PTR<RRValue>& value);
    virtual ~WireSubscription_send_iterator();
};
 
class ROBOTRACONTEUR_CORE_API PipeSubscription_send_iterator
{
  protected:
    RR_SHARED_PTR<PipeSubscriptionBase> subscription;
    boost::mutex::scoped_lock subscription_lock;
    boost::unordered_map<ServiceSubscriptionClientID, RR_SHARED_PTR<PipeSubscription_connection> >::iterator
        connections_iterator;
    boost::unordered_map<ServiceSubscriptionClientID, RR_SHARED_PTR<PipeSubscription_connection> >::iterator
        current_connection;
 
  public:
    PipeSubscription_send_iterator(const RR_SHARED_PTR<PipeSubscriptionBase>& subscription);
    RR_SHARED_PTR<PipeEndpointBase> Next();
    void AsyncSendPacket(const RR_INTRUSIVE_PTR<RRValue>& packet);
    virtual ~PipeSubscription_send_iterator();
};
 
class ROBOTRACONTEUR_CORE_API ServiceSubscription_custom_member_subscribers
{
  public:
    static void SubscribeWire(const RR_SHARED_PTR<ServiceSubscription>& s,
                              const RR_SHARED_PTR<WireSubscriptionBase>& o);
    static void SubscribePipe(const RR_SHARED_PTR<ServiceSubscription>& s,
                              const RR_SHARED_PTR<PipeSubscriptionBase>& o);
};
} // namespace detail
 
#ifndef ROBOTRACONTEUR_NO_CXX11_TEMPLATE_ALIASES
using ServiceSubscriptionFilterNodePtr = RR_SHARED_PTR<ServiceSubscriptionFilterNode>;
using ServiceSubscriptionFilterPtr = RR_SHARED_PTR<ServiceSubscriptionFilter>;
using ServiceInfo2SubscriptionPtr = RR_SHARED_PTR<ServiceInfo2Subscription>;
using ServiceSubscriptionPtr = RR_SHARED_PTR<ServiceSubscription>;
template <typename T>
using PipeSubscriptionPtr = RR_SHARED_PTR<PipeSubscription<T> >;
template <typename T>
using WireSubscriptionPtr = RR_SHARED_PTR<WireSubscription<T> >;
using SubObjectSubscriptionPtr = RR_SHARED_PTR<SubObjectSubscription>;
#endif
 
} // namespace RobotRaconteur