14
$\begingroup$

Mathematica 11.2 can do ZMQ socket programming now! But I cannot find out how to set subscribe channel from searching the doc. When using c language there is a function zsocket_set_subscribe to set the subscribe channel, it seems lacking corresponding function in Mathematica Socket API .

(* To create socket *)
client = SocketConnect[{"Pub IP", Pub Port}, "ZMQ_SUB"]

(* Then how to set channel to receive message? *)

I also ask in the wolfram community and will sync if there is an answer.

$\endgroup$
3
$\begingroup$

I think this part of the interface is broken. From some digging I found: ZeroMQLink`SocketOptions`Private`setSockOptSingle which is called by SetOptions on a SocketObject. This makes use of ZeroMQLink`SocketOptions`$SetSocketOptionTypes to check the type of the argument and ZeroMQLink`SocketOptions`$SetSocketOptionValues to get the integer for what I'm assuming is the enum of the options.

I looked at ZeroMQLink`SocketOptions`$SetSocketOptionValues and found that "ZMQ_SUBSCRIBE" is supported, which we can get by ZeroMQLink`SocketOptions`toZMQOptionString@"Subscribe". So in principle "Subscribe" should be supported.

Then looking at what the type should be we get:

ZeroMQLink`SocketOptions`$SetSocketOptionTypes[
 ZeroMQLink`SocketOptions`toZMQOptionString@"Subscribe"
 ]

ListQ[#1] && ByteArrayQ[ByteArray[#1]] &

Which basically tells us that we need to pass a list of integers. So then I tried:

pub = SocketOpen[Automatic, "ZMQ_PUB"];
sub = SocketConnect["tcp://localhost:8000", "ZMQ_SUB"];

SetOptions[sub, "Subscribe" -> Normal@BinarySerialize[First@pub]]

LibraryFunction::cfsa: Argument {55,58,83,36,99,99,50,101,51,54,102,98,45,53,48,51,98,45,52,98,53,97,45,56,99,55,50,45,55,101,101,53,49,56,99,48,51,55,54,50} at position 3 should be a machine-size integer.

And then digging more, we can patch this like so:

ZeroMQLink`Libraries`iSetZMQSocketOption[
   uuid_, option_?IntegerQ, val : {_?IntegerQ ..}
   ] :=
  Once[
    Quiet[
     LibraryFunctionLoad[FindLibrary["libzeromqlink"],
      "SetZMQSocketOption", {String, Integer, {Integer, 1}}, {Integer,
        1}], LibraryFunction::overload
     ]
    ][uuid, option, val];

SetOptions[sub, "Subscribe" -> Normal@BinarySerialize[First@pub]]

What this does I have no idea. Whether it does what you want I have no idea. But at least it throws no errors.

$\endgroup$
  • $\begingroup$ One can test by writing something to the publisher socket Write[pub, "Hi there"] and then check if the subscription socket has some message to read from by SocketReadyQ@sub. As it returns False I second your opinion that the interface is broken. Even if one starts pub and sub like this pub = SocketOpen[{IPAddress["127.0.0.1"], 8000}, "ZMQ_PUB"]; sub = SocketConnect["tcp://localhost:8000", "ZMQ_SUB"]; it does not seem to help. $\endgroup$ – PlatoManiac Mar 2 '18 at 2:18
3
$\begingroup$

https://www.wolfram.com/language/12/networking-and-system-operations/zeromq-publisher-model.html

pub = SocketOpen[Automatic, "ZMQ_PUB"]

subscribers = Table[SocketConnect[pub["DestinationPort"], "ZMQ_SUB"], 3]

Scan[SetOptions[#, "ZMQ_SUBSCRIBE" -> All] &, subscribers]

WriteString[pub, "Message from publisher socket."]

ByteArrayToString[SocketReadMessage[#]] & /@ subscribers

{"Message from publisher socket.", "Message from publisher socket.", "Message from publisher socket."}

$\endgroup$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.