/// <summary>
        /// Create a Zyre API that communicates with a node on the ZRE bus.
        /// </summary>
        /// <param name="name">The name of the node</param>
        /// <param name="useEvents">Set this to true to disable Receive() and instead subscribe to events for getting messages from peers. Default is true.</param>
        /// <param name="loggerDelegate">An action to take for logging when _verbose is true. Default is null.</param>
        public Zyre (string name, bool useEvents = true, Action<string> loggerDelegate = null)
        {
            _useEvents = useEvents;
            // Create front-to-back pipe pair for data traffic
            // outbox is passed to ZyreNode for sending Zyre message traffic back to _inbox
            PairSocket outbox;
            PairSocket.CreateSocketPair(out outbox, out _inbox);

            // Start node engine and wait for it to be ready
            // All node control is done through _actor

            _actor = ZyreNode.Create(outbox, loggerDelegate);

            if (useEvents)
            {
                _inboxPoller = new NetMQPoller();
                _inbox.ReceiveReady += InboxReceiveReady;
                _inboxPoller.RunAsync();
            }

            // Send name, if any, to node ending
            if (!string.IsNullOrEmpty(name))
            {
                _actor.SendMoreFrame("SET NAME").SendFrame(name);
            }
        }

        public void ResponsePoll()
        {
            using (var rep = new ResponseSocket())
            using (var req = new RequestSocket())
            using (var poller = new NetMQPoller { rep })
            {
                int port = rep.BindRandomPort("tcp://127.0.0.1");

                req.Connect("tcp://127.0.0.1:" + port);

                rep.ReceiveReady += (s, e) =>
                {
                    bool more;
                    Assert.AreEqual("Hello", e.Socket.ReceiveFrameString(out more));
                    Assert.False(more);

                    e.Socket.SendFrame("World");
                };

                poller.RunAsync();

                req.SendFrame("Hello");

                bool more2;
                Assert.AreEqual("World", req.ReceiveFrameString(out more2));
                Assert.IsFalse(more2);

                poller.Stop();
            }
        }

        void IShimHandler.Run(PairSocket shim)
        {
            Poller = new NetMQPoller();


            Shim = shim;
            Shim.ReceiveReady += OnShimReady;
            Poller.Add(Shim);

            Initialize();

            Shim.SignalOK();

            Poller.Run();

            Cleanup();
        }

        public void TwoMessagesFromRouterToDealer()
        {
            using (var server = new RouterSocket())
            using (var client = new DealerSocket())
            using (var poller = new NetMQPoller { client })
            {
                var port = server.BindRandomPort("tcp://*");
                client.Connect("tcp://127.0.0.1:" + port);
                var cnt = 0;
                client.ReceiveReady += (sender, e) =>
                {
                    var strs = e.Socket.ReceiveMultipartStrings();
                    foreach (var str in strs)
                    {
                        Console.WriteLine(str);
                    }
                    cnt++;
                    if (cnt == 2)
                    {
                        poller.Stop();
                    }
                };
                byte[] clientId = Encoding.Unicode.GetBytes("ClientId");
                client.Options.Identity = clientId;

                const string request = "GET /\r\n";

                const string response = "HTTP/1.0 200 OK\r\n" +
                        "Content-Type: text/plain\r\n" +
                        "\r\n" +
                        "Hello, World!";

                client.SendFrame(request);

                byte[] serverId = server.ReceiveFrameBytes();
                Assert.AreEqual(request, server.ReceiveFrameString());

                // two messages in a row, not frames
                server.SendMoreFrame(serverId).SendFrame(response);
                server.SendMoreFrame(serverId).SendFrame(response);

                poller.Run();
            }
        }

        public void WithPoller()
        {
            using (var queue = new NetMQQueue<int>())
            using (var poller = new NetMQPoller { queue })
            {
                var manualResetEvent = new ManualResetEvent(false);

                queue.ReceiveReady += (sender, args) =>
                {
                    Assert.AreEqual(1, queue.Dequeue());
                    manualResetEvent.Set();
                };

                poller.RunAsync();

                Assert.IsFalse(manualResetEvent.WaitOne(100));
                queue.Enqueue(1);
                Assert.IsTrue(manualResetEvent.WaitOne(100));
            }
        }

        public void Monitoring()
        {
            var listeningEvent = new ManualResetEvent(false);
            var acceptedEvent = new ManualResetEvent(false);
            var connectedEvent = new ManualResetEvent(false);

            using (var rep = new ResponseSocket())
            using (var req = new RequestSocket())
            using (var poller = new NetMQPoller())
            using (var repMonitor = new NetMQMonitor(rep, "inproc://rep.inproc", SocketEvents.Accepted | SocketEvents.Listening))
            using (var reqMonitor = new NetMQMonitor(req, "inproc://req.inproc", SocketEvents.Connected))
            {
                repMonitor.Accepted += (s, e) => acceptedEvent.Set();
                repMonitor.Listening += (s, e) => listeningEvent.Set();

                repMonitor.AttachToPoller(poller);

                int port = rep.BindRandomPort("tcp://127.0.0.1");

                reqMonitor.Connected += (s, e) => connectedEvent.Set();

                reqMonitor.AttachToPoller(poller);

                poller.RunAsync();

                req.Connect("tcp://127.0.0.1:" + port);
                req.SendFrame("a");

                rep.SkipFrame();

                rep.SendFrame("b");

                req.SkipFrame();

                Assert.IsTrue(listeningEvent.WaitOne(300));
                Assert.IsTrue(connectedEvent.WaitOne(300));
                Assert.IsTrue(acceptedEvent.WaitOne(300));

                poller.Stop();
            }
        }

        public void CycleCreateTerminate()
        {
            NetMQConfig.ContextCreate(true);
            var isTerminated = VerifyTermination();
            Assert.AreEqual(false, isTerminated);

            // We use the Poller Test code.
            using (var rep = new ResponseSocket())
            using (var req = new RequestSocket())
            using (var poller = new NetMQPoller { rep })
            {
                var port = rep.BindRandomPort("tcp://127.0.0.1");

                req.Connect("tcp://127.0.0.1:" + port);

                rep.ReceiveReady += (s, e) =>
                {
                    bool more;
                    Assert.AreEqual("Hello", e.Socket.ReceiveFrameString(out more));
                    Assert.False(more);

                    e.Socket.SendFrame("World");
                };

                poller.RunAsync();

                req.SendFrame("Hello");

                bool more2;
                Assert.AreEqual("World", req.ReceiveFrameString(out more2));
                Assert.IsFalse(more2);

                poller.Stop();
            }
            NetMQConfig.ContextTerminate();
            isTerminated = VerifyTermination();
            Assert.AreEqual(true, isTerminated);
        }

        public void TwoThreads()
        {
            int count1 = 0;
            int count2 = 0;

            var allTasks = new ConcurrentBag<Task>();

            using (var poller = new NetMQPoller())
            {
                poller.RunAsync();

                Task t1 = Task.Factory.StartNew(() =>
                {
                    for (int i = 0; i < 100; i++)
                    {
                        var task = new Task(() => { count1++; });
                        allTasks.Add(task);
                        task.Start(poller);
                    }
                });

                Task t2 = Task.Factory.StartNew(() =>
                {
                    for (int i = 0; i < 100; i++)
                    {
                        var task = new Task(() => { count2++; });
                        allTasks.Add(task);
                        task.Start(poller);
                    }
                });

                t1.Wait(1000);
                t2.Wait(1000);
                Task.WaitAll(allTasks.ToArray(), 1000);

                Assert.AreEqual(100, count1);
                Assert.AreEqual(100, count2);
            }
        }

        public void ContinueWith()
        {
            int threadId1 = 0;
            int threadId2 = 1;

            int runCount1 = 0;
            int runCount2 = 0;

            using (var poller = new NetMQPoller())
            {
                poller.RunAsync();

                var task = new Task(() =>
                {
                    threadId1 = Thread.CurrentThread.ManagedThreadId;
                    runCount1++;
                });

                var task2 = task.ContinueWith(t =>
                {
                    threadId2 = Thread.CurrentThread.ManagedThreadId;
                    runCount2++;
                }, poller);

                task.Start(poller);
                task.Wait();
                task2.Wait();

                Assert.AreEqual(threadId1, threadId2);
                Assert.AreEqual(1, runCount1);
                Assert.AreEqual(1, runCount2);
            }
        }

        public void CanExecuteTaskInline()
        {
            using (var poller = new NetMQPoller())
            {
                Assert.IsFalse(poller.CanExecuteTaskInline);

                poller.RunAsync();

                Assert.IsFalse(poller.CanExecuteTaskInline);

                var task = new Task(() => Assert.IsTrue(poller.CanExecuteTaskInline));
                task.Start(poller);
                task.Wait();
            }
        }

        public void SetsCurrentTaskScheduler()
        {
            using (var poller = new NetMQPoller())
            {
                poller.RunAsync();

                var task = new Task(() => Assert.AreSame(TaskScheduler.Current, poller));
                task.Start(poller);
                task.Wait();
            }
        }

        public void OneTask()
        {
            bool triggered = false;

            using (var poller = new NetMQPoller())
            {
                poller.RunAsync();

                var task = new Task(() =>
                {
                    triggered = true;
                    Assert.IsTrue(poller.CanExecuteTaskInline, "Should be on NetMQPoller thread");
                });
                task.Start(poller);
                task.Wait();

                Assert.IsTrue(triggered);
            }
        }

        public void NativeSocket()
        {
            using (var streamServer = new StreamSocket())
            using (var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            {
                int port = streamServer.BindRandomPort("tcp://*");

                socket.Connect("127.0.0.1", port);

                var buffer = new byte[] { 1 };
                socket.Send(buffer);

                byte[] identity = streamServer.ReceiveFrameBytes();
                byte[] message = streamServer.ReceiveFrameBytes();

                Assert.AreEqual(buffer[0], message[0]);

                var socketSignal = new ManualResetEvent(false);

                using (var poller = new NetMQPoller())
                {
                    poller.Add(socket, s =>
                    {
                        socket.Receive(buffer);

                        socketSignal.Set();

                        // removing the socket
                        poller.Remove(socket);
                    });

                    poller.RunAsync();

                    // no message is waiting for the socket so it should fail
                    Assert.IsFalse(socketSignal.WaitOne(100));

                    // sending a message back to the socket
                    streamServer.SendMoreFrame(identity).SendFrame("a");

                    Assert.IsTrue(socketSignal.WaitOne(100));

                    socketSignal.Reset();

                    // sending a message back to the socket
                    streamServer.SendMoreFrame(identity).SendFrame("a");

                    // we remove the native socket so it should fail
                    Assert.IsFalse(socketSignal.WaitOne(100));

                    poller.Stop();
                }
            }
        }

        public void Polling()
        {
            using (var speaker = new NetMQBeacon())
            using (var listener = new NetMQBeacon())
            {
                speaker.Configure(9999);

                speaker.Publish("Hello", s_publishInterval);

                var manualResetEvent = new ManualResetEvent(false);

                listener.Configure(9999);
                listener.Subscribe("H");

                string message = "";

                listener.ReceiveReady += (sender, args) =>
                {
                    message = listener.Receive().String;
                    manualResetEvent.Set();
                };

                using (var poller = new NetMQPoller { listener })
                {
                    poller.RunAsync();

                    manualResetEvent.WaitOne();

                    Assert.AreEqual("Hello", message);
                }
            }
        }

        public void ISynchronizeInvokeWorks()
        {
            using (var poller = new NetMQPoller())
            using (var signal = new ManualResetEvent(false))
            using (var timer = new System.Timers.Timer { AutoReset = false, Interval = 100, SynchronizingObject = poller, Enabled = true })
            {
                var isCorrectThread = false;

                poller.RunAsync();

                timer.Elapsed += (sender, args) =>
                {
                    isCorrectThread = poller.CanExecuteTaskInline;
                    Assert.True(signal.Set());
                };

                Assert.True(signal.WaitOne(TimeSpan.FromSeconds(2)));
                Assert.True(isCorrectThread);
            }
        }

        public void TestProxySendAndReceiveWithExternalPoller()
        {
            using (var front = new RouterSocket())
            using (var back = new DealerSocket())
            using (var poller = new NetMQPoller { front, back })
            {
                front.Bind("inproc://frontend");
                back.Bind("inproc://backend");

                var proxy = new Proxy(front, back, null, poller);
                proxy.Start();

                poller.RunAsync();
                
                using (var client = new RequestSocket())
                using (var server = new ResponseSocket())
                {
                    client.Connect("inproc://frontend");
                    server.Connect("inproc://backend");

                    client.SendFrame("hello");
                    Assert.AreEqual("hello", server.ReceiveFrameString());
                    server.SendFrame("reply");
                    Assert.AreEqual("reply", client.ReceiveFrameString());

                    // Now stop the external poller
                    poller.Stop();

                    client.SendFrame("anyone there?");

                    // Should no longer receive any messages
                    Assert.IsFalse(server.TrySkipFrame(TimeSpan.FromMilliseconds(50)));
                }
            }
        }

        public void RunMultipleTimes()
        {
            int count = 0;

            const int timerIntervalMillis = 20;

            var timer = new NetMQTimer(TimeSpan.FromMilliseconds(timerIntervalMillis));
            timer.Elapsed += (s, a) =>
            {
                count++;

                if (count == 3)
                {
                    timer.Enable = false;
                }
            };

            using (var poller = new NetMQPoller { timer })
            {
                poller.RunAsync();

                Thread.Sleep(timerIntervalMillis * 6);

                poller.Stop();

                Assert.AreEqual(3, count);
            }
        }

        public void RemoveTimer()
        {
            using (var router = new RouterSocket())
            using (var dealer = new DealerSocket())
            using (var poller = new NetMQPoller { router })
            {
                int port = router.BindRandomPort("tcp://127.0.0.1");

                dealer.Connect("tcp://127.0.0.1:" + port);

                bool timerTriggered = false;

                var timer = new NetMQTimer(TimeSpan.FromMilliseconds(100));
                timer.Elapsed += (s, a) => { timerTriggered = true; };

                // The timer will fire after 100ms
                poller.Add(timer);

                bool messageArrived = false;

                router.ReceiveReady += (s, e) =>
                {
                    router.SkipFrame();
                    router.SkipFrame();
                    messageArrived = true;
                    // Remove timer
                    poller.Remove(timer);
                };

                poller.RunAsync();

                Thread.Sleep(20);

                dealer.SendFrame("hello");

                Thread.Sleep(300);

                poller.Stop();

                Assert.IsTrue(messageArrived);
                Assert.IsFalse(timerTriggered);
            }
        }

        public void SimpleTimer()
        {
            // TODO it is not really clear what this test is actually testing -- maybe split it into a few smaller tests

            using (var router = new RouterSocket())
            using (var dealer = new DealerSocket())
            using (var poller = new NetMQPoller { router })
            {
                int port = router.BindRandomPort("tcp://127.0.0.1");

                dealer.Connect("tcp://127.0.0.1:" + port);

                bool messageArrived = false;

                router.ReceiveReady += (s, e) =>
                {
                    Assert.IsFalse(messageArrived);
                    router.SkipFrame();
                    router.SkipFrame();
                    messageArrived = true;
                };

                bool timerTriggered = false;

                int count = 0;

                const int timerIntervalMillis = 100;

                var timer = new NetMQTimer(TimeSpan.FromMilliseconds(timerIntervalMillis));
                timer.Elapsed += (s, a) =>
                {
                    // the timer should jump before the message
                    Assert.IsFalse(messageArrived);
                    timerTriggered = true;
                    timer.Enable = false;
                    count++;
                };
                poller.Add(timer);

                poller.RunAsync();

                Thread.Sleep(150);

                dealer.SendFrame("hello");

                Thread.Sleep(300);

                poller.Stop();

                Assert.IsTrue(messageArrived);
                Assert.IsTrue(timerTriggered);
                Assert.AreEqual(1, count);
            }
        }

        public void RemoveSocket()
        {
            using (var router1 = new RouterSocket())
            using (var router2 = new RouterSocket())
            using (var router3 = new RouterSocket())
            using (var dealer1 = new DealerSocket())
            using (var dealer2 = new DealerSocket())
            using (var dealer3 = new DealerSocket())
            using (var poller = new NetMQPoller { router1, router2, router3 })
            {
                int port1 = router1.BindRandomPort("tcp://127.0.0.1");
                int port2 = router2.BindRandomPort("tcp://127.0.0.1");
                int port3 = router3.BindRandomPort("tcp://127.0.0.1");

                dealer1.Connect("tcp://127.0.0.1:" + port1);
                dealer2.Connect("tcp://127.0.0.1:" + port2);
                dealer3.Connect("tcp://127.0.0.1:" + port3);

                bool first = true;

                router1.ReceiveReady += (s, e) =>
                {
                    if (!first)
                        Assert.Fail("This should not happen because we cancelled the socket");
                    first = false;

                    // identity
                    e.Socket.SkipFrame();

                    bool more;
                    Assert.AreEqual("Hello", e.Socket.ReceiveFrameString(out more));
                    Assert.False(more);

                    // cancelling the socket
                    poller.Remove(e.Socket); // remove self
                };

                router2.ReceiveReady += (s, e) =>
                {
                    // identity
                    byte[] identity = e.Socket.ReceiveFrameBytes();

                    // message
                    e.Socket.SkipFrame();

                    e.Socket.SendMoreFrame(identity);
                    e.Socket.SendFrame("2");
                };

                router3.ReceiveReady += (s, e) =>
                {
                    // identity
                    byte[] identity = e.Socket.ReceiveFrameBytes();

                    // message
                    e.Socket.SkipFrame();

                    e.Socket.SendMoreFrame(identity).SendFrame("3");
                };

                Task pollerTask = Task.Factory.StartNew(poller.Run);

                // Send three messages. Only the first will be processed, as then handler removes
                // the socket from the poller.
                dealer1.SendFrame("Hello");
                dealer1.SendFrame("Hello2");
                dealer1.SendFrame("Hello3");

                // making sure the socket defined before the one cancelled still works
                dealer2.SendFrame("1");
                Assert.AreEqual("2", dealer2.ReceiveFrameString());

                // making sure the socket defined after the one cancelled still works
                dealer3.SendFrame("1");
                Assert.AreEqual("3", dealer3.ReceiveFrameString());

                poller.Stop();
                Assert.IsTrue(pollerTask.IsCompleted);
            }
        }