Send python multiprocessing connection via ØMQ

You have to use connection owner's process (producer) authkey to rebuild the handle on the receiver. You may send producer's authkey along with the pickled connection over ØMQ.

producer.py

import zmq
import time
from multiprocessing import Pipe, current_process
from multiprocessing.reduction import reduce_connection

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

time.sleep(1)

a, b = Pipe()
reduced_a = reduce_connection(a)
socket.send_pyobj(reduced_a)
socket.send(current_process().authkey)

print b.recv()

receiver.py

import zmq
from multiprocessing import reduction

def rebuild_handle_with_key(pickled_data, authkey=None):
    from multiprocessing import current_process
    from multiprocessing.util import sub_debug
    from multiprocessing.connection import Client
    from multiprocessing.reduction import recv_handle
    import os
    address, handle, inherited = pickled_data
    if inherited:
        return handle
    sub_debug('rebuilding handle %d', handle)
    conn = Client(address, authkey=authkey or current_process().authkey)
    conn.send((handle, os.getpid()))
    new_handle = recv_handle(conn)
    conn.close()
    return new_handle

def rebuild_connection_with_key(reduced_handle, readable, writable, authkey=None):
    import _multiprocessing
    handle = rebuild_handle_with_key(reduced_handle, authkey)
    return _multiprocessing.Connection(
        handle, readable=readable, writable=writable
        )

reduction.rebuild_connection = rebuild_connection_with_key

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

obj = socket.recv_pyobj()
authkey = socket.recv()
answer_conn = obj[0](obj[1][0],obj[1][1],obj[1][2], authkey)
# answer_conn = obj[0](*obj[1])

answer_conn.send('All is OK!')

I couldn't find a better way to make multiprocessing.reduction use a different authkey so here I have patched it.