Skip to content

Instantly share code, notes, and snippets.

@robertpi robertpi/pathopub.cs
Last active Aug 29, 2015

Embed
What would you like to do?
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!"));
}
}
}
}
}
}
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());
}
}
}
}
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.