public void RemoveSocket(NetMQSocket socket)
        {
            socket.EventsChanged -= OnSocketEventsChanged;

            m_sockets.Remove(socket);
            m_isDirty = true;
        }

 internal Socket(NetMQSocket socket, SocketConfiguration config)
 {
     socket.Options.Linger = config.Linger;
     socket.Options.ReceiveHighWatermark = config.ReceivingHighWatermark;
     socket.Options.SendHighWatermark = config.SendingHighWatermark;
     this.socket = socket;
 }

 public Transport(NetMQSocket value) : this()
 {
     if (value != null)
     {
         this.socket = value;
     }
 }

        /// <summary>
        /// Create a new instance of the <see cref="DeviceBase"/> class.
        /// </summary>
        /// <param name="poller">the <see cref="INetMQPoller"/> to use for detecting when messages are available</param>
        /// <param name="frontendSocket">
        /// A <see cref="NetMQSocket"/> that will pass incoming messages to <paramref name="backendSocket"/>.
        /// </param>
        /// <param name="backendSocket">
        /// A <see cref="NetMQSocket"/> that will receive messages from (and optionally send replies to) <paramref name="frontendSocket"/>.
        /// </param>
        /// <param name="mode">the <see cref="DeviceMode"/> (either Blocking or Threaded) for this device</param>
        /// <exception cref="ArgumentNullException">frontendSocket must not be null.</exception>
        /// <exception cref="ArgumentNullException">backendSocket must not be null.</exception>
        protected DeviceBase(INetMQPoller poller,  NetMQSocket frontendSocket,  NetMQSocket backendSocket, DeviceMode mode)
        {
            m_isInitialized = false;

            if (frontendSocket == null)
                throw new ArgumentNullException("frontendSocket");

            if (backendSocket == null)
                throw new ArgumentNullException("backendSocket");

            FrontendSocket = frontendSocket;
            BackendSocket = backendSocket;

            FrontendSetup = new DeviceSocketSetup(FrontendSocket);
            BackendSetup = new DeviceSocketSetup(BackendSocket);

            m_poller = poller;

            FrontendSocket.ReceiveReady += FrontendHandler;
            BackendSocket.ReceiveReady += BackendHandler;

            m_poller.Add(FrontendSocket);
            m_poller.Add(BackendSocket);

            m_runner = mode == DeviceMode.Blocking
                ? new DeviceRunner(this)
                : new ThreadedDeviceRunner(this);
        }

        public void ConnectOrBindAddress(NetMQSocket socket)
        {
            var port = socket.BindRandomPort(Uri.AbsoluteUri.TrimEnd('/'));
            var address = $"{Uri.Scheme}://{Uri.Host}:{port}";

            Uri = new Uri(address);
        }

        internal WebSocketClient(NetMQSocket streamSocket, byte[] identity)
        {
            m_state = WebSocketClientState.Closed;
            m_streamSocket = streamSocket;
            m_outgoingMessage = null;

            Identity = identity;
        }

 /// <summary>
 /// Create a new instance of a Proxy (NetMQ.Proxy)
 /// with the given sockets to serve as a front-end, a back-end, and a control socket.
 /// </summary>
 /// <param name="frontend">the socket that messages will be forwarded from</param>
 /// <param name="backend">the socket that messages will be forwarded to</param>
 /// <param name="control">this socket will have messages also sent to it - you can set this to null if not needed</param>
 /// <param name="poller">an optional external poller to use within this proxy</param>
 public Proxy([NotNull] NetMQSocket frontend, [NotNull] NetMQSocket backend, [CanBeNull] NetMQSocket control = null, Poller poller = null)
 {
     m_frontend = frontend;
     m_backend = backend;
     m_control = control;
     m_externalPoller = poller != null;
     m_poller = poller;
 }

 protected override void DoServer(NetMQSocket socket, int messageSize)
 {
     for (int i = 0; i < Iterations; i++)
     {
         byte[] message = socket.Receive();
         socket.Send(message);
     }
 }

        public static bool Send(Message message, NetMQSocket socket)
        {
            Send(message.UUID, socket);
            Send(Delimeter, socket);
            Send(string.Empty, socket);
            Send(JsonSerializer.Serialize(message.Header), socket);
            Send(JsonSerializer.Serialize(message.ParentHeader), socket);
            Send(JsonSerializer.Serialize(message.MetaData), socket);
            Send(message.Content, socket, false);

            return true;
        }

        public Action<NetMQMessage> Send(NetMQSocket socket, PendingResRequest pendingRequest, Guid requestId)
        {
            var pending = (PendingResRequest<QueryEventsForStreamResponse>) pendingRequest;
            var msg = new NetMQMessage();
            msg.AppendEmptyFrame();
            msg.Append(ResProtocol.ResClient01);
            msg.Append(ResCommands.QueryEventsByStream);
            msg.Append(requestId.ToByteArray());
            msg.Append(_context);
            msg.Append(_stream);
            msg.Append(_fromVersion.ToNetMqFrame());
            msg.Append(_maxVersion.ToNetMqFrame());

            socket.SendMultipartMessage(msg);

            return m =>
            {
                var command = m.Pop().ConvertToString();

                if (command == ResCommands.Error)
                {
                    var errorCode = m.Pop().ConvertToString();
                    var errorDetails = m.Pop().ConvertToString();
                    ErrorResolver.RaiseException(errorCode, errorDetails, pending.SetException);
                    return;
                }

                if (command != ResCommands.QueryEventsByStreamResponse)
                    pending.SetException(new UnsupportedCommandException(command));

                var count = m.PopInt32();

                var events = new EventInStorage[count];

                for (var i = 0; i < count; i++)
                {
                    var id = new Guid(m.Pop().ToByteArray());
                    var streamId = m.Pop().ConvertToString();
                    var context = m.Pop().ConvertToString();
                    var sequence = m.PopInt64();
                    var timestamp = m.PopDateTime();;
                    var type = m.PopString();
                    var headers = m.PopStringOrNull();
                    var body = m.PopString();

                    events[i] = new EventInStorage(context, streamId, sequence, type, id, headers, body, timestamp);
                }

                var result = new QueryEventsForStreamResponse(_context, _stream, events);
                pending.SetResult(result);
            };
        }

        public void AddSocket(NetMQSocket socket)
        {
            if (m_sockets.Contains(socket))
            {
                throw new ArgumentException("Socket already added to poller");
            }

            m_sockets.Add(socket);

            socket.EventsChanged += OnSocketEventsChanged;

            m_isDirty = true;
        }

        /// <summary>
        /// Create a new instance of a Proxy (NetMQ.Proxy)
        /// with the given sockets to serve as a front-end, a back-end, and a control socket.
        /// </summary>
        /// <param name="frontend">the socket that messages will be forwarded from</param>
        /// <param name="backend">the socket that messages will be forwarded to</param>
        /// <param name="controlIn">this socket will have incoming messages also sent to it - you can set this to null if not needed</param>
        /// <param name="controlOut">this socket will have outgoing messages also sent to it - you can set this to null if not needed</param>
        /// <param name="poller">an optional external poller to use within this proxy</param>
        public Proxy([NotNull] NetMQSocket frontend, [NotNull] NetMQSocket backend, [CanBeNull] NetMQSocket controlIn, [CanBeNull] NetMQSocket controlOut, [CanBeNull] INetMQPoller poller = null)
        {
            if (poller != null)
            {
                m_externalPoller = true;
                m_poller = poller;
            }

            m_frontend = frontend;
            m_backend = backend;
            m_controlIn = controlIn;
            m_controlOut = controlOut ?? controlIn;
        }

        public void ProcessMessage(NetMQMessage message, NetMQSocket socket)
        {
            if (message.FrameCount < 3)
                throw new MalformedMessageReceivedException(message.FrameCount);

            var sender = message.PopUntilEmptyFrame();
            var protocolFrame = message.Pop();
            var protocol = protocolFrame.ConvertToString();
            ensureProtocol(protocol);
            var command = message.Pop().ConvertToString();

            _dispatcher.Dispatch(command, sender, message);
        }

        private NetMQMonitor CreateMonitor(NetMQContext context, NetMQSocket socket, Poller poller)
        {
            var monitor = new NetMQMonitor(context, socket, $"inproc://{Guid.NewGuid()}.inproc",
                SocketEvents.Connected | SocketEvents.Disconnected | SocketEvents.ConnectRetried);

            monitor.Connected += Monitor_Connected;
            monitor.Disconnected += Monitor_Disconnected;
            monitor.ConnectRetried += Monitor_ConnectRetried;

            monitor.AttachToPoller(poller);

            return monitor;
        }

 /// <summary>
 ///     setup the client with standard values
 ///     verbose == false
 ///     timeout == 2500
 ///     reties  == 3
 /// </summary>
 private MDPClient()
 {
     m_ctx = NetMQContext.Create ();
     m_client = null;
     Timeout = TimeSpan.FromMilliseconds (2500);
     Retries = 3;
     m_connected = false;
 }

 internal NmqMessageSender(Uri serviceUri)
 {
     context = NetMQContext.Create();
     socket = context.CreateRequestSocket();
     var address = string.Format("tcp://{0}:{1}", serviceUri.Host, serviceUri.Port);
     socket.Connect(address);
 }

 /// <summary>
 /// Initializes a new instance of the <see cref="Logit.Zmq.OutputWriter"/> class.
 /// </summary>
 /// <param name="zt">ØMQ Transport to use.</param>
 public OutputWriter(Transport zt) : base(CultureInfo.InvariantCulture)
 {
     if (zt != null)
     {
         this.socket = zt.Socket;
     }
 }

        private void SetupPublisher()
        {
            Open(0);

            _publisherSocket = _context.CreatePublisherSocket();
            _publisherSocket.Bind(_configuration.PublisherAddress);
        }

 public FeedZmQPublisher(string address, ILog log)
 {
     _log = log;
     _context = NetMQContext.Create();
     _socket = _context.CreatePushSocket();
     _socket.Bind(address);
 }

		public async Task Start ()
		{
			ThrowIfDisposed ();
			var ct = cancellationTokenSource.Token;

			nmqPoller = new Poller ();
			nmqScheduler = new NetMQScheduler (nmqContext, nmqPoller);
			nmqServer = nmqContext.CreateResponseSocket ();
			nmqServer.Bind (Address.AbsoluteUri.TrimEnd ('/'));

			serverTask = Task.Factory.StartNew (() => {
				ct.ThrowIfCancellationRequested ();

				while (true) {
					if (ct.IsCancellationRequested) {
						// clean up here
						ct.ThrowIfCancellationRequested ();
					}
					var msg = nmqServer.Receive ();
					var request = Request.Deserialize (msg);
					var result = Handle (request);

					byte[] output_buffer = result.Serialize ();
					nmqServer.Send (output_buffer);
				}
			}, ct);

			await serverTask;
		}