Hot questions for Using ZeroMQ in jzmq

Question:

I am trying to use ZeroMQ's pub-sub sockets. However, I don't clearly understand the role of context (zmq::context_t) while creating sockets (zmq::socket_t).

Assuming that I want to create 5 subscriber sockets (zmq::socket_t using ZMQ_SUB), do I need 5 contexts, one for each of the subscriber sockets? Or can I use a single context for all 5 sockets?


Answer:

Assuming that I want to create 5 subscriber sockets ( zmq::socket_t using ZMQ_SUB ), do I need 5 contexts, one for each of the subscriber sockets? Or can I use a single context for all 5 sockets?

You need only one Context instance for this light-weight use-case. Check the part of documentation attached below explaining 0MQ context usage and an example I made for you attached at the end of this post.

ZeroMQ applications always start by creating a context, and then using that for creating sockets. In C, it's the zmq_ctx_new() call. You should create and use exactly one context in your process. Technically, the context is the container for all sockets in a single process, and acts as the transport for inproc sockets, which are the fastest way to connect threads in one process. If at runtime a process has two contexts, these are like separate ZeroMQ instances.

I made an example for you below as a help for you in understanding ZMQ context and ZMQ PUB-SUB pattern. Creating 5 subscriber sockets is fine as long as you have 5 producing services. However if you have one source publishing notifications I would recommend using PUB-SUB pattern and filtering property of ZMQ SUB sockets. You can check how to set that up below in my code in communication between publisher #1 and subscriber.

Publisher #1 sends temperature and humidity updates..

import zmq
from time import sleep

# Server socket
context = zmq.Context()
socket  = context.socket( zmq.PUB )
socket.bind( "tcp://*:5556" )

while True:
    socket.send_multipart( [ "TEMP", "25.40" ] )
    socket.send_multipart( [ "HUMD", "48.90" ] )
    sleep( 1 )

Publisher #2 sends pressure updates..

import zmq
from time import sleep

# Server socket
context = zmq.Context()
socket2 = context.socket( zmq.PUB )
socket2.bind( "tcp://*:5557" )

while True:
    socket2.send_multipart( [ "PRSS", "10000.00" ] )
    sleep( 1 )

Subscriber registered to temperature, humidity and pressure updates on two different servers..

import zmq
from time import sleep

# Sockets to talk to servers
context = zmq.Context()
socket  = context.socket( zmq.SUB )
socket.connect(  "tcp://localhost:5556" )
socket2 = context.socket( zmq.SUB )
socket2.connect( "tcp://localhost:5557" )

# Set filters
socket.setsockopt_string(  zmq.SUBSCRIBE, "TEMP".decode( 'ascii' ) )
socket.setsockopt_string(  zmq.SUBSCRIBE, "HUMD".decode( 'ascii' ) )
socket2.setsockopt_string( zmq.SUBSCRIBE, "PRSS".decode( 'ascii' ) )

poller = zmq.Poller()
poller.register( socket,  zmq.POLLIN )
poller.register( socket2, zmq.POLLIN )

while True:
    socks = dict( poller.poll() )
    if socket in socks and socks[socket] == zmq.POLLIN:
        [ measurement, value ] = socket.recv_multipart()
        print measurement
        print value

    if socket2 in socks and socks[socket2] == zmq.POLLIN:
        [ measurement, value ] = socket2.recv_multipart()
        print measurement
        print value

    sleep( 1 )

Question:

I'm interested in finding out if ZMQ already compresses messages before sending them out so as to not do this myself which would be redundant to compress a message twice.

If it does do this is it automatic, or is there an option param to specify?

I'm using java but it really doesn't matter I guess for libzmq it doesn't matter.


Answer:

ZMQ is not performing any serialization. You can send messages via socket as String or byte[]. See example here: http://zguide.zeromq.org/java:wuserver

publisher.send(update, 0);

where update is String.

In general I recommend guide here http://zguide.zeromq.org/page:all

Question:

Looking the ZeroMQ bindings. I found that jzmq last release was almost three years ago. Is this binding deprecated?

On the other hand there is a native java implementation of ZeroMQ called JeroMQ which is supported by the ZeroMQ community and has a lot of activity in Github and it is based on one of the latest versions of libzmq(4.1.7).

Should all new Java development using ZeroMQ be based on JeroMQ? (Considering that jzmq and JeroMQ APIs are starting to diverge)

How JeroMQ interoperate with ZeroMQ bindings in other languages?


Answer:

How JeroMQ interoperate with ZeroMQ bindings in other languages?

JeroMQ's API has diverged from that of jzmq, but this should have no impact on what's happening at the protocol level. You can write a Java program using JeroMQ and have it communicate with other programs written in other languages / runtimes.

Should all new Java development using ZeroMQ be based on JeroMQ?

jzmq is still potentially useful if performance is very critical to your application, and if requiring end users to install libzmq is acceptable for you.

I looked at the GitHub repo for jzmq, and was surprised to find that, as you said, the last release was 3 years ago in 2014. This is especially surprising given that there have been commits on master as recently as March 2017. So, it looks like the project is still being maintained, but they are rather overdue for a release.

Question:

My program uses ZMQ for communication. Namely, a server (C++, linux) creates an XPUB socket and then in one thread reads it, and in another one publishes data (writes).

The client (java, jzmq, linux) create a SUB socket, and subscribes using it.

After some time, the server side receives SIGABRT in the reading thread.

What may be a source of problem? Read/Write in different threads or creating XPUB/SUB pair?

In case the problem is in multi threading, what is a right paradigm to use XPUB socket?


Answer:

http://zguide.zeromq.org/page:all#Multithreading-with-ZeroMQ

Don't share ZeroMQ sockets between threads. ZeroMQ sockets are not threadsafe. Technically it's possible to migrate a socket from one thread to another but it demands skill. The only place where it's remotely sane to share sockets between threads are in language bindings that need to do magic like garbage collection on sockets.

Question:

I am developing a JAVA multicast application using JZMQ (PGM protocol).

Is it possible to send and receive data through the same socket?

If ZMQ.PUB is used, only send() works and recv() is not working.

If ZMQ.SUB is used, send() doesn't work.

Is there any alternative way for using both send() and recv() using the same Socket?

ZMQ.Context context = ZMQ.context(1);
ZMQ.Socket socket = context.socket(ZMQ.PUB);
socket.send(msg);
socket.recv();

Answer:

Radio broadcast will never deliver your voice into the Main Station

Yes, both parts of the ZeroMQ PUB/SUB Scalable Formal Communication Pattern's archetypes are uni-directional ( by-definition ) one can just .send(), the other(s) may just listen ( and if were configured well, they will ).


How to do what you have asked for? ( ... and forget to have this using pgm:// )

Yes, there are ways to use other ZeroMQ archetypes for this - i.e. a single socket over PAIR/PAIR endpoints( capable of both .send() and .recv() methods ) or a pair of (A)->--PUSH/PULL->-(B) + (A)-<-PULL/PUSH-<-(B) so as to construct the bi-directional signalling / messaging channel by using just uni-directional archetypes.

You also need to select an appropriate transport-class for being used in .bind() + .connect() between the configured ZeroMQ endpoints.

// -------------------------------------------------------- HOST-(A)
   ZMQ.Context aCONTEXT   = ZMQ.context( 1 );

   ZMQ.Socket  aPubSOCKET = aCONTEXT.socket( ZMQ.PUB );
               aPubSOCKET.setsockopt(        ZMQ.LINGER, 0 );
// ----------------------
               aPubSOCKET.bind( "tcp://*:8001" );
// ----------------------
// set msg = ...;
// ----------------------
               aPubSOCKET.send( msg, ZMQ.NOWAIT );

// ...
// ----------------------
               aPubSOCKET.close();
               aCONTEXT.term();
// ----------------------

The SUB-side has one more duty ...

// -------------------------------------------------------- HOST-(B)
   ZMQ.Context aCONTEXT   = ZMQ.context( 1 );

   ZMQ.Socket  aSubSOCKET = aCONTEXT.socket( ZMQ.SUB );
               aSubSOCKET.setsockopt(        ZMQ.LINGER,     0 );
               aSubSOCKET.setsockopt(        ZMQ.SUBSCRIBE, "" );
// ----------------------
               aSubSOCKET.connect( "tcp://<host_A_IP_address>:8001" );
// ----------------------
// def a msg;
// ----------------------
         msg = aSubSOCKET.recv( ZMQ.NOWAIT );

// ...
// ----------------------
               aSubSOCKET.close();
               aCONTEXT.term();
// ----------------------

Question:

My server in python:

import time
import zmq

context = zmq.Context()
socket  = context.socket( zmq.REP )
socket.bind( "tcp://*:5555" )

while True:
    #  Wait for next request from client
    message = socket.recv()
    print( "Received request: %s" % message )

    #  Do some 'work'
    time.sleep( 1 )

    #  Send reply back to client
    socket.send( b"World" )

My C client:

    std::string str = std::to_string(bar.open)   + ';'
                    + std::to_string(bar.high)   + ';'
                    + std::to_string(bar.low)    + ';'
                    + std::to_string(bar.close)  + ';'
                    + std::to_string(bar.volume) + ';'
                    + bar.time                   + ';'
                    + std::to_string(bar.hour)   + ';'
                    + std::to_string(bar.minute) + ';'
                    + bar.date                   + ';'
                    + bar.symbol;
    void *context   = zmq_ctx_new ();
    void *requester = zmq_socket ( context, ZMQ_REQ );
    zmq_connect ( requester, "tcp://localhost:5555" );

    char buffer [10];
    printf (  "Sending data to python module\n" );

    zmq_send ( requester, static_cast<void*>(&str), 1000, 0 );
    zmq_recv ( requester, buffer, 10, 0 );
    printf (  "Received %s\n", buffer );

    zmq_close ( requester );

When I send a message from C client, the data printed in python is garbled, like this:

Received request: @c�SxH���C��
                           %�.075600;C�
                                       %�;C �
                                             %0.075600�C@�
                                                          %�`�
                                                              %���
                                                                  %���
                                                                      %0.075600%���
   %���
       %����C��

How can I decode the message in Python to print out correctly?


Answer:

C-side code sends just "1D" row-of-plain-bytes: char[]whilepython3 thinks in "2D", expecting both: { bytes[], encoding }

this results in desinterpretation inside "string{0:s}".format( message ) execution, as the mini-template expects the message to be indeed a "fully-equipped" python3-string, which it fails to be.

print( ":::{0:s}:::".format( str( message, 'utf-8' ) ) ) # ought fix the game

Another idea is to introduce some wire-line mapper ( or better a protocol )

so as to explicitly control byte-mapped content handling.

In a QuantFX module, the parties in a multiparty distributed FX-engine / ML-predictions processing adhere to a protocol-specification, using struct.unpack() on python side, once aMiniRESOPONDER() has .recv()-ed aMSG.

The whole trouble here reduces to just coordinated protocol-version-control, so any target node can adapt remote-distributed processing to the appropriate protocol version on-the-fly.

            pass;                     #aMSG_STRUCT_MASK = '!IIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIII'
            pass;                      aMSG_STRUCT_MASK = "!" + "I" * ( v41_HeaderSIZE + ( v41_nBarsDEPTH * 7 ) )
            #----DATA------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
            aMSG_DATA = struct.unpack( aMSG_STRUCT_MASK, aMSG )
            #----INT-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
            #DSegINT  = np.asarray(    aMSG_DATA[             2:],              # SKIP HEADER: Open_0, CurrentPRICE
            aDSegINT  = np.asarray(    aMSG_DATA[v41_HeaderSIZE:],              # SKIP HEADER: Open_0, CurrentPRICE [v412: MQL4_Digits, MQL4_GetTickCountOnSEND ]
                                       order = 'F'                              # FORTRAN-column-major-aligned / caching / speed
                                       )

Question:

My Java server runs pretty well for a while with JZMQ. And suddenly it got Assertion failed: check () (msg.cpp:220)

It cannot be caught in JAVA. And the java process is down itself. It seems like Cpp issue in JNI

May I know if any one know what happens in this exception?

Thanks


Answer:

I've seen this during my first ZMQ tests. In my case the exception was caused by a thread closing a socket opened by another thread. It also happened when two threads used one socket at the same time.

I was breaking ZMQ's rule of sharing sockets among threads.

I suggest to check if a thread is using or closing a socket while it's being in use by another thread.

Question:

I'm building a HTML page which renders a graph created using VivaGraph ( https://github.com/anvaka/VivaGraphJS ) .

This graph should be fed by data coming from a ZMQ (ZeroMQ) datastream, where my webpage is connected as "Subscriber" to a "Publisher"; the perfect example code is here: https://github.com/zeromq/zeromq.js/#pubsub

The problem is: VivaGraph is for JS on client side while ZMQ is for NodeJS environment.

How can bind my VivaGraph graph to the ZMQ data stream in the client side?


Answer:

You could look into libraries like JSMQ or zwssock that implement ZMQ over websockets, however these seem harder to use.

I would go the easy way and place a simple express http server in the middle. Html client sends periodical get requests to the express server, server connects to ZMQ and fetches the data and sends it back to the client as json.

Alternatively, you could also do a websockets connection between the html client and the server, while the server also maintains a connection to ZMQ and pipes data to the websockets.

Question:

I am using jzmq package for my project to communicate over network. I am using DEALER ROUTER pair. I have read that the socket with DEALER and ROUTER type is not thread safe. So I can not send or receive from the same socket on 2 different threads.

My questions are: 1 ) What is the purpose of the ZThread class in the jzmq package?

2 ) Does it handle this thread unsafety?

3 ) Can I send and receive from the same socket if using it from a parent thread and its child ZThread?

4 ) Also what is the difference between Attached and AdetachedRunnable?


Answer:

Fact #0: ZeroMQ was never thread-safe ( had Zero sharing Zen )

In spite of the most recent efforts ( published in late 2017 on 4.2+ total re-design efforts towards removing this known initial principle ), ZeroMQ education materials present wherever possible and also explain why there is a bad habit to try to share toys in distributed-system design practice.


Ad 2 )

Even if one gets some API-baked promise, always first benchmark the performance, if one will ever want to pay the costs of lost performance for such ex-post sharing doctrine. As noted about the ZeroMQ native API, there is principally nothing to be shared ( with one exception, which sometimes may make sense, the global Context()-instance ). Threads may "borrow" the IO-socket instantiations from such a global Context()-instance, but never share socket-instances, as the results are not guaranteed inside and under the ZeroMQ native API and so will not be any better even if promised so "above" any kind of higher level API.


Ad 3 ) No,

as per 2 ), never share sockets, there is no reason to try to do that. If managing resources ( and threads are first-class citizen among resources ), better create a private, point to point PAIR/PAIR or PUSH/PULL ( even in tandem of simplex pipes ) over { inproc:// | ipc:// }-transport-class ( where inproc:// can for performance motivated cases even use another "co-locally-isolated" private Context(0)-instance, having indeed zero IO-threads at all ) and enjoy the due separation of concerns with minimum adverse effects on principally lost thread-safety ( if not doing so ) and performance.

Ad 1 + 4 )

( CZMQ/3.0.1 API docs ) zthread - working with system threads (deprecated) ... The zthread class wraps OS thread creation. It creates detached threads that look like normal OS threads, or attached threads that share the caller's ØMQ context, and get an inproc pipe to talk back to the parent thread. Detached threads create their own ØMQ contexts as needed. NOTE: this class is deprecated in favor of zactor.

One may better check the version creeps among native API used, the binding / wrapper version and documentation.

Yet, Zero sharing Zen may lead your steps ( if selected language binding permits one to remain free in design decisions - reading the original design motivations always helps understand the performance and safety insights from The Original ).

Question:

Problem: I have a number of file uploads coming via HTTP in parallel ( uploads receiver ). I'm storing them temporarily on a local disk. Another process ( uploads submitter ) gets notified about new uploads and does specific processing ( parsing, extracting metadata, uploading to S3 etc ). Once upload processing done I want uploads receiver to be notified by submitter to reply back with status ( whether submission is ok or error ) to the remote uploader. Using ZeroMQ PUB/SUB pattern, what would be better:

  • subscribe all upload receiver threads to a single topic. Each receiver thread would have to filter messages based on upload id or something to find a notification that belongs to it.
  • subscribe each receiver thread to a new topic which represents particular upload. This one seems more reasonable assuming topics are cheap in ZeroMQ, i.e. not much resources is needed to keep them and they can be auto-expired. I expect new uploads to come at dozens of files per second, single upload processing may take up to several seconds so theoretically I can have up to thousand of topics active at the same moment of time. Also I may not always be able to unsubscribe due to various failure modes.

Answer:

Initial notice: On Using Different ZeroMQ Version Numbers:

While more recent versions may use PUB-side topic filtering, the early ZeroMQ versions did use SUB-side approach, which means that all the ( network ) message-transport traffic goes to all SUB-s as an acceptable penalty for distributing the processing-workload, that would otherwise be needed to get handled at lowest possible latency on the PUB-side.

This is important for cases, where in an open distributed system association the homogenity of versions is not enforceable.

Whereas you design architecture seems to be co-located on the same <localhost> the performance impact remains non-distributed ( concentrated ) and may implicate just some limited latency/priority tweaking, if overall bottleneck appears during this Use-Case up-scaling.


On Scaleability Ranges - Limits are still farther than your Use-Case:

As Martin Sustrik ( ZeroMQ co-father ) presented in details, ZeroMQ was designed with expected scales up to some small tens of thousands:

(cit.:) " Efficient Subscription Matching In ZeroMQ, simple tries are used to store and match PUB/SUB subscriptions. The subscription mechanism was intended for up to 10,000 subscriptions where simple trie works well. However, there are users who use as much as 150,000,000 subscriptions. In such cases there's a need for a more efficient data structure. "

Further details on design & scaling might be found interesting in this Martin's post.


The Best Next Step?

A fair approach would be to mock-up each of the questioned approaches and benchmark them, scaled to { 1.0x , 1.5x, 2.0x, 5.0x } of the expected static scales in-vitro to have quantitatively supported data about real overheads, performance and latencies relevant to the alternative strategies under review.

Anyway, Vovan, enjoy the worlds of smart signalling/messaging in the distributed processing.

Question:

The documentation for Socket#recv() reads:

Returns: [...] null on error.

How can I tell what the error was? I want to handle EAGAIN specifically.


Answer:

I have very limited knowledge here but from the looks of it, the answer could be: "If Socket#recv() returns null and no ZMQException was thrown, an EAGAIN error occurred."

I followed the method calls and arrived at do_read in Socket.cpp where it gets interesting on line 83:

rc = zmq_recv (socket, message, flags);
int err = zmq_errno();
if (rc < 0 && err == EAGAIN) {
    rc = zmq_msg_close (message);
    err = zmq_errno();
    if (rc != 0) {
        raise_exception (env, err);
        return NULL;
    }
    return NULL;
}
if (rc < 0) {
    raise_exception (env, err);
    rc = zmq_msg_close (message);
    err = zmq_errno();
    if (rc != 0) {
        raise_exception (env, err);
        return NULL;
    }
    return NULL;
}
return message;

What I read here is that if something goes wrong, you get an ZMQException in Java unless the error was EAGAIN and zmq_msg_close does not go wrong (I am not sure what zmq_msg_close does, but I assume it rarely goes wrong). But I don't have the environment to test this and I also don't really understand how raise_exception works (source in util.cpp): what happens if two exceptions are raised/thrown in the same code-path (e.g. when err is not EAGAIN and rc < 0) and you can only catch one runtime-exception in Java?

On a side note, support for the EAGAIN error code was added in this commit on May 15, 2015.

Question:

I'm trying to use Java client with ZeroMQ. When subscribing to any prefix, the Java client matches no messages, although a similar Python client matches messages as expected.

The Python server

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://*:5556")

for i in range(100):
    r = "XXX " + i
    socket.send_string(r)

    time.sleep(random.randint(0,10))

The Python client working fine

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://localhost:5556")

zip_filter = "XXX"
socket.setsockopt_string(zmq.SUBSCRIBE, zip_filter)

for update_nbr in range(5):
    s = socket.recv_string()
    print(s)

The Java client matching no messages

context = ZMQ.context(1);
subscriber = context.socket(ZMQ.SUB);
subscriber.connect("tcp://localhost:5556");

String filter = "XXX";
subscriber.subscribe(filter.getBytes(Charset.forName("UTF-8")));
while (true) {
  String msg = subscriber.recvStr(0, Charset.forName("UTF-8"));
  // ...
}

Using the above Python server, the Python client matches all messages starting with XXX as expected.

Using the same Python server, the Java client matches no messages.

Do you have any idea what is wrong with the call to subscribe() in the Java client?


Answer:

Ok, so I've recreated your configuration and sadly, everything works fine - both in python and java. (here's the proof) )

Java code:

public class Client {

    public static void main(String[] args) {
        final Context context = context(1);
        final Socket subscriber = context.socket(SUB);
        subscriber.connect("tcp://localhost:5556");

        String filter = "XXX";
        subscriber.subscribe(filter.getBytes(Charset.forName("UTF-8")));
        while (true) {
            String msg = subscriber.recvStr();
            System.out.println(msg);
        }
    }
}

Maven dependency:

<dependency>
    <groupId>org.zeromq</groupId>
    <artifactId>jeromq</artifactId>
    <version>0.3.4</version>
</dependency>

zeromq version: 4.1.0

What version of jeromq do you use? I don't even have a method recvStr(int, Java.nio.charset.Charset).

Question:

I am using pub/Sub Socket and currently the server subscribe byte[0] (all topics) while client subscribe byte[16] - a specific header as topic

However, I cannot stop client to subscribe byte[0] which can receive all other messages.

My application is a like a app game which has one single server using ZMQ as connection and many clients have a ZMQ sockets to talk with server.

What pattern or socket I should use in this case?

Thanks


Answer:

" ... cannot stop client to subscribe byte[0] which can receive all other messages."


Stopping a "subscribe to all" mode of the SUB client

For the ZMQ PUB/SUB Formal Communication Pattern archetype, the SUB client has to submit it's subscription request ( via zmq_setsockopt() ).

PUB-side ( a Game Server ) has got no option to do that from it's side.

There is no-subscription state right on a creation of a new SUB socket, thus an absolutely restrictive filter, thas no message pass through. ( For furhter details on methods for SUBSCRIBE / UNSUBSCRIBE ref. below )


ZeroMQ specification details setting for this:
int zmq_setsockopt (       void   *socket,
                           int     option_name,
                     const void   *option_value,
                           size_t  option_len
                     );

Caution: only ZMQ_SUBSCRIBE
              ZMQ_UNSUBSCRIBE
              ZMQ_LINGER
         take effect immediately,
         other options are active only for subsequent socket bind/connects.

ZMQ_SUBSCRIBE: Establish message filter

The ZMQ_SUBSCRIBE option shall establish a new message filter on a ZMQ_SUB socket. Newly created ZMQ_SUB sockets shall filter out all incoming messages, therefore you should call this option to establish an initial message filter.

An empty option_value of length zero shall subscribe to all incoming messages.

A non-empty option_value shall subscribe to all messages beginning with the specified prefix.

Multiple filters may be attached to a single ZMQ_SUB socket, in which case a message shall be accepted if it matches at least one filter.

ZMQ_UNSUBSCRIBE: Remove message filter

The ZMQ_UNSUBSCRIBE option shall remove an existing message filter on a ZMQ_SUB socket. The filter specified must match an existing filter previously established with the ZMQ_SUBSCRIBE option. If the socket has several instances of the same filter attached the ZMQ_UNSUBSCRIBE option shall remove only one instance, leaving the rest in place and functional.


How to enforce an ad-hoc, server-dictated, ZMQ_SUBSCRIBE restrictions?

This is possible via extending the messaging layer and adding a control-mode socket, that will carry server-initiated settings for the client ZMQ_SUB messages filtering.

Upon receiving a new, the server-dictated, ZMQ_SUBSCRIBE/ZMQ_UNSUBSCRIBE setting, the ZMQ_SUB client side code will simply handle that request and add zmq_setsockopt() accordingly.

FSA-driven grammars for this approach are rich of further possibilites, so will allow any Game Server / Game Community to smoothly go this way.


What pattern or socket I should use?

ZeroMQ is rather a library of LEGO-style elements to get assembled into a bigger picture.

Expecting such a smart library to have a one-size-fits-all ninja-element is on a closer look an oxymoron.

So, to avoid a "Never-ending-story" of adding "although this ... and also that ..."

  1. Review all requirements and & list features for the end-to-end scaleable solution,

  2. Design a messaging concept & validate it to meet all the listed requirements & cover all features in [1]

  3. Implement [2]

  4. Test [3] & correct it for meeting 1:1 the end-to-end specification [1]

  5. Enjoy it. You have done it end-to-end right.