1  """ Defines a dummy socket implementing (part of) the zmq.Socket interface. """
   2  
   3  # Copyright (c) IPython Development Team.
   4  # Distributed under the terms of the Modified BSD License.
   5  
   6  from queue import Queue
   7  
   8  import zmq
   9  from traitlets import HasTraits, Instance, Int
  10  
  11  # -----------------------------------------------------------------------------
  12  # Dummy socket class
  13  # -----------------------------------------------------------------------------
  14  
  15  
  16  class DummySocket(HasTraits):
  17      """A dummy socket implementing (part of) the zmq.Socket interface."""
  18  
  19      queue = Instance(Queue, ())
  20      message_sent = Int(0)  # Should be an Event
  21      context = Instance(zmq.Context)
  22  
  23      def _context_default(self):
  24          return zmq.Context()
  25  
  26      # -------------------------------------------------------------------------
  27      # Socket interface
  28      # -------------------------------------------------------------------------
  29  
  30      def recv_multipart(self, flags=0, copy=True, track=False):
  31          """Recv a multipart message."""
  32          return self.queue.get_nowait()
  33  
  34      def send_multipart(self, msg_parts, flags=0, copy=True, track=False):
  35          """Send a multipart message."""
  36          msg_parts = list(map(zmq.Message, msg_parts))
  37          self.queue.put_nowait(msg_parts)
  38          self.message_sent += 1
  39  
  40      def flush(self, timeout=1.0):
  41          """no-op to comply with stream API"""
  42