Attachment 592165 Details for Bug 832512 – reproducer script

reproducer script

test.py (text/x-python), 4.36 KB, created by Leonid Zhaldybin on 2012-06-15 15:28:19 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Leonid Zhaldybin

Created: 2012-06-15 15:28:19 UTC

Size: 4.36 KB

patch

obsolete

>#!/usr/bin/env python
>#
># Licensed to the Apache Software Foundation (ASF) under one
># or more contributor license agreements.  See the NOTICE file
># distributed with this work for additional information
># regarding copyright ownership.  The ASF licenses this file
># to you under the Apache License, Version 2.0 (the
># "License"); you may not use this file except in compliance
># with the License.  You may obtain a copy of the License at
>#
>#   http://www.apache.org/licenses/LICENSE-2.0
>#
># Unless required by applicable law or agreed to in writing,
># software distributed under the License is distributed on an
># "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
># KIND, either express or implied.  See the License for the
># specific language governing permissions and limitations
># under the License.
>#
>
># Usage: mysub.py [host|localhost] [port|5672] [accept_mode|0]
>
>import qpid
>import sys
>import os
>from qpid.util import connect
>from qpid.connection import Connection
>from qpid.datatypes import Message, RangedSet, uuid4
>from qpid.queue import Empty
>from time import time as gettime
>
>class Subscriber:
>  def __init__(self, host="localhost", port=5672, mode=0):
>    self.host = host
>    self.port = port
>    self.mode = mode
>    self.socket = connect(host, port)
>    self.connection = Connection (sock=self.socket)
>    self.connection.start()
>    self.session = self.connection.session(str(uuid4()))
>    self.session_to = 60
>    self.name = ""
>
>  def conf_queues(self):
>    # declare queues on the server
>    self.name = "queue-" + self.session.name
>
>    self.session.queue_declare(queue=self.name, exclusive=True, auto_delete=True)
>    self.session.exchange_bind(exchange="amq.topic", queue=self.name, binding_key="mytopic")
>
>    # Bind each queue to the control queue so we know when to stop
>    self.session.exchange_bind(exchange="amq.topic", queue=self.name, binding_key="control")
>
>  def dump_queue(self, queue):
>    self.session.auto_sync = False
>    content = "" # Content of the last message read
>    final = "That's all, folks!" # In a message body, signals the last message
>    message = 0
>    start_time = gettime()
>
>    while content != final:
>      try:
>        message = queue.get(timeout=self.session_to)
>        content = message.body
>        if self.mode == 0:
>          # send message accept only in explicit mode
>          self.session.message_accept(RangedSet(message.id));
>        print content
>      except Empty:
>        print "No more messages!"
>        end_time = gettime()
>        print "Duration: %i seconds." % (end_time - start_time)
>        return
>
>  def subscribe_queue(self, local_queue_name):
>
>    print "Subscribing local queue '" + local_queue_name + "' to " + \
>          self.name + "'"
>
>    queue = self.session.incoming(local_queue_name)
>
>    if self.mode == 0:
>      self.session.message_subscribe(queue=self.name, \
>                              destination=local_queue_name);
>    else:
>      self.session.message_subscribe(queue=self.name, \
>                                destination=local_queue_name, \
>                                accept_mode=self.session.accept_mode.none, \
>                                acquire_mode=self.session.acquire_mode.pre_acquired);
>    queue.start()
>
>    return queue
>
>  def cleanup(self):
>    self.session.close(timeout=self.session_to)
>
>class Publisher:
>  def __init__(self, host="localhost", port=5672):
>    self.host = host
>    self.port = port
>    self.socket = connect(host, port)
>    self.connection = Connection (sock=self.socket)
>    self.connection.start()
>    self.session = self.connection.session(str(uuid4()))
>    self.session_to = 60
>
>  def send_msg(self, routing_key, msg_num=10000):
>    self.session.auto_sync = False
>    props = self.session.delivery_properties(routing_key=routing_key)
>    for i in range(msg_num):
>      self.session.message_transfer(destination="amq.topic", \
>        message=Message(props,routing_key + " " + str(i)))
>
>
>  def cleanup(self):
>    self.session.close(timeout=self.session_to)
>
>
>subs = Subscriber()
>subs.conf_queues()
>
>publ = Publisher()
>publ.send_msg("mytopic", msg_num=50000)
>
># Subscribe local queues to server queues
>local_queue = subs.subscribe_queue("local_queue")
>
># Call dump_queue to print messages from each queue
>print "Messages:"
>subs.dump_queue(local_queue)
>
>#----- Cleanup ------------------------------------------------
>
># Clean up before exiting so there are no open threads.
>subs.cleanup()
>publ.cleanup()
>
># eof

Actions: View

Attachments on bug 832512: 592165