WireMember.h

Go to the documentation of this file.

 
#pragma once
 
#include "RobotRaconteur/DataTypes.h"
#include "RobotRaconteur/Message.h"
#include "RobotRaconteur/RobotRaconteurNode.h"
#include "RobotRaconteur/AsyncUtils.h"
#include <boost/call_traits.hpp>
 
#pragma warning(push)
#pragma warning(disable : 4250)
#pragma warning(disable : 4996)
#include <boost/signals2.hpp>
 
namespace RobotRaconteur
{
class ROBOTRACONTEUR_CORE_API WireBase;
class ROBOTRACONTEUR_CORE_API WireConnectionBase;
class ROBOTRACONTEUR_CORE_API WireConnectionBaseListener;
namespace detail
{
class WireSubscription_connection;
ROBOTRACONTEUR_CORE_API bool WireConnectionBase_IsValueExpired(RR_WEAK_PTR<RobotRaconteurNode> node,
                                                               const boost::posix_time::ptime& recv_time,
                                                               int32_t lifespan);
} // namespace detail
 
class ROBOTRACONTEUR_CORE_API WireConnectionBase : public RR_ENABLE_SHARED_FROM_THIS<WireConnectionBase>,
                                                   private boost::noncopyable
{
 
    friend class WireBase;
    friend class WireClientBase;
    friend class WireServerBase;
    friend class WireBroadcasterBase;
    friend class WireSubscriptionBase;
    friend class detail::WireSubscription_connection;
 
  public:
    virtual uint32_t GetEndpoint();
 
    virtual TimeSpec GetLastValueReceivedTime();
 
    virtual TimeSpec GetLastValueSentTime();
 
    virtual void Close();
 
    virtual void AsyncClose(boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
                            int32_t timeout);
 
    WireConnectionBase(const RR_SHARED_PTR<WireBase>& parent, uint32_t endpoint = 0,
                       MemberDefinition_Direction direction = MemberDefinition_Direction_both);
 
    virtual ~WireConnectionBase() {}
 
    virtual void WirePacketReceived(TimeSpec timespec, const RR_INTRUSIVE_PTR<RRValue>& packet);
 
    virtual bool GetInValueValid();
 
    virtual bool GetOutValueValid();
 
    bool WaitInValueValid(int32_t timeout = RR_TIMEOUT_INFINITE);
 
    bool WaitOutValueValid(int32_t timeout = RR_TIMEOUT_INFINITE);
 
    RR_SHARED_PTR<RobotRaconteurNode> GetNode();
 
    virtual bool GetIgnoreInValue();
 
    virtual void SetIgnoreInValue(bool ignore);
 
    virtual void AddListener(const RR_SHARED_PTR<WireConnectionBaseListener>& listener);
 
    MemberDefinition_Direction Direction();
 
    virtual int32_t GetInValueLifespan();
    virtual void SetInValueLifespan(int32_t millis);
 
    virtual int32_t GetOutValueLifespan();
    virtual void SetOutValueLifespan(int32_t millis);
 
  protected:
    virtual void RemoteClose();
 
    RR_INTRUSIVE_PTR<RRValue> inval;
    RR_INTRUSIVE_PTR<RRValue> outval;
 
    bool inval_valid;
    TimeSpec lasttime_send;
    boost::posix_time::ptime lasttime_send_local;
 
    bool outval_valid;
    TimeSpec lasttime_recv;
    boost::posix_time::ptime lasttime_recv_local;
 
    boost::condition_variable inval_wait;
    boost::condition_variable outval_wait;
 
    int32_t inval_lifespan;
    int32_t outval_lifespan;
 
    uint32_t endpoint;
    RR_WEAK_PTR<WireBase> parent;
    std::string service_path;
    std::string member_name;
 
    boost::mutex sendlock;
    boost::mutex recvlock;
 
    bool send_closed;
    bool recv_closed;
 
    RR_INTRUSIVE_PTR<RRValue> GetInValueBase();
 
    RR_INTRUSIVE_PTR<RRValue> GetOutValueBase();
 
    void SetOutValueBase(const RR_INTRUSIVE_PTR<RRValue>& value);
 
    bool TryGetInValueBase(RR_INTRUSIVE_PTR<RRValue>& value, TimeSpec& time);
    bool TryGetOutValueBase(RR_INTRUSIVE_PTR<RRValue>& value, TimeSpec& time);
 
    virtual void fire_WireValueChanged(const RR_INTRUSIVE_PTR<RRValue>& value, TimeSpec time) = 0;
 
    virtual void fire_WireClosedCallback() = 0;
 
    void Shutdown();
 
    RR_SHARED_PTR<WireBase> GetParent();
 
    boost::mutex inval_lock;
    boost::mutex outval_lock;
 
    bool ignore_inval;
 
    boost::mutex listeners_lock;
    std::list<RR_WEAK_PTR<WireConnectionBaseListener> > listeners;
 
    detail::async_signal_semaphore wire_value_changed_semaphore;
 
    RR_WEAK_PTR<RobotRaconteurNode> node;
 
    MemberDefinition_Direction direction;
};
 
template <typename T>
class WireConnection : public WireConnectionBase
{
  private:
    boost::function<void(RR_SHARED_PTR<WireConnection<T> >)> WireConnectionClosedCallback;
    boost::mutex WireConnectionClosedCallback_lock;
 
  public:
    boost::signals2::signal<void(const RR_SHARED_PTR<WireConnection<T> >& connection, T value, TimeSpec time)>
        WireValueChanged;
 
    boost::function<void(RR_SHARED_PTR<WireConnection<T> >)> GetWireConnectionClosedCallback()
    {
        boost::mutex::scoped_lock lock(WireConnectionClosedCallback_lock);
        return WireConnectionClosedCallback;
    }
 
    void SetWireConnectionClosedCallback(boost::function<void(const RR_SHARED_PTR<WireConnection<T> >&)> callback)
    {
        boost::mutex::scoped_lock lock(WireConnectionClosedCallback_lock);
        WireConnectionClosedCallback = callback;
    }
 
    RR_OVIRTUAL ~WireConnection() RR_OVERRIDE {}
 
    virtual T GetInValue() { return RRPrimUtil<T>::PreUnpack(GetInValueBase()); }
 
    virtual T GetOutValue() { return RRPrimUtil<T>::PreUnpack(GetOutValueBase()); }
 
    virtual void SetOutValue(typename boost::call_traits<T>::param_type value)
    {
        SetOutValueBase(RRPrimUtil<T>::PrePack(value));
    }
 
    bool TryGetInValue(T& value, TimeSpec& time)
    {
        RR_INTRUSIVE_PTR<RRValue> o;
        if (!TryGetInValueBase(o, time))
            return false;
        value = RRPrimUtil<T>::PreUnpack(o);
        return true;
    }
 
    bool TryGetOutValue(T& value, TimeSpec& time)
    {
        RR_INTRUSIVE_PTR<RRValue> o;
        if (!TryGetOutValueBase(o, time))
            return false;
        value = RRPrimUtil<T>::PreUnpack(o);
        return true;
    }
 
    WireConnection(const RR_SHARED_PTR<WireBase>& parent, uint32_t endpoint = 0,
                   MemberDefinition_Direction direction = MemberDefinition_Direction_both)
        : WireConnectionBase(parent, endpoint, direction)
    {}
 
  protected:
    RR_OVIRTUAL void fire_WireValueChanged(const RR_INTRUSIVE_PTR<RRValue>& value, TimeSpec time) RR_OVERRIDE
    {
        WireValueChanged(RR_STATIC_POINTER_CAST<WireConnection<T> >(shared_from_this()),
                         RRPrimUtil<T>::PreUnpack(value), time);
    }
 
    RR_OVIRTUAL void fire_WireClosedCallback() RR_OVERRIDE
    {
        boost::function<void(RR_SHARED_PTR<WireConnection<T> >)> c = GetWireConnectionClosedCallback();
        if (!c)
            return;
        c(RR_STATIC_POINTER_CAST<WireConnection<T> >(shared_from_this()));
    }
 
  public:
    RR_OVIRTUAL void Close() RR_OVERRIDE
    {
        WireConnectionBase::Close();
        {
            boost::mutex::scoped_lock lock(WireConnectionClosedCallback_lock);
            WireConnectionClosedCallback.clear();
        }
        WireValueChanged.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(WireConnectionClosedCallback_lock);
                WireConnectionClosedCallback.clear();
            }
            WireValueChanged.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
    {
        WireConnectionBase::AsyncClose(boost::bind(&WireConnection<T>::AsyncClose1,
                                                   RR_STATIC_POINTER_CAST<WireConnection<T> >(shared_from_this()),
                                                   RR_BOOST_PLACEHOLDERS(_1), handler),
                                       timeout);
    }
 
  protected:
    RR_OVIRTUAL void RemoteClose() RR_OVERRIDE
    {
        WireConnectionBase::RemoteClose();
        {
            boost::mutex::scoped_lock lock(WireConnectionClosedCallback_lock);
            WireConnectionClosedCallback.clear();
        }
        WireValueChanged.disconnect_all_slots();
    }
};
 
class ROBOTRACONTEUR_CORE_API WireBase : public RR_ENABLE_SHARED_FROM_THIS<WireBase>, private boost::noncopyable
{
 
  public:
    friend class WireConnectionBase;
 
    virtual ~WireBase() {}
 
    virtual std::string GetMemberName() = 0;
 
    virtual std::string GetServicePath() = 0;
 
    virtual void WirePacketReceived(const RR_INTRUSIVE_PTR<MessageEntry>& m, uint32_t e = 0) = 0;
 
    virtual void Shutdown() = 0;
 
    virtual void AsyncClose(const RR_SHARED_PTR<WireConnectionBase>& endpoint, bool remote, uint32_t ee,
                            RR_MOVE_ARG(boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)>) handler,
                            int32_t timeout) = 0;
 
  protected:
    WireBase();
 
    virtual void SendWirePacket(const RR_INTRUSIVE_PTR<RRValue>& data, TimeSpec time, uint32_t endpoint) = 0;
 
    bool rawelements;
 
    void DispatchPacket(const RR_INTRUSIVE_PTR<MessageEntry>& me, const RR_SHARED_PTR<WireConnectionBase>& e);
 
    RR_INTRUSIVE_PTR<RRValue> UnpackPacket(const RR_INTRUSIVE_PTR<MessageEntry>& me, TimeSpec& ts);
 
    RR_INTRUSIVE_PTR<MessageEntry> PackPacket(const RR_INTRUSIVE_PTR<RRValue>& data, TimeSpec time);
 
    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();
};
 
template <typename T>
class Wire : public virtual WireBase
{
 
    friend class WireConnectionBase;
 
  public:
    Wire(boost::function<void(const RR_INTRUSIVE_PTR<RRValue>&)> verify) { this->verify = RR_MOVE(verify); }
 
    RR_OVIRTUAL ~Wire() RR_OVERRIDE {}
 
    virtual RR_SHARED_PTR<WireConnection<T> > Connect() = 0;
 
    virtual void AsyncConnect(
        boost::function<void(const RR_SHARED_PTR<WireConnection<T> >&, const RR_SHARED_PTR<RobotRaconteurException>&)>
            handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) = 0;
 
    virtual T PeekInValue(TimeSpec& ts) = 0;
 
    virtual T PeekOutValue(TimeSpec& ts) = 0;
 
    virtual void PokeOutValue(const T& value) = 0;
 
    virtual void AsyncPeekInValue(
        boost::function<void(const T&, const TimeSpec&, const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) = 0;
 
    virtual void AsyncPeekOutValue(
        boost::function<void(const T&, const TimeSpec&, const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) = 0;
 
    virtual void AsyncPokeOutValue(const T& value,
                                   boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
                                   int32_t timeout = RR_TIMEOUT_INFINITE) = 0;
 
    virtual boost::function<void(const RR_SHARED_PTR<WireConnection<T> >&)> GetWireConnectCallback() = 0;
 
    virtual void SetWireConnectCallback(boost::function<void(const RR_SHARED_PTR<WireConnection<T> >&)> function) = 0;
 
    virtual boost::function<T(const uint32_t&)> GetPeekInValueCallback() = 0;
 
    virtual void SetPeekInValueCallback(boost::function<T(const uint32_t&)> function) = 0;
 
    virtual boost::function<T(const uint32_t&)> GetPeekOutValueCallback() = 0;
 
    virtual void SetPeekOutValueCallback(boost::function<T(const uint32_t&)> function) = 0;
 
    virtual boost::function<void(const T&, const TimeSpec&, const uint32_t&)> GetPokeOutValueCallback() = 0;
 
    virtual void SetPokeOutValueCallback(
        boost::function<void(const T&, const TimeSpec&, const uint32_t&)> function) = 0;
 
  protected:
    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;
        }
    }
 
    boost::function<void(const RR_INTRUSIVE_PTR<RRValue>&)> verify;
};
 
class ROBOTRACONTEUR_CORE_API ServiceStub;
 
class ROBOTRACONTEUR_CORE_API WireClientBase : public virtual WireBase
{
    friend class WireConnectionBase;
    friend class WireSubscriptionBase;
    friend class detail::WireSubscription_connection;
 
  public:
    RR_OVIRTUAL ~WireClientBase() RR_OVERRIDE {}
 
    RR_OVIRTUAL std::string GetMemberName() RR_OVERRIDE;
 
    RR_OVIRTUAL std::string GetServicePath() RR_OVERRIDE;
 
    RR_OVIRTUAL void WirePacketReceived(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<WireConnectionBase>& 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();
 
  protected:
    RR_OVIRTUAL void SendWirePacket(const RR_INTRUSIVE_PTR<RRValue>& packet, TimeSpec time,
                                    uint32_t endpoint) RR_OVERRIDE;
 
    std::string m_MemberName;
    std::string service_path;
    uint32_t endpoint;
 
    RR_SHARED_PTR<WireConnectionBase> connection;
    boost::mutex connection_lock;
 
    RR_WEAK_PTR<ServiceStub> stub;
 
    void AsyncConnect_internal(RR_MOVE_ARG(boost::function<void(const RR_SHARED_PTR<WireConnectionBase>&,
                                                                const RR_SHARED_PTR<RobotRaconteurException>&)>)
                                   handler,
                               int32_t timeout);
 
    void AsyncConnect_internal1(const RR_INTRUSIVE_PTR<MessageEntry>& ret,
                                const RR_SHARED_PTR<RobotRaconteurException>& err,
                                boost::function<void(const RR_SHARED_PTR<WireConnectionBase>&,
                                                     const RR_SHARED_PTR<RobotRaconteurException>&)>& handler);
 
    WireClientBase(boost::string_ref name, const RR_SHARED_PTR<ServiceStub>& stub,
                   MemberDefinition_Direction direction);
 
    virtual RR_SHARED_PTR<WireConnectionBase> CreateNewWireConnection(MemberDefinition_Direction direction) = 0;
 
    RR_INTRUSIVE_PTR<RRValue> PeekInValueBase(TimeSpec& ts);
    RR_INTRUSIVE_PTR<RRValue> PeekOutValueBase(TimeSpec& ts);
    void PokeOutValueBase(const RR_INTRUSIVE_PTR<RRValue>& value);
 
    void AsyncPeekInValueBase(RR_MOVE_ARG(boost::function<void(const RR_INTRUSIVE_PTR<RRValue>&, const TimeSpec&,
                                                               const RR_SHARED_PTR<RobotRaconteurException>&)>) handler,
                              int32_t timeout = RR_TIMEOUT_INFINITE);
    void AsyncPeekOutValueBase(RR_MOVE_ARG(boost::function<void(const RR_INTRUSIVE_PTR<RRValue>&, const TimeSpec&,
                                                                const RR_SHARED_PTR<RobotRaconteurException>&)>)
                                   handler,
                               int32_t timeout = RR_TIMEOUT_INFINITE);
    void AsyncPokeOutValueBase(const RR_INTRUSIVE_PTR<RRValue>& value,
                               RR_MOVE_ARG(boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)>)
                                   handler,
                               int32_t timeout = RR_TIMEOUT_INFINITE);
 
    void AsyncPeekValueBaseEnd1(const RR_INTRUSIVE_PTR<MessageEntry>& m,
                                const RR_SHARED_PTR<RobotRaconteurException>& err,
                                boost::function<void(const RR_INTRUSIVE_PTR<RRValue>&, const TimeSpec&,
                                                     const RR_SHARED_PTR<RobotRaconteurException>&)>& handler);
};
 
template <typename T>
class WireClient : public virtual Wire<T>, public virtual WireClientBase
{
  public:
    WireClient(boost::string_ref name, const RR_SHARED_PTR<ServiceStub>& stub,
               MemberDefinition_Direction direction = MemberDefinition_Direction_both,
               boost::function<void(const RR_INTRUSIVE_PTR<RRValue>&)> verify = RR_NULL_FN)
        : WireClientBase(name, stub, direction), Wire<T>(verify)
    {
        rawelements = (boost::is_same<T, RR_INTRUSIVE_PTR<MessageElement> >::value);
    }
 
    RR_OVIRTUAL ~WireClient() RR_OVERRIDE {}
 
    RR_OVIRTUAL void AsyncConnect(
        boost::function<void(const RR_SHARED_PTR<WireConnection<T> >&, const RR_SHARED_PTR<RobotRaconteurException>&)>
            handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        AsyncConnect_internal(boost::bind(handler,
                                          boost::bind(&WireClient<T>::AsyncConnect_cast, RR_BOOST_PLACEHOLDERS(_1)),
                                          RR_BOOST_PLACEHOLDERS(_2)),
                              timeout);
    }
 
    RR_OVIRTUAL RR_SHARED_PTR<WireConnection<T> > Connect() RR_OVERRIDE
    {
        ROBOTRACONTEUR_ASSERT_MULTITHREADED(node);
 
        RR_SHARED_PTR<detail::sync_async_handler<WireConnection<T> > > t =
            RR_MAKE_SHARED<detail::sync_async_handler<WireConnection<T> > >();
        AsyncConnect(boost::bind(&detail::sync_async_handler<WireConnection<T> >::operator(), t,
                                 RR_BOOST_PLACEHOLDERS(_1), RR_BOOST_PLACEHOLDERS(_2)),
                     GetNode()->GetRequestTimeout());
        return t->end();
    }
 
  protected:
    static RR_SHARED_PTR<WireConnection<T> > AsyncConnect_cast(const RR_SHARED_PTR<WireConnectionBase>& b)
    {
        return rr_cast<WireConnection<T> >(b);
    }
 
    void AsyncPeekValueBaseEnd2(
        const RR_INTRUSIVE_PTR<RRValue>& value, const TimeSpec& ts, const RR_SHARED_PTR<RobotRaconteurException>& err,
        const boost::function<void(const T&, const TimeSpec&, const RR_SHARED_PTR<RobotRaconteurException>&)>& handler)
    {
 
        if (err)
        {
            typename boost::initialized<T> err_value;
            handler(err_value, ts, err);
            return;
        }
 
        T value2;
        try
        {
            value2 = RRPrimUtil<T>::PreUnpack(value);
        }
        catch (std::exception& exp)
        {
            typename boost::initialized<T> err_value;
            RR_SHARED_PTR<RobotRaconteurException> err = RobotRaconteurExceptionUtil::ExceptionToSharedPtr(exp);
            handler(err_value, ts, err);
            return;
        }
 
        handler(value2, ts, err);
    }
 
  public:
    RR_OVIRTUAL T PeekInValue(TimeSpec& ts) RR_OVERRIDE { return RRPrimUtil<T>::PreUnpack(PeekInValueBase(ts)); }
    RR_OVIRTUAL T PeekOutValue(TimeSpec& ts) RR_OVERRIDE { return RRPrimUtil<T>::PreUnpack(PeekOutValueBase(ts)); }
    RR_OVIRTUAL void PokeOutValue(const T& value) RR_OVERRIDE
    {
        return PokeOutValueBase(RRPrimUtil<T>::PrePack(value));
    }
    RR_OVIRTUAL void AsyncPeekInValue(
        boost::function<void(const T&, const TimeSpec&, const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        AsyncPeekInValueBase(boost::bind(&WireClient::AsyncPeekValueBaseEnd2,
                                         RR_DYNAMIC_POINTER_CAST<WireClient>(shared_from_this()),
                                         RR_BOOST_PLACEHOLDERS(_1), RR_BOOST_PLACEHOLDERS(_2),
                                         RR_BOOST_PLACEHOLDERS(_3), RR_MOVE(handler)),
                             timeout);
    }
    RR_OVIRTUAL void AsyncPeekOutValue(
        boost::function<void(const T&, const TimeSpec&, const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        AsyncPeekOutValueBase(boost::bind(&WireClient::AsyncPeekValueBaseEnd2,
                                          RR_DYNAMIC_POINTER_CAST<WireClient>(shared_from_this()),
                                          RR_BOOST_PLACEHOLDERS(_1), RR_BOOST_PLACEHOLDERS(_2),
                                          RR_BOOST_PLACEHOLDERS(_3), RR_MOVE(handler)),
                              timeout);
    }
    RR_OVIRTUAL void AsyncPokeOutValue(const T& value,
                                       boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
                                       int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        AsyncPokeOutValueBase(RRPrimUtil<T>::PrePack(value), RR_MOVE(handler), timeout);
    }
 
    // Unused service-side functions
    RR_OVIRTUAL boost::function<void(const RR_SHARED_PTR<WireConnection<T> >&)> GetWireConnectCallback() RR_OVERRIDE
    {
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, endpoint, service_path, m_MemberName,
                                                "GetWireConnectCallback is not valid for WireClient");
        throw InvalidOperationException("Not valid for client");
    }
    RR_OVIRTUAL void SetWireConnectCallback(boost::function<void(const RR_SHARED_PTR<WireConnection<T> >&)> function)
        RR_OVERRIDE
    {
        RR_UNUSED(function);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, endpoint, service_path, m_MemberName,
                                                "SetWireConnectCallback is not valid for WireClient");
        throw InvalidOperationException("Not valid for client");
    }
    RR_OVIRTUAL boost::function<T(const uint32_t&)> GetPeekInValueCallback() RR_OVERRIDE
    {
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, endpoint, service_path, m_MemberName,
                                                "GetPeekInValueCallback is not valid for WireClient");
        throw InvalidOperationException("Not valid for client");
    }
    RR_OVIRTUAL void SetPeekInValueCallback(boost::function<T(const uint32_t&)> function) RR_OVERRIDE
    {
        RR_UNUSED(function);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, endpoint, service_path, m_MemberName,
                                                "SetPeekInValueCallback is not valid for WireClient");
        throw InvalidOperationException("Not valid for client");
    }
    RR_OVIRTUAL boost::function<T(const uint32_t&)> GetPeekOutValueCallback() RR_OVERRIDE
    {
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, endpoint, service_path, m_MemberName,
                                                "GetPeekOutValueCallback is not valid for WireClient");
        throw InvalidOperationException("Not valid for client");
    }
    RR_OVIRTUAL void SetPeekOutValueCallback(boost::function<T(const uint32_t&)> function) RR_OVERRIDE
    {
        RR_UNUSED(function);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, endpoint, service_path, m_MemberName,
                                                "SetPeekOutValueCallback is not valid for WireClient");
        throw InvalidOperationException("Not valid for client");
    }
    RR_OVIRTUAL boost::function<void(const T&, const TimeSpec&, const uint32_t&)> GetPokeOutValueCallback() RR_OVERRIDE
    {
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, endpoint, service_path, m_MemberName,
                                                "GetPokeOutValueCallback is not valid for WireClient");
        throw InvalidOperationException("Not valid for client");
    }
    RR_OVIRTUAL void SetPokeOutValueCallback(boost::function<void(const T&, const TimeSpec&, const uint32_t&)> function)
        RR_OVERRIDE
    {
        RR_UNUSED(function);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, endpoint, service_path, m_MemberName,
                                                "SetPokeOutValueCallback is not valid for WireClient");
        throw InvalidOperationException("Not valid for client");
    }
 
  protected:
    RR_OVIRTUAL RR_SHARED_PTR<WireConnectionBase> CreateNewWireConnection(MemberDefinition_Direction direction)
        RR_OVERRIDE
    {
        return RR_MAKE_SHARED<WireConnection<T> >(RR_STATIC_POINTER_CAST<WireBase>(shared_from_this()), 0, direction);
    }
};
 
class ROBOTRACONTEUR_CORE_API ServiceSkel;
class ROBOTRACONTEUR_CORE_API WireServerBase : public virtual WireBase
{
    friend class WireConnectionBase;
 
  public:
    RR_OVIRTUAL ~WireServerBase() RR_OVERRIDE {}
 
    RR_OVIRTUAL std::string GetMemberName() RR_OVERRIDE;
 
    RR_OVIRTUAL std::string GetServicePath() RR_OVERRIDE;
 
    RR_OVIRTUAL void WirePacketReceived(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<WireConnectionBase>& 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> WireCommand(const RR_INTRUSIVE_PTR<MessageEntry>& m, uint32_t e);
 
    RR_SHARED_PTR<ServiceSkel> GetSkel();
 
  protected:
    RR_OVIRTUAL void SendWirePacket(const RR_INTRUSIVE_PTR<RRValue>& packet, TimeSpec time,
                                    uint32_t endpoint) RR_OVERRIDE;
 
    std::string m_MemberName;
    std::string service_path;
 
    RR_UNORDERED_MAP<uint32_t, RR_SHARED_PTR<WireConnectionBase> > connections;
    boost::mutex connections_lock;
 
    RR_WEAK_PTR<ServiceSkel> skel;
 
    WireServerBase(boost::string_ref name, const RR_SHARED_PTR<ServiceSkel>& skel,
                   MemberDefinition_Direction direction);
 
    virtual RR_SHARED_PTR<WireConnectionBase> CreateNewWireConnection(uint32_t e,
                                                                      MemberDefinition_Direction direction) = 0;
 
    virtual void fire_WireConnectCallback(const RR_SHARED_PTR<WireConnectionBase>& 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);
 
  protected:
    virtual RR_INTRUSIVE_PTR<RRValue> do_PeekInValue(const uint32_t&) = 0;
    virtual RR_INTRUSIVE_PTR<RRValue> do_PeekOutValue(const uint32_t&) = 0;
    virtual void do_PokeOutValue(const RR_INTRUSIVE_PTR<RRValue>& value, const TimeSpec&, const uint32_t& ep) = 0;
};
 
template <typename T>
class WireServer : public virtual WireServerBase, public virtual Wire<T>
{
 
  public:
    WireServer(boost::string_ref name, const RR_SHARED_PTR<ServiceSkel>& skel,
               MemberDefinition_Direction direction = MemberDefinition_Direction_both,
               boost::function<void(const RR_INTRUSIVE_PTR<RRValue>&)> verify = RR_NULL_FN)
        : WireServerBase(name, skel, direction), Wire<T>(verify)
    {
        rawelements = (boost::is_same<T, RR_INTRUSIVE_PTR<MessageElement> >::value);
    }
 
    RR_OVIRTUAL ~WireServer() RR_OVERRIDE {}
 
    RR_OVIRTUAL void AsyncConnect(
        boost::function<void(const RR_SHARED_PTR<WireConnection<T> >&, const RR_SHARED_PTR<RobotRaconteurException>&)>
            handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        RR_UNUSED(handler);
        RR_UNUSED(timeout);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, -1, service_path, m_MemberName,
                                                "AsyncConnect is not valid for WireServer");
        throw InvalidOperationException("Not valid for server");
    }
    RR_OVIRTUAL RR_SHARED_PTR<WireConnection<T> > Connect() RR_OVERRIDE
    {
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, -1, service_path, m_MemberName,
                                                "Connect is not valid for WireServer");
        throw InvalidOperationException("Not valid for server");
    }
    RR_OVIRTUAL T PeekInValue(TimeSpec& ts) RR_OVERRIDE
    {
        RR_UNUSED(ts);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, -1, service_path, m_MemberName,
                                                "PeekInValue is not valid for WireServer");
        throw InvalidOperationException("Not valid for server");
    }
    RR_OVIRTUAL T PeekOutValue(TimeSpec& ts) RR_OVERRIDE
    {
        RR_UNUSED(ts);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, -1, service_path, m_MemberName,
                                                "PeekOutValue is not valid for WireServer");
        throw InvalidOperationException("Not valid for server");
    }
    RR_OVIRTUAL void PokeOutValue(const T& value) RR_OVERRIDE
    {
        RR_UNUSED(value);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, -1, service_path, m_MemberName,
                                                "PokeOutValue is not valid for WireServer");
        throw InvalidOperationException("Not valid for server");
    }
    RR_OVIRTUAL void AsyncPeekInValue(
        boost::function<void(const T&, const TimeSpec&, const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        RR_UNUSED(handler);
        RR_UNUSED(timeout);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, -1, service_path, m_MemberName,
                                                "AsyncPeekInValue is not valid for WireServer");
        throw InvalidOperationException("Not valid for server");
    }
    RR_OVIRTUAL void AsyncPeekOutValue(
        boost::function<void(const T&, const TimeSpec&, const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        RR_UNUSED(handler);
        RR_UNUSED(timeout);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, -1, service_path, m_MemberName,
                                                "AsyncPeekOutValue is not valid for WireServer");
        throw InvalidOperationException("Not valid for server");
    }
    RR_OVIRTUAL void AsyncPokeOutValue(const T& value,
                                       boost::function<void(const RR_SHARED_PTR<RobotRaconteurException>&)> handler,
                                       int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        RR_UNUSED(value);
        RR_UNUSED(handler);
        RR_UNUSED(timeout);
        ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, -1, service_path, m_MemberName,
                                                "AsyncPokeOutValue is not valid for WireServer");
        throw InvalidOperationException("Not valid for server");
    }
 
    RR_OVIRTUAL boost::function<void(const RR_SHARED_PTR<WireConnection<T> >&)> GetWireConnectCallback() RR_OVERRIDE
    {
        return callback;
    }
    RR_OVIRTUAL void SetWireConnectCallback(boost::function<void(const RR_SHARED_PTR<WireConnection<T> >&)> function)
        RR_OVERRIDE
    {
        callback = function;
    }
    RR_OVIRTUAL boost::function<T(const uint32_t&)> GetPeekInValueCallback() RR_OVERRIDE { return peek_in_callback; }
    RR_OVIRTUAL void SetPeekInValueCallback(boost::function<T(const uint32_t&)> function) RR_OVERRIDE
    {
        peek_in_callback = function;
    }
    RR_OVIRTUAL boost::function<T(const uint32_t&)> GetPeekOutValueCallback() RR_OVERRIDE { return peek_out_callback; }
    RR_OVIRTUAL void SetPeekOutValueCallback(boost::function<T(const uint32_t&)> function) RR_OVERRIDE
    {
        peek_out_callback = function;
    }
    RR_OVIRTUAL boost::function<void(const T&, const TimeSpec&, const uint32_t&)> GetPokeOutValueCallback() RR_OVERRIDE
    {
        return poke_out_callback;
    }
    RR_OVIRTUAL void SetPokeOutValueCallback(boost::function<void(const T&, const TimeSpec&, const uint32_t&)> function)
        RR_OVERRIDE
    {
        poke_out_callback = function;
    }
 
  protected:
    RR_OVIRTUAL RR_SHARED_PTR<WireConnectionBase> CreateNewWireConnection(
        uint32_t e, MemberDefinition_Direction direction) RR_OVERRIDE
    {
        return RR_MAKE_SHARED<WireConnection<T> >(RR_STATIC_POINTER_CAST<WireBase>(shared_from_this()), e, direction);
    }
 
    boost::function<void(RR_SHARED_PTR<WireConnection<T> >)> callback;
    boost::function<T(const uint32_t&)> peek_in_callback;
    boost::function<T(const uint32_t&)> peek_out_callback;
    boost::function<void(const T&, const TimeSpec&, const uint32_t&)> poke_out_callback;
 
    RR_OVIRTUAL void fire_WireConnectCallback(const RR_SHARED_PTR<WireConnectionBase>& e) RR_OVERRIDE
    {
        if (!callback)
            return;
        callback(RR_STATIC_POINTER_CAST<WireConnection<T> >(e));
    }
 
    RR_OVIRTUAL RR_INTRUSIVE_PTR<RRValue> do_PeekInValue(const uint32_t& ep) RR_OVERRIDE
    {
        if (!peek_in_callback)
        {
            ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, ep, service_path, m_MemberName,
                                                    "Attempt to call PeekInValue when callback not set");
            throw InvalidOperationException("Invalid operation");
        }
        return RRPrimUtil<T>::PrePack(peek_in_callback(ep));
    }
 
    RR_OVIRTUAL RR_INTRUSIVE_PTR<RRValue> do_PeekOutValue(const uint32_t& ep) RR_OVERRIDE
    {
        if (!peek_out_callback)
        {
            ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, ep, service_path, m_MemberName,
                                                    "Attempt to call PeekOutValue when callback not set");
            throw InvalidOperationException("Invalid operation");
        }
        return RRPrimUtil<T>::PrePack(peek_out_callback(ep));
    }
 
    RR_OVIRTUAL void do_PokeOutValue(const RR_INTRUSIVE_PTR<RRValue>& value, const TimeSpec& ts,
                                     const uint32_t& ep) RR_OVERRIDE
    {
        if (!poke_out_callback)
        {
            ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, ep, service_path, m_MemberName,
                                                    "Attempt to call PokeOutValue when callback not set");
            throw InvalidOperationException("Invalid operation");
        }
        return poke_out_callback(RRPrimUtil<T>::PreUnpack(value), ts, ep);
    }
 
  public:
    RR_OVIRTUAL void Shutdown() RR_OVERRIDE
    {
        WireServerBase::Shutdown();
        callback.clear();
        peek_in_callback.clear();
        peek_out_callback.clear();
        poke_out_callback.clear();
    }
};
 
namespace detail
{
template <typename T>
class Wire_traits;
 
template <typename T>
class Wire_traits<Wire<T> >
{
  public:
    typedef WireConnection<T> wireconnection_type;
    typedef WireClient<T> wireclient_type;
    typedef WireServer<T> wireserver_type;
};
 
} // namespace detail
 
namespace detail
{
class WireBroadcaster_connected_connection;
}
 
class ROBOTRACONTEUR_CORE_API WireBroadcasterBase : public RR_ENABLE_SHARED_FROM_THIS<WireBroadcasterBase>
{
  public:
    size_t GetActiveWireConnectionCount();
 
    virtual ~WireBroadcasterBase();
 
    boost::function<bool(RR_SHARED_PTR<WireBroadcasterBase>&, uint32_t)> GetPredicate();
 
    void SetPredicate(boost::function<bool(const RR_SHARED_PTR<WireBroadcasterBase>&, uint32_t)> f);
 
    int32_t GetOutValueLifespan();
 
    void SetOutValueLifespan(int32_t millis);
 
  protected:
    WireBroadcasterBase();
 
    void InitBase(const RR_SHARED_PTR<WireBase>& wire);
 
    void ConnectionClosedBase(const RR_SHARED_PTR<detail::WireBroadcaster_connected_connection>& ep);
 
    void ConnectionConnectedBase(const RR_SHARED_PTR<WireConnectionBase>& ep);
 
    void SetOutValueBase(const RR_INTRUSIVE_PTR<RRValue>& value);
 
    virtual void AttachWireServerEvents(const RR_SHARED_PTR<WireServerBase>& w);
 
    virtual void AttachWireConnectionEvents(const RR_SHARED_PTR<WireConnectionBase>& w,
                                            const RR_SHARED_PTR<detail::WireBroadcaster_connected_connection>& cep);
 
    RR_INTRUSIVE_PTR<RRValue> ClientPeekInValueBase();
 
    std::list<RR_SHARED_PTR<detail::WireBroadcaster_connected_connection> > connected_wires;
    boost::mutex connected_wires_lock;
    RR_WEAK_PTR<WireServerBase> wire;
    RR_WEAK_PTR<RobotRaconteurNode> node;
    std::string service_path;
    std::string member_name;
 
    bool copy_element;
 
    boost::function<bool(RR_SHARED_PTR<WireBroadcasterBase>&, uint32_t)> predicate;
 
    RR_INTRUSIVE_PTR<RRValue> out_value;
    boost::initialized<bool> out_value_valid;
 
    int32_t out_value_lifespan;
    boost::posix_time::ptime out_value_lasttime_local;
 
    void ServiceEvent(ServerServiceListenerEventType evt);
 
    RR_SHARED_PTR<WireBase> GetWireBase();
};
 
template <typename T>
class WireBroadcaster : public WireBroadcasterBase
{
  public:
    WireBroadcaster() {}
 
    void Init(RR_SHARED_PTR<Wire<T> > wire) { InitBase(wire); }
 
    void SetOutValue(T value) { SetOutValueBase(RRPrimUtil<T>::PrePack(value)); }
 
    RR_SHARED_PTR<Wire<T> > GetWire() { return rr_cast<Wire<T> >(GetWireBase()); }
 
  protected:
    T ClientPeekInValue() { return RRPrimUtil<T>::PreUnpack(ClientPeekInValueBase()); }
 
    static T ClientPeekOutValue() { throw ReadOnlyMemberException("Read only wire"); }
    static T ClientPokeOutValue() { throw ReadOnlyMemberException("Read only wire"); }
 
    RR_OVIRTUAL void AttachWireServerEvents(const RR_SHARED_PTR<WireServerBase>& w) RR_OVERRIDE
    {
        RR_SHARED_PTR<WireServer<T> > w_T = rr_cast<WireServer<T> >(w);
        w_T->SetWireConnectCallback(boost::bind(&WireBroadcaster::ConnectionConnectedBase, this->shared_from_this(),
                                                RR_BOOST_PLACEHOLDERS(_1)));
        w_T->SetPeekInValueCallback(boost::bind(&WireBroadcaster<T>::ClientPeekInValue,
                                                RR_STATIC_POINTER_CAST<WireBroadcaster<T> >(this->shared_from_this())));
        w_T->SetPeekOutValueCallback(boost::bind(&WireBroadcaster<T>::ClientPeekOutValue));
        w_T->SetPokeOutValueCallback(boost::bind(&WireBroadcaster<T>::ClientPokeOutValue));
    }
 
    RR_OVIRTUAL void AttachWireConnectionEvents(const RR_SHARED_PTR<WireConnectionBase>& w,
                                                const RR_SHARED_PTR<detail::WireBroadcaster_connected_connection>& c)
        RR_OVERRIDE
    {
        RR_SHARED_PTR<WireConnection<T> > w_T = rr_cast<WireConnection<T> >(w);
        w_T->SetWireConnectionClosedCallback(
            boost::bind(&WireBroadcaster::ConnectionClosedBase, this->shared_from_this(), c));
    }
};
 
namespace detail
{
static void WireUnicastReceiverBase_empty_close_handler(const RR_SHARED_PTR<RobotRaconteurException>& err) {}
} // namespace detail
 
template <typename T, typename U>
class WireUnicastReceiverBase : public RR_ENABLE_SHARED_FROM_THIS<WireUnicastReceiverBase<T, U> >
{
  public:
    typedef typename detail::Wire_traits<T>::wireserver_type wireserver_type;
    typedef typename detail::Wire_traits<T>::wireconnection_type wireconnection_type;
 
    WireUnicastReceiverBase() : in_value_lifespan(-1) {}
    virtual ~WireUnicastReceiverBase() {}
 
    void Init(const RR_SHARED_PTR<T>& wire)
    {
        node = wire->GetNode();
        in_value_lifespan = -1;
        RR_SHARED_PTR<wireserver_type> wire_server = RR_DYNAMIC_POINTER_CAST<wireserver_type>(wire);
        if (!wire_server)
        {
            ROBOTRACONTEUR_LOG_DEBUG_COMPONENT_PATH(node, Member, -1, service_path, member_name,
                                                    "WireUnicastReceiver init must be passed a WireServer");
            throw InvalidOperationException("WireServer required for WireUnicastReceiver");
        }
        this->wire = wire_server;
        wire_server->SetWireConnectCallback(boost::bind(&WireUnicastReceiverBase<T, U>::ConnectionConnected,
                                                        this->shared_from_this(), RR_BOOST_PLACEHOLDERS(_1)));
        wire_server->SetPeekInValueCallback(boost::bind(&WireUnicastReceiverBase<T, U>::ClientPeekInValue));
        wire_server->SetPeekOutValueCallback(
            boost::bind(&WireUnicastReceiverBase<T, U>::ClientPeekOutValue, this->shared_from_this()));
        wire_server->SetPokeOutValueCallback(boost::bind(&WireUnicastReceiverBase<T, U>::ClientPokeOutValue,
                                                         this->shared_from_this(), RR_BOOST_PLACEHOLDERS(_1),
                                                         RR_BOOST_PLACEHOLDERS(_2), RR_BOOST_PLACEHOLDERS(_3)));
 
        wire_server->GetSkel()->GetContext()->ServerServiceListener.connect(
            boost::signals2::signal<void(
                const RR_SHARED_PTR<ServerContext>&, ServerServiceListenerEventType,
                const RR_SHARED_PTR<void>&)>::slot_type(boost::bind(&WireUnicastReceiverBase::ServiceEvent, this,
                                                                    RR_BOOST_PLACEHOLDERS(_2)))
                .track(this->shared_from_this()));
 
        this->service_path = wire_server->GetServicePath();
        this->member_name = wire_server->GetMemberName();
 
        ROBOTRACONTEUR_LOG_TRACE_COMPONENT_PATH(node, Member, -1, service_path, member_name,
                                                "WireUnicastReceiver initialized");
    }
 
    U GetInValue(TimeSpec& ts, uint32_t& ep)
    {
        boost::mutex::scoped_lock lock(this_lock);
        if (!in_value_valid.data())
            throw ValueNotSetException("Value not set");
        if (detail::WireConnectionBase_IsValueExpired(node, in_value_lasttime_local, in_value_lifespan))
        {
            throw ValueNotSetException("Value expired");
        }
        ts = in_value_ts;
        ep = in_value_ep;
        return in_value;
    }
 
    bool TryGetInValue(U& value, TimeSpec& ts, uint32_t& ep)
    {
        boost::mutex::scoped_lock lock(this_lock);
        if (!in_value_valid)
            return false;
        if (detail::WireConnectionBase_IsValueExpired(node, in_value_lasttime_local, in_value_lifespan))
        {
            return false;
        }
        value = in_value;
        ts = in_value_ts;
        ep = in_value_ep;
        return true;
    }
 
    boost::signals2::signal<void(const U&, const TimeSpec&, const uint32_t&)> InValueChanged;
 
    int32_t GetInValueLifespan()
    {
        boost::mutex::scoped_lock lock(this_lock);
        return in_value_lifespan;
    }
 
    void SetInValueLifespan(int32_t millis)
    {
        boost::mutex::scoped_lock lock(this_lock);
        in_value_lifespan = millis;
    }
 
    RR_SHARED_PTR<T> GetWire() { return wire; }
 
  protected:
    void ConnectionConnected(const RR_SHARED_PTR<wireconnection_type>& connection)
    {
        boost::mutex::scoped_lock lock(this_lock);
        if (active_connection)
        {
            uint32_t active_ep = active_connection->GetEndpoint();
            try
            {
                active_connection->AsyncClose(&detail::WireUnicastReceiverBase_empty_close_handler);
            }
            catch (std::exception&)
            {}
            active_connection.reset();
 
            ROBOTRACONTEUR_LOG_TRACE_COMPONENT_PATH(node, Member, active_ep, service_path, member_name,
                                                    "WireUnicastReceiver active wire closed for new connection");
        }
        active_connection = connection;
        connection->SetWireConnectionClosedCallback(boost::bind(&WireUnicastReceiverBase<T, U>::ConnectionClosed,
                                                                this->shared_from_this(), RR_BOOST_PLACEHOLDERS(_1)));
        connection->WireValueChanged.connect(boost::bind(&WireUnicastReceiverBase<T, U>::ConnectionInValueChanged,
                                                         this->shared_from_this(), RR_BOOST_PLACEHOLDERS(_1),
                                                         RR_BOOST_PLACEHOLDERS(_2), RR_BOOST_PLACEHOLDERS(_3)));
 
        ROBOTRACONTEUR_LOG_TRACE_COMPONENT_PATH(node, Member, -1, service_path, member_name,
                                                "WireUnicastReceiver wire connected, made active wire");
    }
 
    void ConnectionClosed(const RR_SHARED_PTR<wireconnection_type>& connection)
    {
        boost::mutex::scoped_lock lock(this_lock);
        if (active_connection == connection)
        {
            active_connection.reset();
        }
    }
 
    void ConnectionInValueChanged(const RR_SHARED_PTR<wireconnection_type>& connection, const U& value,
                                  const TimeSpec& time)
    {
        ClientPokeOutValue(value, time, connection->GetEndpoint());
    }
 
    static U ClientPeekInValue() { throw WriteOnlyMemberException("Write only wire"); }
 
    U ClientPeekOutValue()
    {
        boost::mutex::scoped_lock lock(this_lock);
        if (!in_value_valid)
            throw ValueNotSetException("Value not set");
        return in_value;
    }
 
    void ClientPokeOutValue(const U& value, const TimeSpec& ts, const uint32_t& ep)
    {
        RR_SHARED_PTR<RobotRaconteurNode> n = node.lock();
        boost::mutex::scoped_lock lock(this_lock);
        in_value = value;
        in_value_ts = ts;
        in_value_valid.data() = true;
        in_value_ep.data() = ep;
        if (n)
        {
            in_value_lasttime_local = n->NowNodeTime();
        }
 
        lock.unlock();
 
        InValueChanged(value, ts, ep);
 
        ROBOTRACONTEUR_LOG_TRACE_COMPONENT_PATH(node, Member, ep, service_path, member_name,
                                                "WireUnicastReceiver value changed");
    }
 
    void ServiceEvent(ServerServiceListenerEventType evt)
    {
        if (evt != ServerServiceListenerEventType_ServiceClosed)
            return;
        boost::mutex::scoped_lock lock(this_lock);
        InValueChanged.disconnect_all_slots();
    }
 
    RR_SHARED_PTR<wireserver_type> wire;
    RR_SHARED_PTR<wireconnection_type> active_connection;
    boost::mutex this_lock;
    U in_value;
    TimeSpec in_value_ts;
    boost::initialized<bool> in_value_valid;
    boost::initialized<uint32_t> in_value_ep;
    boost::posix_time::ptime in_value_lasttime_local;
    int32_t in_value_lifespan;
 
    std::string member_name;
    std::string service_path;
    RR_WEAK_PTR<RobotRaconteurNode> node;
};
 
template <typename T>
class WireUnicastReceiver : public WireUnicastReceiverBase<Wire<T>, T>
{
};
#ifndef ROBOTRACONTEUR_NO_CXX11_TEMPLATE_ALIASES
using WireConnectionBasePtr = RR_SHARED_PTR<WireConnectionBase>;
template <typename T>
using WireConnectionPtr = RR_SHARED_PTR<WireConnection<T> >;
using WireBasePtr = RR_SHARED_PTR<WireBase>;
template <typename T>
using WirePtr = RR_SHARED_PTR<Wire<T> >;
template <typename T>
using WireBroadcasterPtr = RR_SHARED_PTR<WireBroadcaster<T> >;
template <typename T>
using WireUnicastReceiverPtr = RR_SHARED_PTR<WireUnicastReceiver<T> >;
#endif
 
} // namespace RobotRaconteur
 
#pragma warning(pop)