nanomsg.examples.pubsub source code

1 /**
2     Nanomsg Example: Pubsub
3 */
4 module nanomsg.examples.pubsub;
5 
6 import std.stdio;
7 import std.datetime;
8 import core.thread;
9 import nanomsg;
10 import std.string:toStringz;
11 
12 ///
13 enum SERVER ="server";
14 ///
15 enum CLIENT ="client";
16 
17 ///
18 string date ()
19 {
20   return Clock.currTime.toSimpleString();
21 }
22 
23 ///
24 int server (string surl)
25 {
26   auto url=surl.toStringz;
27   int sock = nn_socket (AF_SP, NN_PUB);
28   assert (sock >= 0);
29   assert (nn_bind (sock, url) >= 0);
30   while (1)
31   {
32       auto d = date();
33       int sz_d = cast(int)d.length+1;
34       writefln("SERVER: PUBLISHING DATE %s", d);
35       int bytes = nn_send(sock, d.toStringz, sz_d, 0);
36       assert (bytes == sz_d);
37       Thread.sleep( dur!("seconds")( 1 ) );
38   }
39   return nn_shutdown (sock, 0);
40 }
41 
42 ///
43 int client (string surl, string sname)
44 {
45   auto url=surl.toStringz;
46   auto name=sname.toStringz;
47   int sock = nn_socket (AF_SP, NN_SUB);
48   assert (sock >= 0);
49   // TODO learn more about publishing/subscribe keys
50   assert (nn_setsockopt(sock, NN_SUB, NN_SUB_SUBSCRIBE, "".toStringz, 0) >= 0);
51   assert (nn_connect (sock, url) >= 0);
52   while (1)
53     {
54       char *buf = cast(char*)0;
55       int bytes = nn_recv (sock, &buf, NN_MSG, 0);
56       assert (bytes >= 0);
57       printf ("CLIENT (%s): RECEIVED %s\n", name, buf);
58       nn_freemsg (buf);
59     }
60   return nn_shutdown (sock, 0);
61 }
62 
63 ///
64 int main (string[] argv)
65 {
66   if (argv.length>=2)
67     if (SERVER==argv[1])
68       return server (argv[2]);
69   if (argv.length>=3)
70     if (CLIENT==argv[1])
71       return client (argv[2], argv[3]);
72   writefln("Usage: pubsub %s|%s <URL> <ARG> ...",SERVER, CLIENT);
73   return 1;
74 }