public virtual NetMQMessage ToNetMQMessage()
        {
            NetMQMessage message = new NetMQMessage();
            message.Append(new byte[] { (byte)HandshakeType });

            return message;
        }

        public NetMQMessage Send (string serviceName, NetMQMessage request)
        {
            // first call is [mmi.service][servicename]
            // return is [Ok]
            // second call is [service][request]
            // return is [reply]
            m_count++;

            if (m_count == 1)
            {
                // proceed only if commanded to -> is automatically called by TitanicBroker.Run
                // and askes for a reply from a service, so wait until we want an answer
                waitHandle.WaitOne ();

                var reply = new NetMQMessage ();
                reply.Push (MmiCode.Ok.ToString ());

                return reply;
            }

            // wait to proceed until signaled
            waitHandle.WaitOne ();

            return request; // as echo service :-)
        }

        public void RequestResponseMultipartMessageWithRetryFails()
        {
            const string address = "tcp://127.0.0.1:50002";
            const string pubAddress = "tcp://127.0.0.1:60002";
            const int numTries = 5;
            var requestTimeout = TimeSpan.FromMilliseconds(100);
            var requestMessage = new NetMQMessage(1);
            requestMessage.Append("Hi");

            using (var progressPublisher = new PublisherSocket(pubAddress))
            using (var progressSubscriber = new SubscriberSocket(pubAddress))
            using (var server = new RouterSocket(address))
            {
                progressSubscriber.SubscribeToAnyTopic();
                var progressProactor = new NetMQProactor(progressSubscriber, (socket, message) =>
                    Console.WriteLine("C: {0} {1:ss.fff}", message[0].ConvertToString(), DateTime.Now));

                var serverProactor = new NetMQProactor(server, (socket, message) =>
                {
                    Console.WriteLine("ResponseEcho recieved message {0} at {1:ss.fff}", message[2].ConvertToString(),
                        DateTime.Now);
                });

                using (serverProactor)
                using (progressProactor)
                {
                    var responseMessage = RequestSocket.RequestResponseMultipartMessageWithRetry(address, requestMessage,
                        numTries, requestTimeout, progressPublisher);
                    Assert.IsNull(responseMessage);
                }
            }
        }

		public Task Send(ArraySegment<byte> data, params object[] connectionIDs)
		{
			var task = new Task(() =>
			{
				var msg = new NetMQMessage();
				if (_socket is RouterSocket)
				{
					msg.Append(new byte[0]);
					msg.AppendEmptyFrame();
				}
				msg.Append(data.Count == data.Array.Length ? data.Array : data.ToArray());

				if (connectionIDs.Length <= 0)
					_socket.SendMultipartMessage(msg);
				else
				{
					foreach (var connection in connectionIDs)
					{
						if (_socket is RouterSocket && connection is byte[])
						{
							msg.Pop();
							msg.Push(((byte[])connection));
						}
						_socket.SendMultipartMessage(msg);
					}
				}
			});
			task.Start(_scheduler);
			return task;
		}

        public void SaveMessage_ReplyMessage_SholdCreateNewFileWithGuidAsName ()
        {
            var sut = new TitanicFileIO (Path.GetTempPath ());

            var message = new NetMQMessage ();
            message.Push ("Hello World");
            message.Push ("echo");

            var messageSize = message[0].BufferSize + message[1].BufferSize + 4;    // 2 lines with \r\n

            var id = Guid.NewGuid ();

            sut.SaveMessage (TitanicOperation.Reply, id, message);

            var expectedDir = sut.TitanicDirectory;
            var expectedFile = Path.Combine (expectedDir, id + _reply_ending);

            File.Exists (expectedFile).Should ().BeTrue ("because the file exists");

            var info = new FileInfo (expectedFile);

            info.Length.Should ().Be (messageSize);

            File.Delete (expectedFile);
        }

        /// <summary>
        /// Try to send request message and return the response as a message, or return null if not successful
        /// </summary>
        /// <param name="address">a string denoting the address to connect to</param>
        /// <param name="requestMessage">The request message</param>
        /// <param name="numTries">The number of times to try</param>
        /// <param name="requestTimeout">The timeout for each request</param>
        /// <param name="progressPublisher">Report topics: Failure, Retry, Send, Success</param>
        /// <returns>the response message, or null if not successful</returns>
        public static NetMQMessage RequestResponseMultipartMessageWithRetry([NotNull] string address, [NotNull] NetMQMessage requestMessage,
            int numTries, TimeSpan requestTimeout, PublisherSocket progressPublisher = null)
        {
            var responseMessage = new NetMQMessage();

            while (numTries-- > 0)
            {
                using (var requestSocket = new RequestSocket(address))
                {
                    progressPublisher?.SendFrame(ProgressTopic.Send.ToString());

                    requestSocket.SendMultipartMessage(requestMessage);

                    if (requestSocket.TryReceiveMultipartMessage(requestTimeout, ref responseMessage))
                    {
                        progressPublisher?.SendFrame(ProgressTopic.Success.ToString());

                        return responseMessage;
                    }

                    progressPublisher?.SendFrame(ProgressTopic.Retry.ToString());
                }
            }

            progressPublisher?.SendFrame(ProgressTopic.Failure.ToString());

            return null;
        }

 public static NetMQMessage CreateMessage(string topic, byte[] data)
 {
     var message = new NetMQMessage();
     if (!string.IsNullOrEmpty(topic)) message.Append(Encoding.Unicode.GetBytes(topic));
     message.Append(data);
     return message;
 }

        public void DesktopClient_WillAccept_SinglePartMessages()
        {
            Queue<Packet> packets = new Queue<Packet>();
            // Mock the the message handler processor
            var messageHandler = new Mock<IDesktopMessageHandler>();
            messageHandler.Setup(mh => mh.DisplayLogPacket(It.IsAny<LogPacket>())).Callback(
                (LogPacket packet) =>
                {
                    packets.Enqueue(packet);
                }).Verifiable();

            // Setup Client
            StartClientThread(messageHandler.Object);

            using (PushSocket server = new PushSocket("@tcp://*:" + _port))
            {
                var message = new NetMQMessage();

                message.Append(JsonConvert.SerializeObject(new LogPacket()));

                server.SendMultipartMessage(message);
            }

            // Give NetMQ time to send the message
            Thread.Sleep(500);

            Assert.IsTrue(packets.Count == 1);
        }

 protected static void EncodeVector01(Vector3 val, NetMQMessage msg)
 {
     // TODO: Maybe optimize this to pack into a single integer? Probably not really necessary
     EncodeFloat01(val.x, msg);
     EncodeFloat01(val.y, msg);
     EncodeFloat01(val.z, msg);
 }

        public NetMQMessage DecryptApplicationMessage(NetMQMessage cipherMessage)
        {
            if (!SecureChannelReady)
            {
                throw new NetMQSecurityException(NetMQSecurityErrorCode.SecureChannelNotReady, "Cannot decrypt messages until the secure channel is ready");
            }

            if (cipherMessage == null)
            {
                throw new ArgumentNullException("cipherMessage is null");
            }

            if (cipherMessage.FrameCount < 2)
            {
                throw new NetMQSecurityException(NetMQSecurityErrorCode.InvalidFramesCount, "cipher message should have at least 2 frames");
            }

            NetMQFrame protocolVersionFrame = cipherMessage.Pop();
            NetMQFrame contentTypeFrame = cipherMessage.Pop();

            if (!protocolVersionFrame.ToByteArray().SequenceEqual(m_protocolVersion))
            {
                throw new NetMQSecurityException(NetMQSecurityErrorCode.InvalidProtocolVersion, "Wrong protocol version");
            }

            ContentType contentType = (ContentType)contentTypeFrame.Buffer[0];

            if (contentType != ContentType.ApplicationData)
            {
                throw new NetMQSecurityException(NetMQSecurityErrorCode.InvalidContentType, "Not an applicagtion data message");
            }

            return m_recordLayer.DecryptMessage(ContentType.ApplicationData, cipherMessage);
        }

        /// <summary>
        /// Remove the three frames from the given NetMQMessage, interpreting them thusly:
        /// 1. a byte with the HandshakeType, presumed here to be ClientHello,
        /// 2. a byte-array containing the RandomNumber,
        /// 3. a byte-array with the list of CipherSuites.
        /// </summary>
        /// <param name="message">a NetMQMessage - which must have 2 frames</param>
        /// <exception cref="NetMQSecurityException"><see cref="NetMQSecurityErrorCode.InvalidFramesCount"/>: FrameCount must be 3.</exception>
        public override void SetFromNetMQMessage(NetMQMessage message)
        {
            base.SetFromNetMQMessage(message);

            if (message.FrameCount != 3)
            {
                throw new NetMQSecurityException(NetMQSecurityErrorCode.InvalidFramesCount, "Malformed message");
            }

            // get the random number
            NetMQFrame randomNumberFrame = message.Pop();
            RandomNumber = randomNumberFrame.ToByteArray();

            // get the length of the cipher-suites array
            NetMQFrame ciphersLengthFrame = message.Pop();
            int ciphersLength = BitConverter.ToInt32(ciphersLengthFrame.Buffer, 0);

            // get the cipher-suites
            NetMQFrame ciphersFrame = message.Pop();
            CipherSuites = new CipherSuite[ciphersLength];
            for (int i = 0; i < ciphersLength; i++)
            {
                CipherSuites[i] = (CipherSuite)ciphersFrame.Buffer[i * 2 + 1];
            }
        }

        private CommitForStorage getCommit(NetMQMessage message)
        {
            var context = message.Pop().ConvertToString();
            var stream = message.Pop().ConvertToString();
            var expectedVersion = message.PopInt64();
            var eventCount = message.PopInt32();

            var events = new EventForStorage[eventCount];

            for (int i = 0; i < eventCount; i++)
            {
                var eventId = new Guid(message.Pop().ToByteArray());
                var timestamp = message.PopDateTime();
                var typeKey = message.PopString();
                var headers = message.PopStringOrNull();
                var body = message.PopString();

                //-1 to override concurrency check. Being lazy and not using a constant.
                var version = expectedVersion == -1 ? -1 : expectedVersion + i;

                events[i] = new EventForStorage(eventId, version, timestamp, typeKey, headers, body);
            }

            return new CommitForStorage(context, stream, events);
        }

 protected static Vector3 ReadVector01(NetMQMessage msg, ref int curIndex)
 {
     Vector3 ret = Vector3.zero;
     ret.x = ReadFloat01(msg, ref curIndex);
     ret.y = ReadFloat01(msg, ref curIndex);
     ret.z = ReadFloat01(msg, ref curIndex);
     return ret;
 }

        public void ReceiveImplicitConnect_ValidScenario_ShouldReturnRequest()
        {
            const string hostAddress = "tcp://localhost:5557";
            var loggingMessages = new List<string> ();

            // setup the counter socket for communication
            using (var context = NetMQContext.Create ())
            using (var broker = context.CreateRouterSocket ())
            using (var poller = new Poller ())
            using (var session = new MDPWorker (hostAddress, "test", new[] { (byte) '1' }))
            {
                broker.Bind (hostAddress);
                // we need to pick up any message in order to avoid errors
                broker.ReceiveReady += (s, e) =>
                {
                    var msg = e.Socket.ReceiveMultipartMessage ();
                    // we expect to receive a 5 Frame message
                    // [WORKER ADR][EMPTY]["MDPW01"]["READY"]["test"]
                    if (msg.FrameCount != 5)
                        Assert.Fail ("Message with wrong count of frames {0}", msg.FrameCount);
                    // make sure the frames are as expected
                    Assert.That (msg[1], Is.EqualTo (NetMQFrame.Empty));
                    Assert.That (msg[2].ConvertToString (), Is.EqualTo ("MDPW01"));
                    Assert.That (msg[3].BufferSize, Is.EqualTo (1));
                    Assert.That (msg[3].Buffer[0], Is.EqualTo ((byte) MDPCommand.Ready));
                    Assert.That (msg[4].ConvertToString (), Is.EqualTo ("test"));

                    // tell worker to stop gracefully
                    var reply = new NetMQMessage ();
                    reply.Push (new[] { (byte) MDPCommand.Kill });
                    // push MDP Version
                    reply.Push (msg[2]);
                    // push separator
                    reply.Push (NetMQFrame.Empty);
                    // push worker address
                    reply.Push (msg[0]);
                    // send reply which is a request for the worker
                    e.Socket.SendMessage (reply);
                };

                poller.AddSocket (broker);
                Task.Factory.StartNew (poller.PollTillCancelled);

                // set the event handler to receive the logging messages
                session.LogInfoReady += (s, e) => loggingMessages.Add (e.Info);
                // initialise the worker - broker protocol
                session.Receive (null);

                poller.CancelAndJoin ();
                poller.RemoveSocket (broker);

                Assert.That (loggingMessages.Count, Is.EqualTo (5));
                Assert.That (loggingMessages[0], Is.EqualTo ("[WORKER] connected to broker at tcp://localhost:5557"));
                Assert.That (loggingMessages[1].Contains ("[WORKER] sending"), Is.True);
                Assert.That (loggingMessages[2].Contains ("[WORKER] received"));
                Assert.That (loggingMessages[4].Contains ("abandoning"));
            }
        }

        public override void SetFromNetMQMessage(NetMQMessage message)
        {
            base.SetFromNetMQMessage(message);

            if (message.FrameCount != 0)
            {
                throw new NetMQSecurityException(NetMQSecurityErrorCode.InvalidFramesCount, "Malformed message");
            }
        }

        /// <summary>
        /// Send a message to the _server using ZeroMQ
        /// </summary>
        /// <param name="packet">Packet to transmit</param>
        public void Transmit(Packet packet)
        {
            var payload = JsonConvert.SerializeObject(packet);

            var message = new NetMQMessage();

            message.Append(payload);

            _server.SendMultipartMessage(message);
        }

        public void AppendInt32()
        {
            NetMQMessage message = new NetMQMessage();

            message.Append("Hello");
            message.Append(5);

            Assert.AreEqual(4, message[1].MessageSize);
            Assert.AreEqual(5, message[1].ConvertToInt32());
        }

 public async Task SendMessage(NetMQMessage message)
 {
     // instead of creating inproc socket which listen to messages and then send 
     //to the server we just creating task and run a code on
     // the poller thread which the the thread of the clientSocket
     Task task = new Task(() => clientSocket.SendMessage(message));
     task.Start(scheduler);
     await task;
     await ReceiveMessage();
 }

 public void Handle(NetMQFrame[] sender, NetMQMessage message)
 {
     Log.Debug("[CommitHandler] Got a commit to write...");
     var requestId = message.Pop();
     var commit = getCommit(message);
     var task = _writer.Store(commit);
     var commitContinuationContext = new CommitContinuationContext(sender, commit.CommitId, requestId);
     task.ContinueWith(onComplete, commitContinuationContext, TaskContinuationOptions.ExecuteSynchronously);
     Log.Debug("[CommitHandler] Commit queued up...");
 }

        public virtual void SetFromNetMQMessage(NetMQMessage message)
        {
            if (message.FrameCount == 0)
            {
                throw new NetMQSecurityException(NetMQSecurityErrorCode.InvalidFramesCount, "Malformed message");
            }

            // remove the handshake type column
            message.Pop();
        }