def setup(self, socket_type, complete_or_error_queue):
        """Setup the asyncio event loop.

        Args:
            socket_type (int from zmq.*): One of zmq.DEALER or zmq.ROUTER
            complete_or_error_queue (queue.Queue): A way to propagate errors
                back to the calling thread. Needed since this function is
                directly used in Thread.

        Returns:
            None
        """
        try:
            if self._secured:
                if self._server_public_key is None or \
                        self._server_private_key is None:
                    raise LocalConfigurationError(
                        "Attempting to start socket in secure mode, "
                        "but complete server keys were not provided")

            self._event_loop = zmq.asyncio.ZMQEventLoop()
            asyncio.set_event_loop(self._event_loop)
            self._context = zmq.asyncio.Context()
            self._socket = self._context.socket(socket_type)

            self._socket.set(zmq.TCP_KEEPALIVE, 1)
            self._socket.set(zmq.TCP_KEEPALIVE_IDLE, self._connection_timeout)
            self._socket.set(zmq.TCP_KEEPALIVE_INTVL, self._heartbeat_interval)

            if socket_type == zmq.DEALER:
                self._socket.identity = "{}-{}".format(
                    self._zmq_identity,
                    hashlib.sha512(uuid.uuid4().hex.encode()
                                   ).hexdigest()[:23]).encode('ascii')

                if self._secured:
                    # Generate ephemeral certificates for this connection

                    public_key, secretkey = zmq.curve_keypair()
                    self._socket.curve_publickey = public_key
                    self._socket.curve_secretkey = secretkey
                    self._socket.curve_serverkey = self._server_public_key

                self._socket.connect(self._address)
            elif socket_type == zmq.ROUTER:
                if self._secured:
                    auth = AsyncioAuthenticator(self._context)
                    self._auth = auth
                    auth.start()
                    auth.configure_curve(domain='*',
                                         location=zmq.auth.CURVE_ALLOW_ANY)

                    self._socket.curve_secretkey = self._server_private_key
                    self._socket.curve_publickey = self._server_public_key
                    self._socket.curve_server = True

                try:
                    self._socket.bind(self._address)
                except zmq.error.ZMQError as e:
                    raise LocalConfigurationError(
                        "Can't bind to {}: {}".format(self._address,
                                                      str(e)))
                else:
                    LOGGER.info("Listening on %s", self._address)

            self._dispatcher.add_send_message(self._connection,
                                              self.send_message)
            self._dispatcher.add_send_last_message(self._connection,
                                                   self.send_last_message)

            asyncio.ensure_future(self._remove_expired_futures(),
                                  loop=self._event_loop)

            asyncio.ensure_future(self._receive_message(),
                                  loop=self._event_loop)

            asyncio.ensure_future(self._dispatch_message(),
                                  loop=self._event_loop)

            self._dispatcher_queue = asyncio.Queue()

            if self._monitor:
                self._monitor_fd = "inproc://monitor.s-{}".format(
                    _generate_id()[0:5])
                self._monitor_sock = self._socket.get_monitor_socket(
                    zmq.EVENT_DISCONNECTED,
                    addr=self._monitor_fd)
                asyncio.ensure_future(self._monitor_disconnects(),
                                      loop=self._event_loop)

        except Exception as e:
            # Put the exception on the queue where in start we are waiting
            # for it.
            complete_or_error_queue.put_nowait(e)
            self._close_sockets()
            raise

        if self._heartbeat:
            asyncio.ensure_future(self._do_heartbeat(), loop=self._event_loop)

        # Put a 'complete with the setup tasks' sentinel on the queue.
        complete_or_error_queue.put_nowait(_STARTUP_COMPLETE_SENTINEL)

        asyncio.ensure_future(self._notify_started(), loop=self._event_loop)

        self._event_loop.run_forever()
        # event_loop.stop called elsewhere will cause the loop to break out
        # of run_forever then it can be closed and the context destroyed.
        self._event_loop.close()
        self._close_sockets()
def setup ( self , socket_type , complete_or_error_queue ) : try : if self . _secured : if self . _server_public_key is None or self . _server_private_key is None : raise LocalConfigurationError ( "Attempting to start socket in secure mode, " "but complete server keys were not provided" ) self . _event_loop = zmq . asyncio . ZMQEventLoop ( ) asyncio . set_event_loop ( self . _event_loop ) self . _context = zmq . asyncio . Context ( ) self . _socket = self . _context . socket ( socket_type ) self . _socket . set ( zmq . TCP_KEEPALIVE , 1 ) self . _socket . set ( zmq . TCP_KEEPALIVE_IDLE , self . _connection_timeout ) self . _socket . set ( zmq . TCP_KEEPALIVE_INTVL , self . _heartbeat_interval ) if socket_type == zmq . DEALER : self . _socket . identity = "{}-{}" . format ( self . _zmq_identity , hashlib . sha512 ( uuid . uuid4 ( ) . hex . encode ( ) ) . hexdigest ( ) [ : 23 ] ) . encode ( 'ascii' ) if self . _secured : public_key , secretkey = zmq . curve_keypair ( ) self . _socket . curve_publickey = public_key self . _socket . curve_secretkey = secretkey self . _socket . curve_serverkey = self . _server_public_key self . _socket . connect ( self . _address ) elif socket_type == zmq . ROUTER : if self . _secured : auth = AsyncioAuthenticator ( self . _context ) self . _auth = auth auth . start ( ) auth . configure_curve ( domain = '*' , location = zmq . auth . CURVE_ALLOW_ANY ) self . _socket . curve_secretkey = self . _server_private_key self . _socket . curve_publickey = self . _server_public_key self . _socket . curve_server = True try : self . _socket . bind ( self . _address ) except zmq . error . ZMQError as e : raise LocalConfigurationError ( "Can't bind to {}: {}" . format ( self . _address , str ( e ) ) ) else : LOGGER . info ( "Listening on %s" , self . _address ) self . _dispatcher . add_send_message ( self . _connection , self . send_message ) self . _dispatcher . add_send_last_message ( self . _connection , self . send_last_message ) asyncio . ensure_future ( self . _remove_expired_futures ( ) , loop = self . _event_loop ) asyncio . ensure_future ( self . _receive_message ( ) , loop = self . _event_loop ) asyncio . ensure_future ( self . _dispatch_message ( ) , loop = self . _event_loop ) self . _dispatcher_queue = asyncio . Queue ( ) if self . _monitor : self . _monitor_fd = "inproc://monitor.s-{}" . format ( _generate_id ( ) [ 0 : 5 ] ) self . _monitor_sock = self . _socket . get_monitor_socket ( zmq . EVENT_DISCONNECTED , addr = self . _monitor_fd ) asyncio . ensure_future ( self . _monitor_disconnects ( ) , loop = self . _event_loop ) except Exception as e : complete_or_error_queue . put_nowait ( e ) self . _close_sockets ( ) raise if self . _heartbeat : asyncio . ensure_future ( self . _do_heartbeat ( ) , loop = self . _event_loop ) complete_or_error_queue . put_nowait ( _STARTUP_COMPLETE_SENTINEL ) asyncio . ensure_future ( self . _notify_started ( ) , loop = self . _event_loop ) self . _event_loop . run_forever ( ) self . _event_loop . close ( ) self . _close_sockets ( )
Setup the asyncio event loop.

        Args:
            socket_type (int from zmq.*): One of zmq.DEALER or zmq.ROUTER
            complete_or_error_queue (queue.Queue): A way to propagate errors
                back to the calling thread. Needed since this function is
                directly used in Thread.

        Returns:
            None