haigha_perf.py
def runBlockingSocketAltPubConsumeTest(implClassName,
                                       numMessages,
                                       messageSize,
                                       useConsumerAcks,
                                       deliveryConfirmation):
  """Alternates publishing/consuming the given number of messages of the
  given size one message at a time via default exchange
  """
  g_log.info(
    "runBlockingSocketAltPubConsumeTest: impl=%s; numMessages=%d; "
    "messageSize=%s; useConsumerAcks=%s, deliveryConfirmation=%s", implClassName,
    numMessages, messageSize, useConsumerAcks, deliveryConfirmation)

  implClass = getattr(socket_transport, implClassName)
  assert implClass is socket_transport.SocketTransport, implClass


  payload = "a" * messageSize

  class State(object):
    closing = False
    publishAckMessageId = None
    channelClosed = False
    connectionClosed = False
    connection = None
    incomingMsgs = collections.deque()


  def onConnectionClosed():
    State.connectionClosed = True
    g_log.info("%s: connection closed; close_info=%s", implClassName,
               State.connection.close_info if State.connection else None)
    assert State.closing, "unexpected onnection-close"


  conn = RabbitConnection(
    transport="socket",
    sock_opts={(socket.IPPROTO_TCP, socket.TCP_NODELAY) : 1},
    close_cb=onConnectionClosed,
    **getConnectionParameters())
  g_log.info("%s: opened connection", implClassName)


  def onChannelClosed(ch):
    State.channelClosed = True
    g_log.info("%s: channel closed; close_info=%s",
               implClassName, ch.close_info)
    assert State.closing, "unexpected channel-close"


  channel = conn.channel()
  channel.add_close_listener(onChannelClosed)
  g_log.info("%s: opened channel", implClassName)

  if deliveryConfirmation:
    channel.confirm.select()

    def onAck(mid):
      State.publishAckMessageId = mid

    def onNack(mid):
      msg = "Got Nack from broker: %r" % (mid,)
      g_log.error(msg)
      raise RuntimeError(msg)

    channel.basic.set_ack_listener( onAck )
    channel.basic.set_nack_listener( onNack )

    g_log.info("%s: enabled message delivery confirmation", implClassName)


  # Create transient queue
  qname = channel.queue.declare(passive=False, durable=False, exclusive=False,
                                auto_delete=True, nowait=False)[0]
  g_log.info("%s: Created queue=%s", implClassName, qname)


  def publish():
    assert State.publishAckMessageId is None
    message = Message(payload)
    msgId = channel.basic.publish(message, exchange="",
                                  routing_key=qname,
                                  immediate=False, mandatory=False)
    if deliveryConfirmation:
      while State.publishAckMessageId is None:
        conn.read_frames()
      else:
        assert State.publishAckMessageId == msgId, (
          State.publishAckMessageId, msgId)
        State.publishAckMessageId = None

    assert State.publishAckMessageId is None

    return msgId


  # Create consumer

  def onIncomingMessage(msg):
    State.incomingMsgs.append(msg)

  channel.basic.consume(qname, consumer=onIncomingMessage,
                        no_ack=not useConsumerAcks, nowait=False)
  g_log.info("%s: created consumer", implClassName)

  # Publish/consume
  for i in xrange(numMessages):
    assert not State.incomingMsgs, State.incomingMsgs

    msgId = publish()

    # Wait for incoming
    while not State.incomingMsgs:
      conn.read_frames()

    assert len(State.incomingMsgs) == 1, State.incomingMsgs

    msg = State.incomingMsgs.pop()
    assert len(msg.body) == len(payload)

    if useConsumerAcks:
      # print >> sys.stderr, "ZZZ delivery_info:", msg.delivery_info
      channel.basic.ack(msg.delivery_info["delivery_tag"])

  else:
    g_log.info("Published %d messages of size=%d via=%s",
               i+1, messageSize, implClass)


  State.closing = True

  g_log.info("%s: closing channel", implClassName)
  channel.close()
  while not State.channelClosed:
    conn.read_frames()

  g_log.info("%s: closing connection", implClassName)
  conn.close()
  while not State.connectionClosed:
    conn.read_frames()

  assert State.publishAckMessageId is None

  assert not State.incomingMsgs

  g_log.info("%s: DONE", implClassName)
Alternates publishing/consuming the given number of messages of the
  given size one message at a time via default exchange
  
Alternates publishing/consuming the given number of messages of the
given size one message at a time via default exchange
Alternates publishing / consuming the given number of messages of the given size one message at a time via default exchange
def runBlockingSocketAltPubConsumeTest(implClassName,
                                       numMessages,
                                       messageSize,
                                       useConsumerAcks,
                                       deliveryConfirmation):
  g_log.info(
    "runBlockingSocketAltPubConsumeTest: impl=%s; numMessages=%d; "
    "messageSize=%s; useConsumerAcks=%s, deliveryConfirmation=%s", implClassName,
    numMessages, messageSize, useConsumerAcks, deliveryConfirmation)
  implClass = getattr(socket_transport, implClassName)
  assert implClass is socket_transport.SocketTransport, implClass
  payload = "a" * messageSize
  class State(object):
    closing = False
    publishAckMessageId = None
    channelClosed = False
    connectionClosed = False
    connection = None
    incomingMsgs = collections.deque()
  def onConnectionClosed():
    State.connectionClosed = True
    g_log.info("%s: connection closed; close_info=%s", implClassName,
               State.connection.close_info if State.connection else None)
    assert State.closing, "unexpected onnection-close"
  conn = RabbitConnection(
    transport="socket",
    sock_opts={(socket.IPPROTO_TCP, socket.TCP_NODELAY) : 1},
    close_cb=onConnectionClosed,
    **getConnectionParameters())
  g_log.info("%s: opened connection", implClassName)
  def onChannelClosed(ch):
    State.channelClosed = True
    g_log.info("%s: channel closed; close_info=%s",
               implClassName, ch.close_info)
    assert State.closing, "unexpected channel-close"
  channel = conn.channel()
  channel.add_close_listener(onChannelClosed)
  g_log.info("%s: opened channel", implClassName)
  if deliveryConfirmation:
    channel.confirm.select()
    def onAck(mid):
      State.publishAckMessageId = mid
    def onNack(mid):
      msg = "Got Nack from broker: %r" % (mid,)
      g_log.error(msg)
      raise RuntimeError(msg)
    channel.basic.set_ack_listener( onAck )
    channel.basic.set_nack_listener( onNack )
    g_log.info("%s: enabled message delivery confirmation", implClassName)
  qname = channel.queue.declare(passive=False, durable=False, exclusive=False,
                                auto_delete=True, nowait=False)[0]
  g_log.info("%s: Created queue=%s", implClassName, qname)
  def publish():
    assert State.publishAckMessageId is None
    message = Message(payload)
    msgId = channel.basic.publish(message, exchange="",
                                  routing_key=qname,
                                  immediate=False, mandatory=False)
    if deliveryConfirmation:
      while State.publishAckMessageId is None:
        conn.read_frames()
      else:
        assert State.publishAckMessageId == msgId, (
          State.publishAckMessageId, msgId)
        State.publishAckMessageId = None
    assert State.publishAckMessageId is None
    return msgId
  def onIncomingMessage(msg):
    State.incomingMsgs.append(msg)
  channel.basic.consume(qname, consumer=onIncomingMessage,
                        no_ack=not useConsumerAcks, nowait=False)
  g_log.info("%s: created consumer", implClassName)
  for i in xrange(numMessages):
    assert not State.incomingMsgs, State.incomingMsgs
    msgId = publish()
    while not State.incomingMsgs:
      conn.read_frames()
    assert len(State.incomingMsgs) == 1, State.incomingMsgs
    msg = State.incomingMsgs.pop()
    assert len(msg.body) == len(payload)
    if useConsumerAcks:
      channel.basic.ack(msg.delivery_info["delivery_tag"])
  else:
    g_log.info("Published %d messages of size=%d via=%s",
               i+1, messageSize, implClass)
  State.closing = True
  g_log.info("%s: closing channel", implClassName)
  channel.close()
  while not State.channelClosed:
    conn.read_frames()
  g_log.info("%s: closing connection", implClassName)
  conn.close()
  while not State.connectionClosed:
    conn.read_frames()
  assert State.publishAckMessageId is None
  assert not State.incomingMsgs
  g_log.info("%s: DONE", implClassName)
def runBlockingSocketAltPubConsumeTest ( implClassName , numMessages , messageSize , useConsumerAcks , deliveryConfirmation ) : g_log . info ( "runBlockingSocketAltPubConsumeTest: impl=%s; numMessages=%d; " "messageSize=%s; useConsumerAcks=%s, deliveryConfirmation=%s" , implClassName , numMessages , messageSize , useConsumerAcks , deliveryConfirmation ) implClass = getattr ( socket_transport , implClassName ) assert implClass is socket_transport . SocketTransport , implClass payload = "a" * messageSize class State ( object ) : closing = False publishAckMessageId = None channelClosed = False connectionClosed = False connection = None incomingMsgs = collections . deque ( ) def onConnectionClosed ( ) : State . connectionClosed = True g_log . info ( "%s: connection closed; close_info=%s" , implClassName , State . connection . close_info if State . connection else None ) assert State . closing , "unexpected onnection-close" conn = RabbitConnection ( transport = "socket" , sock_opts = { ( socket . IPPROTO_TCP , socket . TCP_NODELAY ) : 1 } , close_cb = onConnectionClosed , ** getConnectionParameters ( ) ) g_log . info ( "%s: opened connection" , implClassName ) def onChannelClosed ( ch ) : State . channelClosed = True g_log . info ( "%s: channel closed; close_info=%s" , implClassName , ch . close_info ) assert State . closing , "unexpected channel-close" channel = conn . channel ( ) channel . add_close_listener ( onChannelClosed ) g_log . info ( "%s: opened channel" , implClassName ) if deliveryConfirmation : channel . confirm . select ( ) def onAck ( mid ) : State . publishAckMessageId = mid def onNack ( mid ) : msg = "Got Nack from broker: %r" % ( mid , ) g_log . error ( msg ) raise RuntimeError ( msg ) channel . basic . set_ack_listener ( onAck ) channel . basic . set_nack_listener ( onNack ) g_log . info ( "%s: enabled message delivery confirmation" , implClassName ) qname = channel . queue . declare ( passive = False , durable = False , exclusive = False , auto_delete = True , nowait = False ) [ 0 ] g_log . info ( "%s: Created queue=%s" , implClassName , qname ) def publish ( ) : assert State . publishAckMessageId is None message = Message ( payload ) msgId = channel . basic . publish ( message , exchange = "" , routing_key = qname , immediate = False , mandatory = False ) if deliveryConfirmation : while State . publishAckMessageId is None : conn . read_frames ( ) else : assert State . publishAckMessageId == msgId , ( State . publishAckMessageId , msgId ) State . publishAckMessageId = None assert State . publishAckMessageId is None return msgId def onIncomingMessage ( msg ) : State . incomingMsgs . append ( msg ) channel . basic . consume ( qname , consumer = onIncomingMessage , no_ack = not useConsumerAcks , nowait = False ) g_log . info ( "%s: created consumer" , implClassName ) for i in xrange ( numMessages ) : assert not State . incomingMsgs , State . incomingMsgs msgId = publish ( ) while not State . incomingMsgs : conn . read_frames ( ) assert len ( State . incomingMsgs ) == 1 , State . incomingMsgs msg = State . incomingMsgs . pop ( ) assert len ( msg . body ) == len ( payload ) if useConsumerAcks : channel . basic . ack ( msg . delivery_info [ "delivery_tag" ] ) else : g_log . info ( "Published %d messages of size=%d via=%s" , i + 1 , messageSize , implClass ) State . closing = True g_log . info ( "%s: closing channel" , implClassName ) channel . close ( ) while not State . channelClosed : conn . read_frames ( ) g_log . info ( "%s: closing connection" , implClassName ) conn . close ( ) while not State . connectionClosed : conn . read_frames ( ) assert State . publishAckMessageId is None assert not State . incomingMsgs g_log . info ( "%s: DONE" , implClassName )
Alternates publishing/consuming the given number of messages of the
given size one message at a time via default exchange
Alternates publishing / consuming the given number of messages of the given size one message at a time via default exchange
"""Alternates publishing/consuming the given number of messages of the
  given size one message at a time via default exchange
  """ # Create transient queue # Create consumer # Publish/consume # Wait for incoming # print >> sys.stderr, "ZZZ delivery_info:", msg.delivery_info
implClassName
numMessages
messageSize
useConsumerAcks
deliveryConfirmation