pathopub.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

using ZeroMQ;

namespace Examples
{
    static partial class Program
    {

        private static void ZapLoop(ZContext context)
        {
            using (var socket = ZSocket.Create(context, ZSocketType.REP))
            {
                socket.ReceiveTimeout = TimeSpan.FromSeconds(1);
                socket.Bind("inproc://zeromq.zap.01");

                while (true)
                {
                    ZError error;
                    using (ZFrame msg = socket.ReceiveFrame(ZSocketFlags.None, out error))
                    {
                        // ZError.EAGAIN occurs when socket timeout occurs.
                        if (msg == null && error.Number == ZError.EAGAIN)
                        {
                            return;
                        }

                        // if the error code isn't empty throw an exception
                        if (error != null)
                        {
                            throw new Exception(error.ToString());
                        }

                        Console.WriteLine(msg.ReadString());
                    }
                }
            }
        }

        public static void Main(string[] args)
        {
            //
            // Pathological publisher
            // Sends out 1,000 topics and then one random update per second
            //
            // Author: metadings
            //

            if (args == null || args.Length < 1)
            {
                Console.WriteLine();
                Console.WriteLine("Usage: ./{0} PathoPub [Endpoint]", AppDomain.CurrentDomain.FriendlyName);
                Console.WriteLine();
                Console.WriteLine("    Endpoint  Where PathoPub should connect to.");
                Console.WriteLine("              Default is null, Binding on tcp://*:5556");
                Console.WriteLine();
                args = new string[] { null };
            }

            using (var context = new ZContext())
            {
                var thread = new Thread(() => ZapLoop(context));
                thread.Start();

                using (var publisher = new ZSocket(context, ZSocketType.PUB))
                {

                    publisher.CurveServer = true;
                    publisher.CurveSecretKey = Encoding.ASCII.GetBytes(Z85.Decode("JTKVSB%%)wK0E.X)V>+}o?pNmC{O&4W4b!Ni{Lh6"));

                    if (args[0] != null)
                    {
                        publisher.Connect(args[0]);
                    }
                    else
                    {
                        publisher.Bind("tcp://*:5556");
                    }

                    // Ensure subscriber connection has time to complete
                    Thread.Sleep(100);

                    // Send out all 1,000 topic messages
                    for (int topic = 0; topic < 100; ++topic)
                    {
                        publisher.SendMore(new ZFrame(string.Format("{0:D3}", topic)));
                        publisher.Send(new ZFrame("Save Roger"));
                    }

                    // Send one random update per second
                    var rnd = new Random();
                    while (true)
                    {
                        Thread.Sleep(10);
                        publisher.SendMore(new ZFrame(string.Format("{0:D3}", rnd.Next(100))));
                        publisher.Send(new ZFrame("Off with his head!"));
                    }
                }
            }
        }
    }
}

pathosub.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

using ZeroMQ;

namespace Examples
{
    static partial class Program
    {
        public static void Main(string[] args)
        {
            //
            // Pathological subscriber
            // Subscribes to one random topic and prints received messages
            //
            // Author: metadings
            //

            if (args == null || args.Length < 1)
            {
                Console.WriteLine();
                Console.WriteLine("Usage: ./{0} PathoSub [Endpoint]", AppDomain.CurrentDomain.FriendlyName);
                Console.WriteLine();
                Console.WriteLine("    Endpoint  Where PathoSub should connect to.");
                Console.WriteLine("              Default is tcp://127.0.0.1:5556");
                Console.WriteLine();
                args = new string[] { "tcp://127.0.0.1:5556" };
            }

            using (var context = new ZContext())
            using (var subscriber = new ZSocket(context, ZSocketType.SUB))
            {

                subscriber.CurveServerKey = Encoding.ASCII.GetBytes(Z85.Decode("rq:rM>}U?@Lns47E1%kR.o@n%FcmmsL/@{H8]yf7"));
                subscriber.CurveSecretKey = Encoding.ASCII.GetBytes(Z85.Decode("D:)Q[IlAW!ahhC2ac:9*A}h:p?([4%wOTJ%JR%cs"));
                subscriber.CurvePublicKey = Encoding.ASCII.GetBytes(Z85.Decode("Yne@$w-vo<fVvi]a<NY6T1ed:M$fCG*[IaLV{hID"));

                subscriber.Connect(args[0]);

                var rnd = new Random();
                var subscription = string.Format("{0:D3}", rnd.Next(100));
                subscriber.Subscribe(subscription);

                ZMessage msg;
                ZError error;
                while (true)
                {
                    if (null == (msg = subscriber.ReceiveMessage(out error)))
                    {
                        if (error == ZError.ETERM)
                            break;    // Interrupted
                        throw new ZException(error);
                    }
                    using (msg)
                    {
                        if (msg[0].ReadString() != subscription)
                        {
                            throw new InvalidOperationException();
                        }
                        Console.WriteLine(msg[1].ReadString());
                    }
                }
            }
        }
    }
}