Python zmq 模块，Again() 实例源码

def start(self, socket):
        """
        Start the monitoring thread and socket.
        :param socket: Socket to monitor.
        """
        # Start a thread only if it is not already running.
        if self.monitor_listening.is_set():
            return

        # Setup monitor socket.
        monitor_socket = socket.get_monitor_socket(events=self.events)
        monitor_socket.setsockopt(zmq.RCVTIMEO, self.receive_timeout)
        self.monitor_listening.set()

        def event_listener(monitor_listening):
            while monitor_listening.is_set():
                try:
                    event = recv_monitor_message(monitor_socket)
                    # The socket is closed, just stop listening now.
                    if event["event"] == zmq.EVENT_CLOSED:
                        monitor_listening.clear()

                    self._notify_listeners(event)
                # In case the receive cannot be completed before the timeout.
                except zmq.Again:
                    # Heartbeat for listeners - we do not need an additional thread for time based listeners.
                    self._notify_listeners(None)

            # Cleanup monitor socket.
            socket.disable_monitor()
            monitor_socket.close()

        self.monitor_thread = threading.Thread(target=event_listener, args=(self.monitor_listening,))
        # In case someone does not call disconnect, this will stop the thread anyway.
        self.monitor_thread.daemon = True
        self.monitor_thread.start()

def send(self, message, send_more=False, block=True, as_json=False):

        flags = 0
        if send_more:
            flags = zmq.SNDMORE
        if not block:
            flags = flags | zmq.NOBLOCK

        try:
            if as_json:
                self.socket.send_json(message, flags)
            else:
                self.socket.send(message, flags, copy=self.zmq_copy, track=self.zmq_track)
        except zmq.Again as e:
            if not block:
                pass
            else:
                raise e
        except zmq.ZMQError as e:
            logger.error(sys.exc_info()[1])
            raise e

def forward(self, data):

        try:
            # self.logger.debug('sending message')
            self.list_communication_channel.send(data)
            # self.logger.debug('ok with the message')
        except zmq.NotDone:
            # time.sleep(TRY_TIMEOUT)
            self.logger.debug('my recipient is dead, not done')
            self.list_communication_channel.close()
        except zmq.Again:
            self.logger.debug('my recipient is dead')
            # self.list_communication_channel.close()
            raise zmq.Again
        except zmq.ZMQError as a:
            self.logger.debug("Error in message forward " + a.strerror)
            self.context.destroy()
            self.context = zmq.Context()

def test_disconnection(self):
        """ Test the disconnection of subscribers. """
        from supvisors.utils import InternalEventHeaders
        # get the local address
        local_address = self.supvisors.address_mapper.local_address
        # test remote disconnection
        address = next(address
                       for address in self.supvisors.address_mapper.addresses
                       if address != local_address)
        self.subscriber.disconnect([address])
        # send a tick event from the local publisher
        payload = {'date': 1000}
        self.publisher.send_tick_event(payload)
        # check the reception of the tick event
        msg = self.receive('Tick')
        self.assertTupleEqual((InternalEventHeaders.TICK,
                               local_address, payload), msg)
        # test local disconnection
        self.subscriber.disconnect([local_address])
        # send a tick event from the local publisher
        self.publisher.send_tick_event(payload)
        # check the non-reception of the tick event
        with self.assertRaises(zmq.Again):
            self.subscriber.receive()

def generator_from_zmq_pull(context, host):
    socket = context.socket(zmq.PULL)
    # TODO: Configure socket with clean properties to avoid message overload.
    if host.endswith('/'):
        host = host[:-1]
    print_item("+", "Binding ZMQ pull socket : " + colorama.Fore.CYAN + "{0}".format(host) + colorama.Style.RESET_ALL)
    socket.bind(host)

    while True:
        try:
            message = socket.recv(flags=zmq.NOBLOCK)
        except zmq.Again as e:
            message = None
        if message is None:
            yield None # NOTE: We have to make the generator non blocking.
        else:
            task = json.loads(message)
            yield task

def get_messages(self, timeout=0.1, count=1):
        started = time()
        sleep_time = timeout / 10.0
        while count:
            try:
                msg = self.subscriber.recv_multipart(copy=True, flags=zmq.NOBLOCK)
            except zmq.Again:
                if time() - started > timeout:
                    break
                sleep(sleep_time)
            else:
                partition_seqno, global_seqno = unpack(">II", msg[2])
                seqno = global_seqno if self.count_global else partition_seqno
                if not self.counter:
                    self.counter = seqno
                elif self.counter != seqno:
                    if self.seq_warnings:
                        self.logger.warning("Sequence counter mismatch: expected %d, got %d. Check if system "
                                            "isn't missing messages." % (self.counter, seqno))
                    self.counter = None
                yield msg[1]
                count -= 1
                if self.counter:
                    self.counter += 1
                self.stats[self.stat_key] += 1

def _get_data(self, blocking=True):
        """Get batch of data."""
        # TODO complete docstring.

        if not blocking:
            try:
                batch = self.socket.recv(flags=zmq.NOBLOCK)
            except zmq.Again:
                return None
        else:
            batch = self.socket.recv()

        if batch == TERM_MSG:
            raise EOCError()

        if self.structure == 'array':
            batch = numpy.fromstring(batch, dtype=self.dtype)
            batch = numpy.reshape(batch, self.shape)
        elif self.structure == 'dict':
            batch = json.loads(batch)
        elif self.structure == 'boolean':
            batch = bool(batch)

        return batch

def _receiveFromListener(self, quota) -> int:
        """
        Receives messages from listener
        :param quota: number of messages to receive
        :return: number of received messages
        """
        assert quota
        i = 0
        while i < quota:
            try:
                ident, msg = self.listener.recv_multipart(flags=zmq.NOBLOCK)
                if not msg:
                    # Router probing sends empty message on connection
                    continue
                i += 1
                if self.onlyListener and ident not in self.remotesByKeys:
                    self.peersWithoutRemotes.add(ident)
                self._verifyAndAppend(msg, ident)
            except zmq.Again:
                break
        if i > 0:
            logger.trace('{} got {} messages through listener'.
                         format(self, i))
        return i

def test_retry_recv(self):
        pull = self.socket(zmq.PULL)
        pull.rcvtimeo = self.timeout_ms
        self.alarm()
        self.assertRaises(zmq.Again, pull.recv)
        assert self.timer_fired

def test_retry_send(self):
        push = self.socket(zmq.PUSH)
        push.sndtimeo = self.timeout_ms
        self.alarm()
        self.assertRaises(zmq.Again, push.send, b('buf'))
        assert self.timer_fired

def test_again(self):
        s = self.context.socket(zmq.REP)
        self.assertRaises(Again, s.recv, zmq.NOBLOCK)
        self.assertRaisesErrno(zmq.EAGAIN, s.recv, zmq.NOBLOCK)
        s.close()

def start_listener(self):
        print('ZMQ listener started')
        while True:
            try:
                self.s.recv(zmq.NOBLOCK)  # note NOBLOCK here
            except zmq.Again:
                # no message to recv, do other things
                time.sleep(0.05)
            else:
                self.on_q.put(ON_SIGNAL)

def client_behavior(settings, logger):

    internal_channel = InternalChannel(addr="127.0.0.1", port=settings.getIntPort(), logger=logger)

    try:
        internal_channel.generate_internal_channel_client_side()
    except ZMQError as e:
        logger.debug(e)

    message = Message()
    message.priority = ALIVE
    message.source_flag = INT
    message.source_id = '1'
    message.target_id = '1'
    message.target_addr = '192.168.1.1'
    message.target_key = '{}:{}'.format(0, 19)

    internal_channel.send_first_internal_channel_message(dumps(message))

    msg = internal_channel.wait_int_message(dont_wait=False)

    logger.debug("msg : " + msg)

    external_channel = ExternalChannel(addr="127.0.0.1", port=settings.getExtPort(), logger=logger)
    external_channel.generate_external_channel_client_side()
    external_channel.external_channel_subscribe()

    logger.debug(loads(external_channel.wait_ext_message()).printable_message())

    logger.debug("try_to_connect TEST COMPLETED")

    stop = False
    while not stop:
        try:
            logger.debug(loads(external_channel.wait_ext_message()).printable_message())
            sleep(1)
        except Again:
            logger.debug("my master is DEAD")
            stop = True

def server_behavior(settings, logger):

    internal_channel = InternalChannel(addr="127.0.0.1", port=settings.getIntPort(), logger=logger)

    try:
        internal_channel.generate_internal_channel_server_side()
        msg = loads(internal_channel.wait_int_message(dont_wait=False))
        logger.debug("msg : ")
        logger.debug(msg.printable_message())
        internal_channel.reply_to_int_message(OK)
    except ZMQError as e:
        logger.debug(e)

    external_channel = ExternalChannel(addr="127.0.0.1", port=settings.getExtPort(), logger=logger)

    external_channel.generate_external_channel_server_side()
    external_channel.external_channel_publish()

    message = Message()
    message.priority = ALIVE
    message.source_flag = EXT
    message.source_id = '1'
    message.target_id = '1'
    message.target_addr = '192.168.1.1'
    message.target_key = '{}:{}'.format(0, 19)

    sleep(1)

    external_channel.forward(dumps(message))

    logger.debug("try_to_connect TEST COMPLETED")

    stop = False
    while not stop:
        try:
            external_channel.forward(dumps(message))
            sleep(1)
        except zmq.Again:
            stop = True

def wait_int_message(self, dont_wait=True):
        if dont_wait:
            # wait for internal message
            try:
                msg = self.list_communication_channel.recv(zmq.DONTWAIT)
                return msg
            except zmq.Again:
                raise zmq.Again
        else:
            self.logger.debug('waiting for a request')
            msg = self.list_communication_channel.recv()
            return msg

def receive(self, event_type):
        """ This method performs a checked reception on the subscriber. """
        try:
            self.subscriber.socket.poll(1000)
            return self.subscriber.receive()
        except zmq.Again:
            self.fail('Failed to get {} event'.format(event_type))

def receive(self, event_type):
        """ This method performs a checked reception on the puller. """
        try:
            return self.puller.receive()
        except zmq.Again:
            self.fail('Failed to get {} request'. format(event_type))

def check_reception(self, header=None, data=None):
        """ The method tests that the message is received correctly
        or not received at all. """
        if header and data:
            # check that subscriber receives the message
            try:
                msg = self.subscriber.receive()
            except zmq.Again:
                self.fail('Failed to get {} status'.format(header))
            self.assertTupleEqual((header, data), msg)
        else:
            # check the non-reception of the Supvisors status
            with self.assertRaises(zmq.Again):
                self.subscriber.receive()

def _send_and_receive(self, message):
    """Sending payloads to NM and returning Response instances.

    Or, if the action failed, an error will be raised during the instantiation
    of the Response.  Can also timeout if the socket receives no data for some
    period.

    Args:
      message: dict of a message to send to NM

    Returns:
      Response instance if the request succeeded

    Raises:
      TimeoutError: if nothing is received for the timeout
    """
    # zmq is thread unsafe: if we send a second request before
    # we get back the first response, we throw an exception
    # fix that -kheimerl
    with self.lock:
      # Send the message and poll for responses.
      self.socket.send(json.dumps(message))
      responses = self.socket.poll(timeout=self.socket_timeout * 1000)
      if responses:
        try:
          raw_response_data = self.socket.recv()
          return Response(raw_response_data)
        except zmq.Again:
          pass
      # If polling fails or recv failes, we reset the socket or
      # it will be left in a bad state, waiting for a response.
      self.socket.close()
      self.setup_socket()
      self.socket.connect(self.address)
      raise TimeoutError('did not receive a response')

def handle_in(self):
        try:
            tmp = self.socket.recv_multipart()
        except zmq.Again:
            return
        if len(tmp) != 2:
            self.logger.critical('Received a msg with len != 2, something seriously wrong. ')
            return

        sender, msg_buf = tmp
        msg = msg_factory(msg_buf)

        data = self.controllers.get(sender)
        if not data:
            self.logger.critical('Received a msg from %s - this is an unknown sender' % sender)
            return
        data['last_seen'] = time.time()
        # self.logger.debug('Received from %s' % sender)

        # TODO Notify Controllers that we are busy, no more messages to be sent
        # The above busy notification is not perfect as other messages might be on their way already
        # but for long-running queries it will at least ensure other controllers
        # don't try and overuse this node by filling up a queue
        busy_msg = BusyMessage()
        self.send_to_all(busy_msg)

        try:
            tmp = self.handle(msg)
        except Exception, e:
            tmp = ErrorMessage(msg)
            tmp['payload'] = traceback.format_exc()
            self.logger.exception(tmp['payload'])
        if tmp:
            self.send(sender, tmp)

        self.send_to_all(DoneMessage())  # Send a DoneMessage to all controllers, this flags you as 'Done'. Duh

def run(self):
        while True:
            try:
                message = self.pull.recv(flags=zmq.NOBLOCK)
            except zmq.Again as e:
                message = None
            if message is not None:
                task = json.loads(message)
                self.redis.setex(
                    task['transaction'],
                    self.result_expiration,
                    json.dumps(task['data'])
                )

def test_again(self):
        s = self.context.socket(zmq.REP)
        self.assertRaises(Again, s.recv, zmq.NOBLOCK)
        self.assertRaisesErrno(zmq.EAGAIN, s.recv, zmq.NOBLOCK)
        s.close()

def test_again(self):
        s = self.context.socket(zmq.REP)
        self.assertRaises(Again, s.recv, zmq.NOBLOCK)
        self.assertRaisesErrno(zmq.EAGAIN, s.recv, zmq.NOBLOCK)
        s.close()

def test_again(self):
        s = self.context.socket(zmq.REP)
        self.assertRaises(Again, s.recv, zmq.NOBLOCK)
        self.assertRaisesErrno(zmq.EAGAIN, s.recv, zmq.NOBLOCK)
        s.close()

def test_again(self):
        s = self.context.socket(zmq.REP)
        self.assertRaises(Again, s.recv, zmq.NOBLOCK)
        self.assertRaisesErrno(zmq.EAGAIN, s.recv, zmq.NOBLOCK)
        s.close()

def test_again(self):
        s = self.context.socket(zmq.REP)
        self.assertRaises(Again, s.recv, zmq.NOBLOCK)
        self.assertRaisesErrno(zmq.EAGAIN, s.recv, zmq.NOBLOCK)
        s.close()

def send_raw_msg (self, msg):

        tries = 0
        while True:
            try:
                self.socket.send(msg)
                break
            except zmq.Again:
                tries += 1
                if tries > 5:
                    self.disconnect()
                    return RC_ERR("*** [RPC] - Failed to send message to server")


        tries = 0
        while True:
            try:
                response = self.socket.recv()
                break
            except zmq.Again:
                tries += 1
                if tries > 5:
                    self.disconnect()
                    return RC_ERR("*** [RPC] - Failed to get server response at {0}".format(self.transport))


        return response



    # processs a single response from server

def receive(self):
        '''
        Return the message received.

        ..note::
            In ZMQ we are unable to get the address where we got the message from.
        '''
        try:
            msg = self.sub.recv()
        except zmq.Again as error:
            log.error('Unable to receive messages: %s', error, exc_info=True)
            raise ListenerException(error)
        log.debug('[%s] Received %s', time.time(), msg)
        return msg, ''

def test_process_does_not_block(self):
        mock_socket = Mock()
        mock_socket.recv_unicode.side_effect = zmq.Again()

        server = ControlServer(None, connection_string='127.0.0.1:10000')
        server._socket = mock_socket
        assertRaisesNothing(self, server.process)

        mock_socket.recv_unicode.assert_has_calls([call(flags=zmq.NOBLOCK)])

def process(self, blocking=False):
        """
        Each time this method is called, the socket tries to retrieve data and passes
        it to the JSONRPCResponseManager, which in turn passes the RPC to the
        ExposedObjectCollection.

        In case no data are available, the method does nothing. This behavior is required for
        Lewis where everything is running in one thread. The central loop can call process
        at some point to process remote calls, so the RPC-server does not introduce its own
        infinite processing loop.

        If the server has not been started yet (via :meth:`start_server`), a RuntimeError
        is raised.

        :param blocking: If True, this function will block until it has received data or a timeout
                         is triggered. Default is False to preserve behavior of prior versions.
        """
        if self._socket is None:
            raise RuntimeError('The server has not been started yet, use start_server to do so.')

        try:
            request = self._socket.recv_unicode(flags=zmq.NOBLOCK if not blocking else 0)

            self.log.debug('Got request %s', request)

            try:
                response = JSONRPCResponseManager.handle(request, self._exposed_object)
                self._socket.send_unicode(response.json)

                self.log.debug('Sent response %s', response.json)
            except TypeError as e:
                self._socket.send_json(
                    self._unhandled_exception_response(json.loads(request)['id'], e))
        except zmq.Again:
            pass

def receive_message(socket, blocking=True):
        flags = 0 if blocking else zmq.NOBLOCK
        try:
            cmd, data = socket.recv_multipart(flags=flags)
            return cmd, data
        except zmq.Again:
            return None, None
        except zmq.ContextTerminated:
            print("Context terminated ..")
            return None, None
        except KeyboardInterrupt:
            return None, None

def _receiveFromRemotes(self, quotaPerRemote) -> int:
        """
        Receives messages from remotes
        :param quotaPerRemote: number of messages to receive from one remote
        :return: number of received messages
        """

        assert quotaPerRemote
        totalReceived = 0
        for ident, remote in self.remotesByKeys.items():
            if not remote.socket:
                continue
            i = 0
            sock = remote.socket
            while i < quotaPerRemote:
                try:
                    msg, = sock.recv_multipart(flags=zmq.NOBLOCK)
                    if not msg:
                        # Router probing sends empty message on connection
                        continue
                    i += 1
                    self._verifyAndAppend(msg, ident)
                except zmq.Again:
                    break
            if i > 0:
                logger.trace('{} got {} messages through remote {}'.
                             format(self, i, remote))
            totalReceived += i
        return totalReceived

def transmit(self, msg, uid, timeout=None, serialized=False):
        remote = self.remotes.get(uid)
        err_str = None
        if not remote:
            logger.debug("Remote {} does not exist!".format(uid))
            return False, err_str
        socket = remote.socket
        if not socket:
            logger.debug('{} has uninitialised socket '
                         'for remote {}'.format(self, uid))
            return False, err_str
        try:
            if not serialized:
                msg = self.prepare_to_send(msg)
            # socket.send(self.signedMsg(msg), flags=zmq.NOBLOCK)
            socket.send(msg, flags=zmq.NOBLOCK)
            logger.debug('{} transmitting message {} to {}'
                         .format(self, msg, uid))
            if not remote.isConnected and msg not in self.healthMessages:
                logger.debug('Remote {} is not connected - '
                             'message will not be sent immediately.'
                             'If this problem does not resolve itself - '
                             'check your firewall settings'.format(uid))
            return True, err_str
        except zmq.Again:
            logger.debug(
                '{} could not transmit message to {}'.format(self, uid))
        except InvalidMessageExceedingSizeException as ex:
            err_str = '{}Cannot transmit message. Error {}'.format(
                CONNECTION_PREFIX, ex)
            logger.error(err_str)
        return False, err_str

def transmitThroughListener(self, msg, ident) -> Tuple[bool, Optional[str]]:
        if isinstance(ident, str):
            ident = ident.encode()
        if ident not in self.peersWithoutRemotes:
            logger.debug('{} not sending message {} to {}'.
                         format(self, msg, ident))
            logger.debug("This is a temporary workaround for not being able to "
                         "disconnect a ROUTER's remote")
            return False, None
        try:
            msg = self.prepare_to_send(msg)
            # noinspection PyUnresolvedReferences
            # self.listener.send_multipart([ident, self.signedMsg(msg)],
            #                              flags=zmq.NOBLOCK)
            logger.trace('{} transmitting {} to {} through listener socket'.
                         format(self, msg, ident))
            self.listener.send_multipart([ident, msg], flags=zmq.NOBLOCK)
            return True, None
        except zmq.Again:
            return False, None
        except InvalidMessageExceedingSizeException as ex:
            err_str = '{}Cannot transmit message. Error {}'.format(
                CONNECTION_PREFIX, ex)
            logger.error(err_str)
            return False, err_str
        except Exception as e:
            err_str = '{}{} got error {} while sending through listener to {}'\
                .format(CONNECTION_PREFIX, self, e, ident)
            logger.error(err_str)
            return False, err_str
        return True, None

def receive(self, handler=None, block=True):
        """
        :param handler:     Reference to a specific message handler function to use for interpreting
                            the message to be received
        :param block:       Blocking receive call
        :return:            Map holding the data, timestamp, data and main header
        """

        message = None
        # Set blocking flag in receiver
        self.receiver.block = block
        receive_is_successful = False

        if not handler:
            try:
                # Dynamically select handler
                htype = self.receiver.header()["htype"]
            except zmq.Again:
                # not clear if this is needed
                self.receiver.flush(receive_is_successful)
                return message
            except KeyboardInterrupt:
                raise
            except:
                logger.exception('Unable to read header - skipping')
                # Clear remaining sub-messages if exist
                self.receiver.flush(receive_is_successful)
                return message

            try:
                handler = receive_handlers[htype]
            except:
                logger.debug(sys.exc_info()[1])
                logger.warning('htype - ' + htype + ' -  not supported')

        try:
            data = handler(self.receiver)
            # as an extra safety margin
            if data:
                receive_is_successful = True
                message = Message(self.receiver.statistics, data)
        except KeyboardInterrupt:
            raise
        except:
            logger.exception('Unable to decode message - skipping')

        # Clear remaining sub-messages if exist
        self.receiver.flush(receive_is_successful)

        return message

def _run (self):

        # socket must be created on the same thread 
        self.socket.setsockopt(zmq.SUBSCRIBE, b'')
        self.socket.setsockopt(zmq.RCVTIMEO, 5000)
        self.socket.connect(self.tr)

        got_data = False

        self.monitor.reset()


        while self.active:
            try:

                with self.monitor:
                    line = self.socket.recv_string()

                self.monitor.on_recv_msg(line)

                self.last_data_recv_ts = time.time()

                # signal once
                if not got_data:
                    self.event_handler.on_async_alive()
                    got_data = True


            # got a timeout - mark as not alive and retry
            except zmq.Again:
                # signal once
                if got_data:
                    self.event_handler.on_async_dead()
                    got_data = False

                continue

            except zmq.ContextTerminated:
                # outside thread signaled us to exit
                assert(not self.active)
                break

            msg = json.loads(line)

            name = msg['name']
            data = msg['data']
            type = msg['type']
            baseline = msg.get('baseline', False)

            self.raw_snapshot[name] = data

            self.__dispatch(name, type, data, baseline)


        # closing of socket must be from the same thread
        self.socket.close(linger = 0)

def recv_loop(self, configfile=None):
        """
        This is the main loop receiving data and calling functions. First it calls
        the read_config function if not done previously. Afterwards it connects the
        ZeroMQ publisher.
        The reception is non-blocking. If nothing is received, the JobMonitor sleeps
        for a second. This is no problem since ZeroMQ queues the strings.
        Each loop checks whether it is time to call the update function.
        If the filter applies, it is analyzed for the status attribute and if it exists,
        the value is checked whether a function is registered for it and finally calls it.
        """
        if not self.config:
            self.read_config(configfile=configfile)
        if not self.context:
            self.connect()

        updatetime = datetime.datetime.now() + datetime.timedelta(seconds=self.interval)
        while not self.terminate:
            s = None
            try:
                s = self.socket.recv(flags=zmq.NOBLOCK)
            except zmq.Again as e:
                time.sleep(1)
            except KeyboardInterrupt:
                self.terminate = True
                pass
            if not self.terminate:
                if datetime.datetime.now() > updatetime:
                    logging.debug("Calling update function")
                    self.update()
                    updatetime = datetime.datetime.now() + datetime.timedelta(seconds=self.interval)
                if s and self._filter(s):
                    logging.debug("Received string: %s" % s)
                    m = Measurement(s)
                    if self.status_attr:
                        logging.debug("Checking status_attr: %s" % self.status_attr)
                        stat = m.get_attr(self.status_attr)
                        if stat:
                            for key in self.stat_funcs:
                                if key == stat:
                                    logging.debug("Calling %s function" % key)
                                    self.stat_funcs[key](m)
                    self.get(m)
        self.disconnect()

def testSimpleZStacksMsgs(tdir, looper):
    names = ['Alpha', 'Beta']
    genKeys(tdir, names)
    names = ['Alpha', 'Beta']
    aseed = randomSeed()
    bseed = randomSeed()

    size = 100000
    msg = json.dumps({'random': randomSeed(size).decode()}).encode()

    def aHandler(m):
        str_m = "{}".format(m)
        print('{} printing... {}'.format(names[0], str_m[:100]))
        d, _ = m
        print('Message size is {}'.format(len(d['random'])))
        assert len(d['random']) == size

    def bHandler(m):
        print(beta.msgHandler)
        a = list(beta.peersWithoutRemotes)[0]
        try:
            beta.listener.send_multipart([a, msg],
                                         flags=zmq.NOBLOCK)
        except zmq.Again:
            return False
        str_m = "{}".format(m)
        print('{} printing... {}'.format(names[1], str_m[:100]))

    stackParams = {
        "name": names[0],
        "ha": genHa(),
        "auto": 2,
        "basedirpath": tdir
    }
    alpha = SimpleZStack(stackParams, aHandler, aseed, False)

    stackParams = {
        "name": names[1],
        "ha": genHa(),
        "auto": 2,
        "basedirpath": tdir
    }
    beta = SimpleZStack(stackParams, bHandler, bseed, True)

    amotor = SMotor(alpha)
    looper.add(amotor)

    bmotor = SMotor(beta)
    looper.add(bmotor)

    alpha.connect(name=beta.name, ha=beta.ha,
                  verKeyRaw=beta.verKeyRaw, publicKeyRaw=beta.publicKeyRaw)

    looper.runFor(0.25)
    alpha.send({'greetings': 'hi'}, beta.name)
    looper.runFor(1)