How to create ZeroMQ socket suitable both for sending and consuming?

Question

Could you please advice an ZeroMQ socket(s) architecture for the following scenario:

1) there is server listening on port

2) there are several clients connecting server simultaneously

3) server accept all connections from clients and provide bi-directional queue for each client, means both party (client N or server) can send or consume messages, i.e. both party can be INITIATOR of the communication and other party should have a callback to process the message.

Should we create additional ZeroMQ socket on each accepted connection for pushing messages from server? Could you please advice of which ZeroMQ socket type to google for such architecture?

Answer 1

Q : …create additional ZeroMQ socket on each accepted connection for pushing messages from server?

The best trivial composition-based design - both scaling-wise and security-wise

The native ZeroMQ primitives ( the smart primitive Scalable Formal Communications Pattern Archetypes ) are like a LEGO building-blocks for us - we compose their further use on the Messaging/Signalling Plane of the intended target use in our application-domain.

Q : Could you please advice of which ZeroMQ socket type to google for such architecture?

No, given no detailed requirements list was available for such an advice. A pair of PUSH/PULL-s need not suffice on their own, temporarily conducted ( episodic ) REQ/REP may help for the client-(re-)discovery phases, as might other co-existent, persistent or episodic Archetypes do for composing any additional System/Service-Planes.

Answer 2

I encountered basically the same problem. For me the issue seems to be that even in python with GIL's protection, you cannot operate one socket from two threads at a time. This will cause interpreter crash. Thus one cannot have send- and receive- threads, but one thread, which would then have to be able to receive and send simultaneously while pyzmq provides no way to poll() a socket and non-socket at the same time.

user3666197 replied to my question too, but neither of the replies are helpful at all. Not least for the reason that one cannot assume the 1000+ free and open ports for generating host sockets for all client connections.

There is a relatively ugly solution. This is a draft that lacks appropriate NOWAIT-flags etc. make_process_pull_socket() makes a socket for communication inside the process. As follows:

class TwoWay(threading.Thread):
def __init__(self, ip_string, port, inque:queue.Queue):
    super(TwoWay, self).__init__()
    self.sock = cntxt.socket(zmq.ROUTER)
    addr = "tcp://{}:{}".format(ip_string, str(port))
    self.sock.bind(addr)
    self.pull = make_process_pull_socket(4456)
    self.push = make_process_push_socket(4456)
    self.inque = inque

def run(self):
    pl = zmq.Poller()
    pl.register(self.sock,zmq.POLLIN)
    pl.register(self.pull,zmq.POLLIN)
    while True:
        try:
            p = pl.poll(timeout=2000)
            if not p:
                continue
            s,i = p[0]
            if s == self.sock:
                a,m = s.recv_multipart()
                self.inque.put((a,m))
            if s == self.pull:
                r = s.recv()
                if len(r) > 5:
                    self.sock.send_multipart([r[:5], r[5:]])
        except Exception as e:
            print(e)

And then you get incoming messages into the inque-queue provided as argument, and outgoing messages (with identifiers) can be sent with TwoWay.push.send(id+msg)