PipeMember.h

Go to the documentation of this file.

 
#pragma once
 
#include "RobotRaconteur/Message.h"
#include "RobotRaconteur/RobotRaconteurNode.h"
#include "RobotRaconteur/Endpoint.h"
#include "RobotRaconteur/DataTypes.h"
#include <boost/function.hpp>
#include <boost/unordered_map.hpp>
#include <boost/call_traits.hpp>
#include <list>
 
#ifdef _MSVC_VER
#pragma warning(push)
#pragma warning(disable : 4250)
#pragma warning(disable : 4996)
#endif
 
#include <boost/signals2.hpp>
 
namespace RobotRaconteur
{
 
class ROBOTRACONTEUR_CORE_API PipeBase;
class ROBOTRACONTEUR_CORE_API PipeEndpointBaseListener;
namespace detail
{
class PipeSubscription_connection;
}

class ROBOTRACONTEUR_CORE_API PipeEndpointBase : public RR_ENABLE_SHARED_FROM_THIS<PipeEndpointBase>,
                                                 private boost::noncopyable
{
    friend class PipeBase;
    friend class PipeClientBase;
    friend class PipeServerBase;
    friend class PipeBroadcasterBase;
    friend class PipeSubscriptionBase;
    friend class detail::PipeSubscription_connection;
 
  public:
    virtual ~PipeEndpointBase() {}

    virtual int32_t GetIndex();

    virtual uint32_t GetEndpoint();

    bool GetRequestPacketAck();

    void SetRequestPacketAck(bool ack);

    virtual void Close();

    virtual void AsyncClose(boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
                            int32_t timeout = RR_TIMEOUT_INFINITE);

    virtual size_t Available();

    bool IsUnreliable();

    MemberDefinition_Direction Direction();

    bool GetIgnoreReceived();
    void SetIgnoreReceived(bool ignore);
 
    virtual void AddListener(const RR_SHARED_PTR<PipeEndpointBaseListener>& listener);
 
    RR_SHARED_PTR<RobotRaconteurNode> GetNode();
 
  protected:
    virtual void RemoteClose();
 
    PipeEndpointBase(const RR_SHARED_PTR<PipeBase>& parent, int32_t index, uint32_t endpoint = 0,
                     bool unreliable = false, MemberDefinition_Direction direction = MemberDefinition_Direction_both);
 
    bool unreliable;
    MemberDefinition_Direction direction;
 
    bool RequestPacketAck;
 
    void AsyncSendPacketBase(const RR_INTRUSIVE_PTR<RRValue>& packet,
                             RR_MOVE_ARG(boost::function<void(uint32_t, const RR_SHARED_PTR<RobotRaconteurException>&)>)
                                 handler);
 
    RR_INTRUSIVE_PTR<RRValue> ReceivePacketBase();
    RR_INTRUSIVE_PTR<RRValue> PeekPacketBase();
 
    RR_INTRUSIVE_PTR<RRValue> ReceivePacketBaseWait(int32_t timeout = RR_TIMEOUT_INFINITE);
    RR_INTRUSIVE_PTR<RRValue> PeekPacketBaseWait(int32_t timeout = RR_TIMEOUT_INFINITE);
 
    bool TryReceivePacketBaseWait(RR_INTRUSIVE_PTR<RRValue>& packet, int32_t timeout = RR_TIMEOUT_INFINITE,
                                  bool peek = false);
 
    boost::mutex sendlock;
    boost::mutex recvlock;
 
    std::deque<RR_INTRUSIVE_PTR<RRValue> > recv_packets;
    boost::condition_variable recv_packets_wait;
 
    uint32_t increment_packet_number(uint32_t packetnum);
 
    void PipePacketReceived(const RR_INTRUSIVE_PTR<RRValue>& packet, uint32_t packetnum);
 
    void PipePacketAckReceived(uint32_t packetnum);
 
    void Shutdown();
 
    virtual void fire_PipeEndpointClosedCallback() = 0;
 
    virtual void fire_PacketReceivedEvent() = 0;
 
    virtual void fire_PacketAckReceivedEvent(uint32_t packetnum) = 0;
 
    RR_SHARED_PTR<PipeBase> GetParent();
 
    bool closed;
 
    uint32_t send_packet_number;
    uint32_t recv_packet_number;
 
    RR_WEAK_PTR<PipeBase> parent;
    int32_t index;
    uint32_t endpoint;
    std::string service_path;
    std::string member_name;
 
    RR_UNORDERED_MAP<uint32_t, RR_INTRUSIVE_PTR<RRValue> > out_of_order_packets;
 
    bool ignore_incoming_packets;
 
    boost::mutex listeners_lock;
    std::list<RR_WEAK_PTR<PipeEndpointBaseListener> > listeners;
 
    detail::async_signal_semaphore pipe_packet_received_semaphore;
 
    RR_WEAK_PTR<RobotRaconteurNode> node;
};

 
template <typename T>
class PipeEndpoint : public PipeEndpointBase
{
  private:
    boost::function<void(RR_SHARED_PTR<PipeEndpoint<T> >)> PipeEndpointClosedCallback;
    boost::mutex PipeEndpointClosedCallback_lock;
 
  public:
    boost::function<void(RR_SHARED_PTR<PipeEndpoint<T> >)> GetPipeEndpointClosedCallback()
    {
        boost::mutex::scoped_lock lock(PipeEndpointClosedCallback_lock);
        return PipeEndpointClosedCallback;
    }

    void SetPipeEndpointClosedCallback(boost::function<void(const RR_SHARED_PTR<PipeEndpoint<T> >&)> callback)
    {
        boost::mutex::scoped_lock lock(PipeEndpointClosedCallback_lock);
        PipeEndpointClosedCallback = callback;
    }

    boost::signals2::signal<void(const RR_SHARED_PTR<PipeEndpoint<T> >&)> PacketReceivedEvent;

    boost::signals2::signal<void(const RR_SHARED_PTR<PipeEndpoint<T> >&, uint32_t)> PacketAckReceivedEvent;

    virtual uint32_t SendPacket(typename boost::call_traits<T>::param_type packet)
    {
        ROBOTRACONTEUR_ASSERT_MULTITHREADED(node);
 
        RR_SHARED_PTR<detail::sync_async_handler<uint32_t> > t =
            RR_MAKE_SHARED<detail::sync_async_handler<uint32_t> >();
        boost::function<void(const RR_SHARED_PTR<uint32_t>&, const RR_SHARED_PTR<RobotRaconteurException>&)> h =
            boost::bind(&detail::sync_async_handler<uint32_t>::operator(), t, RR_BOOST_PLACEHOLDERS(_1),
                        RR_BOOST_PLACEHOLDERS(_2));
        AsyncSendPacket(
            packet, boost::bind(&PipeEndpoint::send_handler, RR_BOOST_PLACEHOLDERS(_1), RR_BOOST_PLACEHOLDERS(_2), h));
        return *t->end();
    }

    virtual void AsyncSendPacket(typename boost::call_traits<T>::param_type packet,
                                 boost::function<void(uint32_t, const RR_SHARED_PTR<RobotRaconteurException>&)> handler)
    {
        AsyncSendPacketBase(RRPrimUtil<T>::PrePack(packet), RR_MOVE(handler));
    }

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

    virtual T PeekNextPacket() { return RRPrimUtil<T>::PreUnpack(PeekPacketBase()); }

    virtual T ReceivePacketWait(int32_t timeout = RR_TIMEOUT_INFINITE)
    {
        return RRPrimUtil<T>::PreUnpack(ReceivePacketBaseWait(timeout));
    }

    virtual T PeekNextPacketWait(int32_t timeout = RR_TIMEOUT_INFINITE)
    {
        return RRPrimUtil<T>::PreUnpack(PeekPacketBaseWait(timeout));
    }

    virtual bool TryReceivePacketWait(T& val, int32_t timeout = RR_TIMEOUT_INFINITE, bool peek = false)
    {
        RR_INTRUSIVE_PTR<RRValue> o;
        if (!TryReceivePacketBaseWait(o, timeout, peek))
            return false;
        val = RRPrimUtil<T>::PreUnpack(o);
        return true;
    }
 
    PipeEndpoint(const RR_SHARED_PTR<PipeBase>& parent, int32_t index, uint32_t endpoint = 0, bool unreliable = false,
                 MemberDefinition_Direction direction = MemberDefinition_Direction_both)
        : PipeEndpointBase(parent, index, endpoint, unreliable, direction){};
 
  protected:
    RR_OVIRTUAL void fire_PipeEndpointClosedCallback() RR_OVERRIDE
    {
        boost::function<void(RR_SHARED_PTR<PipeEndpoint<T> >)> c = GetPipeEndpointClosedCallback();
        if (!c)
            return;
        c(RR_STATIC_POINTER_CAST<PipeEndpoint<T> >(shared_from_this()));
    }
 
    RR_OVIRTUAL void fire_PacketReceivedEvent() RR_OVERRIDE
    {
        PacketReceivedEvent(RR_STATIC_POINTER_CAST<PipeEndpoint<T> >(shared_from_this()));
    }
 
    RR_OVIRTUAL void fire_PacketAckReceivedEvent(uint32_t packetnum) RR_OVERRIDE
    {
        PacketAckReceivedEvent(RR_STATIC_POINTER_CAST<PipeEndpoint<T> >(shared_from_this()), packetnum);
    }
 
    static void send_handler(uint32_t packetnumber, const RR_SHARED_PTR<RobotRaconteurException>& err,
                             const boost::function<void(const RR_SHARED_PTR<uint32_t>&,
                                                        const RR_SHARED_PTR<RobotRaconteurException>&)>& handler)
    {
        handler(RR_MAKE_SHARED<uint32_t>(packetnumber), err);
    }
 
  public:
    RR_OVIRTUAL void Close() RR_OVERRIDE
    {
        PipeEndpointBase::Close();
        {
 
            boost::mutex::scoped_lock lock(PipeEndpointClosedCallback_lock);
            PipeEndpointClosedCallback.clear();
        }
        PacketReceivedEvent.disconnect_all_slots();
        PacketAckReceivedEvent.disconnect_all_slots();
    }
 
  protected:
    virtual void AsyncClose1(const RR_SHARED_PTR<RobotRaconteurException>& err,
                             const boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)>& handler)
    {
        try
        {
            {
                boost::mutex::scoped_lock lock(PipeEndpointClosedCallback_lock);
                PipeEndpointClosedCallback.clear();
            }
            PacketReceivedEvent.disconnect_all_slots();
            PacketAckReceivedEvent.disconnect_all_slots();
        }
        catch (std::exception&)
        {}
 
        handler(err);
    }
 
  public:
    RR_OVIRTUAL void AsyncClose(boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
                                int32_t timeout = 2000) RR_OVERRIDE
    {
        PipeEndpointBase::AsyncClose(boost::bind(&PipeEndpoint<T>::AsyncClose1,
                                                 RR_STATIC_POINTER_CAST<PipeEndpoint<T> >(shared_from_this()),
                                                 RR_BOOST_PLACEHOLDERS(_1), handler),
                                     timeout);
    }
 
  protected:
    RR_OVIRTUAL void RemoteClose() RR_OVERRIDE
    {
        PipeEndpointBase::RemoteClose();
        {
            boost::mutex::scoped_lock lock(PipeEndpointClosedCallback_lock);
            PipeEndpointClosedCallback.clear();
        }
        PacketReceivedEvent.disconnect_all_slots();
        PacketAckReceivedEvent.disconnect_all_slots();
    }
};

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

    static const int32_t ANY_INDEX = -1;

    virtual std::string GetMemberName() = 0;
 
    virtual void PipePacketReceived(const RR_INTRUSIVE_PTR<MessageEntry>& m, uint32_t e = 0) = 0;
 
    virtual void Shutdown() = 0;
 
    virtual std::string GetServicePath() = 0;
 
    virtual void AsyncClose(const RR_SHARED_PTR<PipeEndpointBase>& endpoint, bool remote, uint32_t ee,
                            RR_MOVE_ARG(boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)>) handler,
                            int32_t timeout) = 0;
 
  protected:
    PipeBase();
 
    bool unreliable;
 
    virtual void AsyncSendPipePacket(
        const RR_INTRUSIVE_PTR<RRValue>& data, int32_t index, uint32_t packetnumber, bool requestack, uint32_t endpoint,
        bool unreliable,
        RR_MOVE_ARG(boost::function<void(uint32_t, const RR_SHARED_PTR<RobotRaconteurException>&)>) handler) = 0;
 
    bool rawelements;
 
    void DispatchPacketAck(const RR_INTRUSIVE_PTR<MessageElement>& me, const RR_SHARED_PTR<PipeEndpointBase>& e);
 
    bool DispatchPacket(const RR_INTRUSIVE_PTR<MessageElement>& me, const RR_SHARED_PTR<PipeEndpointBase>& e,
                        uint32_t& packetnumber);
 
    RR_INTRUSIVE_PTR<MessageElement> PackPacket(const RR_INTRUSIVE_PTR<RRValue>& data, int32_t index,
                                                uint32_t packetnumber, bool requestack);
 
    virtual void DeleteEndpoint(const RR_SHARED_PTR<PipeEndpointBase>& e) = 0;
 
    virtual RR_INTRUSIVE_PTR<MessageElementData> PackData(const RR_INTRUSIVE_PTR<RRValue>& data)
    {
        return GetNode()->PackVarType(data);
    }
 
    virtual RR_INTRUSIVE_PTR<RRValue> UnpackData(const RR_INTRUSIVE_PTR<MessageElement>& mdata)
    {
        return GetNode()->UnpackVarType(mdata);
    }
 
    RR_WEAK_PTR<RobotRaconteurNode> node;
 
    MemberDefinition_Direction direction;
 
  public:
    RR_SHARED_PTR<RobotRaconteurNode> GetNode();

    MemberDefinition_Direction Direction();

    bool IsUnreliable();
};

template <typename T>
class Pipe : public virtual PipeBase
{
  public:
    friend class PipeEndpointBase;
 
    Pipe(boost::function<void(const RR_INTRUSIVE_PTR<RRValue>&)> verify) { this->verify = RR_MOVE(verify); }
 
    RR_OVIRTUAL ~Pipe() RR_OVERRIDE {}

    virtual boost::function<void(const RR_SHARED_PTR<PipeEndpoint<T> >&)> GetPipeConnectCallback() = 0;

    virtual void SetPipeConnectCallback(boost::function<void(const RR_SHARED_PTR<PipeEndpoint<T> >&)> function) = 0;

    virtual RR_SHARED_PTR<PipeEndpoint<T> > Connect(int32_t index) = 0;

    virtual void AsyncConnect(
        int32_t index,
        boost::function<void(const RR_SHARED_PTR<PipeEndpoint<T> >&, const RR_SHARED_PTR<RobotRaconteurException>&)>
            handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) = 0;

    virtual void AsyncConnect(
        boost::function<void(const RR_SHARED_PTR<PipeEndpoint<T> >&, const RR_SHARED_PTR<RobotRaconteurException>&)>
            handler,
        int32_t timeout = RR_TIMEOUT_INFINITE)
    {
        AsyncConnect(-1, RR_MOVE(handler), timeout);
    }
 
    RR_OVIRTUAL RR_INTRUSIVE_PTR<MessageElementData> PackData(const RR_INTRUSIVE_PTR<RRValue>& data) RR_OVERRIDE
    {
        if (verify)
        {
            verify(data);
        }
        return GetNode()->template PackAnyType<typename RRPrimUtil<T>::BoxedType>(data);
    }
 
    RR_OVIRTUAL RR_INTRUSIVE_PTR<RRValue> UnpackData(const RR_INTRUSIVE_PTR<MessageElement>& mdata) RR_OVERRIDE
    {
        if (!verify)
        {
            return GetNode()->template UnpackAnyType<typename RRPrimUtil<T>::BoxedType>(mdata);
        }
        else
        {
            RR_INTRUSIVE_PTR<RRValue> ret = GetNode()->template UnpackAnyType<typename RRPrimUtil<T>::BoxedType>(mdata);
            verify(ret);
            return ret;
        }
    }
 
  protected:
    boost::function<void(const RR_INTRUSIVE_PTR<RRValue>&)> verify;
};
 
class ROBOTRACONTEUR_CORE_API ServiceStub;
 
class ROBOTRACONTEUR_CORE_API PipeClientBase : public virtual PipeBase
{
  public:
    friend class PipeSubscriptionBase;
    friend class detail::PipeSubscription_connection;
 
    RR_OVIRTUAL ~PipeClientBase() RR_OVERRIDE {}
 
    RR_OVIRTUAL std::string GetMemberName() RR_OVERRIDE;
 
    RR_OVIRTUAL void PipePacketReceived(const RR_INTRUSIVE_PTR<MessageEntry>& m, uint32_t e = 0) RR_OVERRIDE;
 
    RR_OVIRTUAL void Shutdown() RR_OVERRIDE;
 
    RR_OVIRTUAL void AsyncClose(const RR_SHARED_PTR<PipeEndpointBase>& endpoint, bool remote, uint32_t ee,
                                RR_MOVE_ARG(boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)>)
                                    handler,
                                int32_t timeout) RR_OVERRIDE;
 
    RR_SHARED_PTR<ServiceStub> GetStub();
 
    RR_OVIRTUAL std::string GetServicePath() RR_OVERRIDE;
 
  protected:
    RR_OVIRTUAL void AsyncSendPipePacket(
        const RR_INTRUSIVE_PTR<RRValue>& data, int32_t index, uint32_t packetnumber, bool requestack, uint32_t endpoint,
        bool unreliable,
        RR_MOVE_ARG(boost::function<void(uint32_t, const RR_SHARED_PTR<RobotRaconteurException>&)>)
            handler) RR_OVERRIDE;
 
    std::string m_MemberName;
 
    RR_UNORDERED_MAP<int32_t, RR_SHARED_PTR<PipeEndpointBase> > pipeendpoints;
    boost::mutex pipeendpoints_lock;
 
    RR_WEAK_PTR<ServiceStub> stub;
 
    std::list<boost::tuple<int32_t, int32_t> > connecting_endpoints;
    int32_t connecting_key_count;
    RR_UNORDERED_MAP<int32_t, RR_SHARED_PTR<PipeEndpointBase> > early_endpoints;
    std::string service_path;
    uint32_t endpoint;
 
    void AsyncConnect_internal(int32_t index,
                               RR_MOVE_ARG(boost::function<void(const RR_SHARED_PTR<PipeEndpointBase>&,
                                                                const RR_SHARED_PTR<RobotRaconteurException>&)>)
                                   handler,
                               int32_t timeout);
 
    void AsyncConnect_internal1(const RR_INTRUSIVE_PTR<MessageEntry>& ret,
                                const RR_SHARED_PTR<RobotRaconteurException>& err, int32_t index, int32_t key,
                                boost::function<void(const RR_SHARED_PTR<PipeEndpointBase>&,
                                                     const RR_SHARED_PTR<RobotRaconteurException>&)>& handler);
 
    PipeClientBase(boost::string_ref name, const RR_SHARED_PTR<ServiceStub>& stub, bool unreliable,
                   MemberDefinition_Direction direction);
 
    virtual RR_SHARED_PTR<PipeEndpointBase> CreateNewPipeEndpoint(int32_t index, bool unreliable,
                                                                  MemberDefinition_Direction direction) = 0;
 
    RR_OVIRTUAL void DeleteEndpoint(const RR_SHARED_PTR<PipeEndpointBase>& e) RR_OVERRIDE;
};
 
template <typename T>
class PipeClient : public virtual Pipe<T>, public virtual PipeClientBase
{
  public:
    RR_OVIRTUAL ~PipeClient() RR_OVERRIDE {}
 
    RR_OVIRTUAL boost::function<void(const RR_SHARED_PTR<PipeEndpoint<T> >&)> GetPipeConnectCallback() RR_OVERRIDE
    {
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, endpoint, service_path, m_MemberName,
                                                "GetPipeConnectCallback is not valid for PipeClient");
        throw InvalidOperationException("Not valid for client");
    }
 
    RR_OVIRTUAL void SetPipeConnectCallback(boost::function<void(const RR_SHARED_PTR<PipeEndpoint<T> >&)> function)
        RR_OVERRIDE
    {
        RR_UNUSED(function);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, endpoint, service_path, m_MemberName,
                                                "SetPipeConnectCallback is not valid for PipeClient");
        throw InvalidOperationException("Not valid for client");
    }
 
    RR_OVIRTUAL void AsyncConnect(
        int32_t index,
        boost::function<void(const RR_SHARED_PTR<PipeEndpoint<T> >&, const RR_SHARED_PTR<RobotRaconteurException>&)>
            handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
 
        AsyncConnect_internal(index,
                              boost::bind(handler,
                                          boost::bind(&PipeClient<T>::AsyncConnect_cast, RR_BOOST_PLACEHOLDERS(_1)),
                                          RR_BOOST_PLACEHOLDERS(_2)),
                              timeout);
    }
 
    RR_OVIRTUAL RR_SHARED_PTR<PipeEndpoint<T> > Connect(int32_t index) RR_OVERRIDE
    {
        ROBOTRACONTEUR_ASSERT_MULTITHREADED(node);
 
        RR_SHARED_PTR<detail::sync_async_handler<PipeEndpoint<T> > > t =
            RR_MAKE_SHARED<detail::sync_async_handler<PipeEndpoint<T> > >();
        AsyncConnect(index,
                     boost::bind(&detail::sync_async_handler<PipeEndpoint<T> >::operator(), t,
                                 RR_BOOST_PLACEHOLDERS(_1), RR_BOOST_PLACEHOLDERS(_2)),
                     GetNode()->GetRequestTimeout());
        return t->end();
    }
 
    PipeClient(boost::string_ref name, const RR_SHARED_PTR<ServiceStub>& stub, bool unreliable = false,
               MemberDefinition_Direction direction = MemberDefinition_Direction_both,
               boost::function<void(const RR_INTRUSIVE_PTR<RRValue>&)> verify = RR_NULL_FN)
        : Pipe<T>(verify), PipeClientBase(name, stub, unreliable, direction)
    {
        rawelements = (boost::is_same<T, RR_INTRUSIVE_PTR<MessageElement> >::value);
    }
 
    using PipeClientBase::AsyncClose;
    using PipeClientBase::AsyncSendPipePacket;
    using PipeClientBase::GetMemberName;
    using PipeClientBase::PipePacketReceived;
    using PipeClientBase::Shutdown;
 
  protected:
    static RR_SHARED_PTR<PipeEndpoint<T> > AsyncConnect_cast(const RR_SHARED_PTR<PipeEndpointBase>& b)
    {
        return rr_cast<PipeEndpoint<T> >(b);
    }
 
    RR_OVIRTUAL RR_SHARED_PTR<PipeEndpointBase> CreateNewPipeEndpoint(int32_t index, bool unreliable,
                                                                      MemberDefinition_Direction direction) RR_OVERRIDE
    {
        return RR_MAKE_SHARED<PipeEndpoint<T> >(RR_STATIC_POINTER_CAST<PipeBase>(shared_from_this()), index, 0,
                                                unreliable, direction);
    }
};
 
class ROBOTRACONTEUR_CORE_API ServiceSkel;
class ROBOTRACONTEUR_CORE_API PipeServerBase : public virtual PipeBase
{
  public:
    RR_OVIRTUAL ~PipeServerBase() RR_OVERRIDE {}
 
    RR_OVIRTUAL std::string GetMemberName() RR_OVERRIDE;
 
    RR_OVIRTUAL void PipePacketReceived(const RR_INTRUSIVE_PTR<MessageEntry>& m, uint32_t e = 0) RR_OVERRIDE;
 
    RR_OVIRTUAL void Shutdown() RR_OVERRIDE;
 
    RR_OVIRTUAL void AsyncSendPipePacket(
        const RR_INTRUSIVE_PTR<RRValue>& data, int32_t index, uint32_t packetnumber, bool requestack, uint32_t endpoint,
        bool unreliable,
        RR_MOVE_ARG(boost::function<void(uint32_t, const RR_SHARED_PTR<RobotRaconteurException>&)>)
            handler) RR_OVERRIDE;
 
    RR_OVIRTUAL void AsyncClose(const RR_SHARED_PTR<PipeEndpointBase>& endpoint, bool remote, uint32_t ee,
                                RR_MOVE_ARG(boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)>)
                                    handler,
                                int32_t timeout) RR_OVERRIDE;
 
    virtual RR_INTRUSIVE_PTR<MessageEntry> PipeCommand(const RR_INTRUSIVE_PTR<MessageEntry>& m, uint32_t e);
 
    RR_SHARED_PTR<ServiceSkel> GetSkel();
 
    RR_OVIRTUAL std::string GetServicePath() RR_OVERRIDE;
 
  protected:
    std::string m_MemberName;
    std::string service_path;
 
    struct pipe_endpoint_server_id
    {
        pipe_endpoint_server_id(uint32_t endpoint, int32_t index)
        {
            this->endpoint = endpoint;
            this->index = index;
        }
 
        uint32_t endpoint;
        int32_t index;
 
        bool operator==(const pipe_endpoint_server_id& rhs) const
        {
            return (endpoint == rhs.endpoint && index == rhs.index);
        }
    };
 
    struct hash_value
    {
        std::size_t operator()(pipe_endpoint_server_id const& e) const
        {
            std::size_t seed = 0;
            boost::hash_combine(seed, e.endpoint);
            boost::hash_combine(seed, e.index);
            return seed;
        }
    };
 
    RR_UNORDERED_MAP<pipe_endpoint_server_id, RR_SHARED_PTR<PipeEndpointBase>, hash_value> pipeendpoints;
    boost::mutex pipeendpoints_lock;
 
    RR_WEAK_PTR<ServiceSkel> skel;
 
    PipeServerBase(boost::string_ref name, const RR_SHARED_PTR<ServiceSkel>& skel, bool unreliable,
                   MemberDefinition_Direction direction);
 
    virtual RR_SHARED_PTR<PipeEndpointBase> CreateNewPipeEndpoint(int32_t index, uint32_t endpoint, bool unreliable,
                                                                  MemberDefinition_Direction direction) = 0;
 
    RR_OVIRTUAL void DeleteEndpoint(const RR_SHARED_PTR<PipeEndpointBase>& e) RR_OVERRIDE;
 
    virtual void fire_PipeConnectCallback(const RR_SHARED_PTR<PipeEndpointBase>& e) = 0;
 
    bool init;
    boost::signals2::connection listener_connection;
 
  public:
    void ClientDisconnected(const RR_SHARED_PTR<ServerContext>& context, ServerServiceListenerEventType ev,
                            const RR_SHARED_PTR<void>& param);
};
 
template <typename T>
class PipeServer : public virtual PipeServerBase, public virtual Pipe<T>
{
 
  public:
    RR_OVIRTUAL ~PipeServer() RR_OVERRIDE {}
 
    RR_OVIRTUAL boost::function<void(const RR_SHARED_PTR<PipeEndpoint<T> >&)> GetPipeConnectCallback() RR_OVERRIDE
    {
        return callback;
    }
 
    RR_OVIRTUAL void SetPipeConnectCallback(boost::function<void(const RR_SHARED_PTR<PipeEndpoint<T> >&)> function)
        RR_OVERRIDE
    {
        callback = function;
    }
 
    RR_OVIRTUAL RR_SHARED_PTR<PipeEndpoint<T> > Connect(int32_t index) RR_OVERRIDE
    {
        RR_UNUSED(index);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, -1, service_path, m_MemberName,
                                                "Connect is not valid for PipeServer");
        throw InvalidOperationException("Not valid for server");
    }
 
    RR_OVIRTUAL void AsyncConnect(
        int32_t index,
        boost::function<void(const RR_SHARED_PTR<PipeEndpoint<T> >&, const RR_SHARED_PTR<RobotRaconteurException>&)>
            handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        RR_UNUSED(index);
        RR_UNUSED(handler);
        RR_UNUSED(timeout);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, -1, service_path, m_MemberName,
                                                "AsyncConnect is not valid for PipeServer");
        throw InvalidOperationException("Not valid for server");
    }
 
    PipeServer(boost::string_ref name, const RR_SHARED_PTR<ServiceSkel>& skel, bool unreliable = false,
               MemberDefinition_Direction direction = MemberDefinition_Direction_both,
               boost::function<void(const RR_INTRUSIVE_PTR<RRValue>&)> verify = RR_NULL_FN)
        : PipeServerBase(name, skel, unreliable, direction), Pipe<T>(verify)
    {
        rawelements = (boost::is_same<T, RR_INTRUSIVE_PTR<MessageElement> >::value);
    }
 
  protected:
    RR_OVIRTUAL RR_SHARED_PTR<PipeEndpointBase> CreateNewPipeEndpoint(int32_t index, uint32_t endpoint, bool unreliable,
                                                                      MemberDefinition_Direction direction) RR_OVERRIDE
    {
        return RR_MAKE_SHARED<PipeEndpoint<T> >(RR_STATIC_POINTER_CAST<PipeBase>(shared_from_this()), index, endpoint,
                                                unreliable, direction);
    }
 
    boost::function<void(RR_SHARED_PTR<PipeEndpoint<T> >)> callback;
 
    RR_OVIRTUAL void fire_PipeConnectCallback(const RR_SHARED_PTR<PipeEndpointBase>& e) RR_OVERRIDE
    {
        if (!callback)
            return;
        callback(RR_STATIC_POINTER_CAST<PipeEndpoint<T> >(e));
    }
 
  public:
    RR_OVIRTUAL void Shutdown() RR_OVERRIDE
    {
        PipeServerBase::Shutdown();
 
        callback.clear();
    }
};
 
namespace detail
{
class PipeBroadcasterBase_connected_endpoint;
struct PipeBroadcasterBase_async_send_operation;
} // namespace detail

class ROBOTRACONTEUR_CORE_API PipeBroadcasterBase : public RR_ENABLE_SHARED_FROM_THIS<PipeBroadcasterBase>,
                                                    private boost::noncopyable
{
  public:
    virtual ~PipeBroadcasterBase();
 
    size_t GetActivePipeEndpointCount();

    boost::function<bool(const RR_SHARED_PTR<PipeBroadcasterBase>&, uint32_t, int32_t)> GetPredicate();

    void SetPredicate(boost::function<bool(const RR_SHARED_PTR<PipeBroadcasterBase>&, uint32_t, int32_t)> f);

    int32_t GetMaxBacklog();

    void SetMaxBacklog(int32_t maximum_backlog);
 
  protected:
    PipeBroadcasterBase();
 
    void InitBase(const RR_SHARED_PTR<PipeBase>& pipe, int32_t maximum_backlog = -1);
 
    void EndpointConnectedBase(const RR_SHARED_PTR<PipeEndpointBase>& ep);
 
    void EndpointClosedBase(const RR_SHARED_PTR<detail::PipeBroadcasterBase_connected_endpoint>& ep);
 
    void PacketAckReceivedBase(const RR_SHARED_PTR<detail::PipeBroadcasterBase_connected_endpoint>& ep, uint32_t id);
 
    void handle_send(int32_t id, const RR_SHARED_PTR<RobotRaconteurException>& err,
                     const RR_SHARED_PTR<detail::PipeBroadcasterBase_connected_endpoint>& ep,
                     const RR_SHARED_PTR<detail::PipeBroadcasterBase_async_send_operation>& op, int32_t key,
                     int32_t send_key, const boost::function<void()>& handler);
 
    void SendPacketBase(const RR_INTRUSIVE_PTR<RRValue>& packet);
 
    void AsyncSendPacketBase(const RR_INTRUSIVE_PTR<RRValue>& packet, RR_MOVE_ARG(boost::function<void()>) handler);
 
    virtual void AttachPipeServerEvents(const RR_SHARED_PTR<PipeServerBase>& p);
 
    virtual void AttachPipeEndpointEvents(const RR_SHARED_PTR<PipeEndpointBase>& p,
                                          const RR_SHARED_PTR<detail::PipeBroadcasterBase_connected_endpoint>& cep);
 
    RR_SHARED_PTR<PipeBase> GetPipeBase();
 
    std::list<RR_SHARED_PTR<detail::PipeBroadcasterBase_connected_endpoint> > endpoints;
    boost::mutex endpoints_lock;
 
    RR_WEAK_PTR<PipeServerBase> pipe;
    RR_WEAK_PTR<RobotRaconteurNode> node;
    int32_t maximum_backlog;
    std::string service_path;
    std::string member_name;
 
    bool copy_element;
 
    boost::function<bool(const RR_SHARED_PTR<PipeBroadcasterBase>&, uint32_t, int32_t)> predicate;
};

template <typename T>
class PipeBroadcaster : public PipeBroadcasterBase
{
 
  public:
    PipeBroadcaster() {}

    void Init(RR_SHARED_PTR<Pipe<T> > pipe, int32_t maximum_backlog = -1) { InitBase(pipe, maximum_backlog); }

    void SendPacket(T packet) { SendPacketBase(RRPrimUtil<T>::PrePack(packet)); }

    void AsyncSendPacket(T packet, boost::function<void()> handler)
    {
        AsyncSendPacketBase(RRPrimUtil<T>::PrePack(packet), RR_MOVE(handler));
    }

    RR_SHARED_PTR<Pipe<T> > GetPipe() { return rr_cast<Pipe<T> >(GetPipeBase()); }
 
  protected:
    RR_OVIRTUAL void AttachPipeServerEvents(const RR_SHARED_PTR<PipeServerBase>& p) RR_OVERRIDE
    {
        RR_SHARED_PTR<PipeServer<T> > p_T = rr_cast<PipeServer<T> >(p);
 
        p_T->SetPipeConnectCallback(
            boost::bind(&PipeBroadcaster::EndpointConnectedBase, shared_from_this(), RR_BOOST_PLACEHOLDERS(_1)));
    }
 
    RR_OVIRTUAL void AttachPipeEndpointEvents(const RR_SHARED_PTR<PipeEndpointBase>& ep,
                                              const RR_SHARED_PTR<detail::PipeBroadcasterBase_connected_endpoint>& cep)
        RR_OVERRIDE
    {
        RR_SHARED_PTR<PipeEndpoint<T> > ep_T = rr_cast<PipeEndpoint<T> >(ep);
 
        ep_T->SetPipeEndpointClosedCallback(boost::bind(&PipeBroadcaster::EndpointClosedBase, shared_from_this(), cep));
        ep_T->PacketAckReceivedEvent.connect(
            boost::bind(&PipeBroadcaster::PacketAckReceivedBase, shared_from_this(), cep, RR_BOOST_PLACEHOLDERS(_2)));
    }
};
#ifndef ROBOTRACONTEUR_NO_CXX11_TEMPLATE_ALIASES
using PipeEndpointBasePtr = RR_SHARED_PTR<PipeEndpointBase>;
template <typename T>
using PipeEndpointPtr = RR_SHARED_PTR<PipeEndpoint<T> >;
using PipeBasePtr = RR_SHARED_PTR<PipeBase>;
template <typename T>
using PipePtr = RR_SHARED_PTR<Pipe<T> >;
template <typename T>
using PipeBroadcasterPtr = RR_SHARED_PTR<PipeBroadcaster<T> >;
#endif
} // namespace RobotRaconteur
 
#ifdef _MSVC_VER
#pragma warning(pop)
#endif