def _resize_test(self, queue_name, num_msgs, msg_size, resize_num_files, resize_file_size, init_num_files = 8,
                 init_file_size = 24, exp_fail = False, wait_time = None):
     # Using a sender will force the creation of an empty persistent queue which is needed for some tests
     broker = self.broker(store_args(), name="broker", expect=EXPECT_EXIT_OK, wait=wait_time)
     ssn = broker.connect().session()
     snd = ssn.sender("%s; {create:always, node:{durable:True}}" % queue_name)
     
     msgs = []
     for index in range(0, num_msgs):
         msg = Message(self.make_message(index, msg_size), durable=True, id=uuid4(), correlation_id="msg-%04d"%index)
         msgs.append(msg)
         snd.send(msg)
     broker.terminate()
     
     res = self._resize_store(os.path.join(self.dir, "broker", "rhm", "jrnl"), queue_name, resize_num_files,
                          resize_file_size, exp_fail)
     if res != 0:
         if exp_fail:
             return
         self.fail("ERROR: Resize operation failed with return code %d" % res)
     elif exp_fail:
         self.fail("ERROR: Resize operation succeeded, but a failure was expected")
     
     broker = self.broker(store_args(), name="broker")
     self.check_messages(broker, queue_name, msgs, True)

def send_message(connection, notif):
    session = connection.session(str(uuid4()))
    
    props = session.delivery_properties(routing_key=TOPIC_NAME)
    head = session.message_properties(application_headers={'sender':notif.sender,
                                                                'response':notif.response})
    session.message_transfer(destination=DESTINATION, message=Message(props, head, notif.messageId))
    session.close(timeout=10)
    connection.close()

    def run(self):
        # Create connection and session
        socket = connect( self.host, self.port)
        
        connection = Connection(sock=socket, username=self.username, password=self.password)
        
        print("consumer "+self.queueName+": starting connection...")
        connection.start()
        print("consumer "+self.queueName+": ...connection started")

        print("consumer "+self.queueName+": getting session...")
        session = connection.session(str(uuid4()))
        print("consumer "+self.queueName+": ...session got")

        # Define local queue
        local_queue_name = 'my_local_queue_' +self.queueName

        # Create local queue
        print("consumer "+self.queueName+": getting queue...")
        queue = session.incoming(local_queue_name)
        print("consumer "+self.queueName+": ...queue got")

        # Route messages from message_queue to my_local_queue
        print("consumer "+self.queueName+": subscribing...")
        session.message_subscribe(queue = self.queueName, destination=local_queue_name)
        print("consumer "+self.queueName+": ...subscribed")

        print("consumer "+self.queueName+": starting queue...")
        queue.start()
        print("consumer "+self.queueName+": ...queue started")

        content = ''
        index = 0
        
        while (self.running):
            try:
                # Get message from the local queue, timeout 5 seconds
                message = queue.get(timeout=5)
            except:
                break # exit this thread, consumer

            # Get body of the message
            content = message.body

            #message_properties = message.get("message_properties")

            # Accept message (removes it from the queue)
            session.message_accept(RangedSet(message.id))

            if (content != ""):
                try:
                    self.readGPB(content)

                except Exception, e: 
                    print( "Unexpected error: %s\n" % str(e) )

def set_application_headers(message_properties):

  message_properties.application_headers = {}
  message_properties.application_headers["void"] = None
  message_properties.application_headers["boolean_true"] =  boolean_true
  message_properties.application_headers["boolean_false"] = boolean_false
  message_properties.application_headers["Uint8_0"] = Uint8_0
  message_properties.application_headers["Uint8_max"] = Uint8_max
  message_properties.application_headers["Uint16_0"] = Uint16_0
  message_properties.application_headers["Uint16_max"] = Uint16_max
  message_properties.application_headers["Uint32_0"] = Uint32_0
  message_properties.application_headers["Uint32_max"] = Uint32_max
  message_properties.application_headers["Uint64_0"] = Uint64_0
#  message_properties.application_headers["Uint64_max"] = Uint64_max
  message_properties.application_headers["Int8_min"] = Int8_min
  message_properties.application_headers["Int8_0"] = Int8_0
  message_properties.application_headers["Int8_max"] = Int8_max
  message_properties.application_headers["Int16_min"] = Int16_min
  message_properties.application_headers["Int16_0"] = Int16_0
  message_properties.application_headers["Int16_max"] = Int16_max
  message_properties.application_headers["Int32_min"] = Int32_min
  message_properties.application_headers["Int32_0"] = Int32_0
  message_properties.application_headers["Int32_max"] = Int32_max
  message_properties.application_headers["Int64_min"] = Int64_min
  message_properties.application_headers["Int64_0"] = Int64_0
  message_properties.application_headers["Int64_max"] = Int64_max
 
  message_properties.application_headers["Float_pi"] = Float_pi
  message_properties.application_headers["Float_neg"] = Float_neg
  message_properties.application_headers["Float_big"] = Float_big
  message_properties.application_headers["Float_small"] = Float_small
  message_properties.application_headers["Float_neg0"] = Float_neg0
  message_properties.application_headers["Float_pos0"] = Float_pos0
  message_properties.application_headers["Float_INF"] = Float_INF
  message_properties.application_headers["Float_Negative_INF"] = Float_Negative_INF

  message_properties.application_headers["Double_pi"] = Double_pi
  message_properties.application_headers["Double_neg"] = Double_neg
  message_properties.application_headers["Double_big"] = Double_big
  message_properties.application_headers["Double_small"] = Double_small
  message_properties.application_headers["Double_neg0"] = Double_neg0
  message_properties.application_headers["Double_pos0"] = Double_pos0
  message_properties.application_headers["Double_INF"] = Double_INF
  message_properties.application_headers["Double_Negative_INF"] = Double_Negative_INF

  message_properties.application_headers["char_1byte"] = char_1byte
  message_properties.application_headers["char_2byte"] = char_2byte
  message_properties.application_headers["char_3byte"] = char_3byte
  message_properties.application_headers["char_4byte"] = char_4byte

  message_properties.application_headers["timestamp"] = timestamp
  message_properties.application_headers["UUID"] = uuid4() 
  message_properties.application_headers["String_Greek"] = String_Greek 
  message_properties.application_headers["String_Empty"] = String_Empty

 def invalid_policy_args(self, args, name="test-queue"):
     # go through invalid declare attempts twice to make sure that
     # the queue doesn't actually get created first time around
     # even if exception is thrown
     for i in range(1, 3):
         try:
             self.session.queue_declare(queue=name, arguments=args)
             self.session.queue_delete(queue=name) # cleanup
             self.fail("declare with invalid policy args suceeded: %s (iteration %d)" % (args, i))
         except SessionException, e:
             self.session = self.conn.session(str(uuid4()))

 def test_timed_autodelete(self):
     session = self.session
     session2 = self.conn.session("another-session")
     name=str(uuid4())
     session2.queue_declare(queue=name, exclusive=True, auto_delete=True, arguments={"qpid.auto_delete_timeout":3})
     session2.close()
     result = session.queue_query(queue=name)
     self.assertEqual(name, result.queue)
     sleep(5)
     result = session.queue_query(queue=name)
     self.assert_(not result.queue)

 def __init__(self, broker, **kw):
     self.set_broker(broker)
     self.socket = connect(self.host, self.port)
     if self.url.scheme == URL.AMQPS:
         self.socket = ssl(self.socket)
     self.connection = Connection(sock=self.socket,
                                  username=self.user,
                                  password=self.password)
     self.connection.start()
     log.info("Connected to AMQP Broker %s" % self.host)
     self.session = self.connection.session(str(uuid4()))

 def testMapAll(self):
   decoded = self.check("map", {"string": "this is a test",
                                "unicode": u"this is a unicode test",
                                "binary": "\x7f\xb4R^\xe5\xf0:\x89\x96E1\xf6\xfe\xb9\x1b\xf5",
                                "int": 3,
                                "long": 2**32,
                                "timestamp": timestamp(0),
                                "none": None,
                                "map": {"string": "nested map"},
                                "list": [1, "two", 3.0, -4],
                                "uuid": uuid4()})
   assert isinstance(decoded["timestamp"], timestamp)

 def __init__(self, hub, config):
     self.config = config
     self.set_broker(self.config.get('amqp_broker'))
     self.socket = connect(self.host, self.port)
     if self.url.scheme == URL.AMQPS:
         self.socket = ssl(self.socket)
     self.connection = Connection(sock=self.socket,
                                  username=self.user,
                                  password=self.password)
     self.connection.start()
     log.info("Connected to AMQP Broker %s" % self.host)
     self.session = self.connection.session(str(uuid4()))
     self.local_queues = []
     super(QpidAMQPHubExtension, self).__init__()

def createSchedule(title, uuidStart, uuidEnd, dateStart, dateEnd, color, repeat):
    """create schedule structure
       @see http://arshaw.com/fullcalendar/docs2/event_data/Event_Object/"""
    return {
        'id': str(uuid4()),
        'title': title,
        'start': dateStart,
        'end': dateEnd,
        'color': color,
        'uuidStart': uuidStart,
        'uuidEnd': uuidEnd,
        'repeat': repeat,
        'allDay': 0
    }

    def __init__(self, host='localhost', port=5672):
        '''
        Connect to QPID and make bindings to route message to external.dropbox queue
        @param host: string hostname of computer running EDEX and QPID (default localhost)
        @param port: integer port used to connect to QPID (default 5672)
        '''

        try:
            #
            self.socket = connect(host, port)
            self.connection = Connection (sock=self.socket, username=QPID_USERNAME, password=QPID_PASSWORD)
            self.connection.start()
            self.session = self.connection.session(str(uuid4()))
            self.session.exchange_bind(exchange='amq.direct', queue='external.dropbox', binding_key='external.dropbox')
            print('Connected to Qpid')
        except:
            print('Unable to connect to Qpid')

 def test_priority_fairshare(self):
     """Verify priority queues replicate correctly"""
     primary  = HaBroker(self, name="primary")
     primary.promote()
     backup = HaBroker(self, name="backup", brokers_url=primary.host_port())
     session = primary.connect().session()
     levels = 8
     priorities = [4,5,3,7,8,8,2,8,2,8,8,16,6,6,6,6,6,6,8,3,5,8,3,5,5,3,3,8,8,3,7,3,7,7,7,8,8,8,2,3]
     limits={7:0,6:4,5:3,4:2,3:2,2:2,1:2}
     limit_policy = ",".join(["'qpid.fairshare':5"] + ["'qpid.fairshare-%s':%s"%(i[0],i[1]) for i in limits.iteritems()])
     s = session.sender("priority-queue; {create:always, node:{x-declare:{arguments:{'qpid.priorities':%s, %s}}}}"%(levels,limit_policy))
     messages = [Message(content=str(uuid4()), priority = p) for p in priorities]
     for m in messages: s.send(m)
     backup.wait_backup(s.target)
     r = backup.connect_admin().session().receiver("priority-queue")
     received = [r.fetch().content for i in priorities]
     sort = sorted(messages, key=lambda m: priority_level(m.priority, levels), reverse=True)
     fair = [m.content for m in fairshare(sort, lambda l: limits.get(l,0), levels)]
     self.assertEqual(received, fair)

    def subscribe(self, topic, callback):
        queue_name = '_'.join([
            "moksha_consumer", self.session.name, str(uuid4()),
        ])
        server_queue_name = local_queue_name = queue_name

        self.queue_declare(queue=server_queue_name, exclusive=True,
                           auto_delete=True)
        self.exchange_bind(server_queue_name, binding_key=topic)

        self.local_queues.append(self.session.incoming(local_queue_name))

        self.message_subscribe(queue=server_queue_name,
                               destination=local_queue_name)

        self.local_queues[-1].start()
        self.local_queues[-1].listen(callback)

        super(QpidAMQPHubExtension, self).subscribe(topic, callback)

    def test_catchup_store(self):
        """Verify that a backup erases queue data from store recovery before
        doing catch-up from the primary."""
        cluster = HaCluster(self, 2)
        sn = cluster[0].connect().session()
        s1 = sn.sender("q1;{create:always,node:{durable:true}}")
        for m in ["foo","bar"]: s1.send(Message(m, durable=True))
        s2 = sn.sender("q2;{create:always,node:{durable:true}}")
        sk2 = sn.sender("ex/k2;{create:always,node:{type:topic, durable:true, x-declare:{type:'direct'}, x-bindings:[{exchange:ex,key:k2,queue:q2}]}}")
        sk2.send(Message("hello", durable=True))
        # Wait for backup to catch up.
        cluster[1].assert_browse_backup("q1", ["foo","bar"]) 
        cluster[1].assert_browse_backup("q2", ["hello"])

        # Make changes that the backup doesn't see
        cluster.kill(1, promote_next=False)
        r1 = cluster[0].connect().session().receiver("q1")
        for m in ["foo", "bar"]: self.assertEqual(r1.fetch().content, m)
        r1.session.acknowledge()
        for m in ["x","y","z"]: s1.send(Message(m, durable=True))
        # Use old connection to unbind
        us = cluster[0].connect_old().session(str(uuid4()))
        us.exchange_unbind(exchange="ex", binding_key="k2", queue="q2")
        us.exchange_bind(exchange="ex", binding_key="k1", queue="q1")
        # Restart both brokers from store to get inconsistent sequence numbering.
        cluster.bounce(0, promote_next=False)
        cluster[0].promote()
        cluster[0].wait_status("active")
        cluster.restart(1)
        cluster[1].wait_status("ready")

        # Verify state
        cluster[0].assert_browse("q1",  ["x","y","z"])
        cluster[1].assert_browse_backup("q1",  ["x","y","z"])
        sn = cluster[0].connect().session() # FIXME aconway 2012-09-25: should fail over!
        sn.sender("ex/k1").send("boo")
        cluster[0].assert_browse_backup("q1", ["x","y","z", "boo"])
        cluster[1].assert_browse_backup("q1", ["x","y","z", "boo"])
        sn.sender("ex/k2").send("hoo") # q2 was unbound so this should be dropped.
        sn.sender("q2").send("end")    # mark the end of the queue for assert_browse
        cluster[0].assert_browse("q2", ["hello", "end"])
        cluster[1].assert_browse_backup("q2", ["hello", "end"])

 def setup(p, prefix, primary):
     """Create config, send messages on the primary p"""
     s = p.sender(queue(prefix+"q1", "all"))
     for m in ["a", "b", "1"]: s.send(Message(m))
     # Test replication of dequeue
     self.assertEqual(p.receiver(prefix+"q1").fetch(timeout=0).content, "a")
     p.acknowledge()
     p.sender(queue(prefix+"q2", "configuration")).send(Message("2"))
     p.sender(queue(prefix+"q3", "none")).send(Message("3"))
     p.sender(exchange(prefix+"e1", "all", prefix+"q1")).send(Message("4"))
     p.sender(exchange(prefix+"e2", "all", prefix+"q2")).send(Message("5"))
     # Test  unbind
     p.sender(queue(prefix+"q4", "all")).send(Message("6"))
     s3 = p.sender(exchange(prefix+"e4", "all", prefix+"q4"))
     s3.send(Message("7"))
     # Use old connection to unbind
     us = primary.connect_old().session(str(uuid4()))
     us.exchange_unbind(exchange=prefix+"e4", binding_key="", queue=prefix+"q4")
     p.sender(prefix+"e4").send(Message("drop1")) # Should be dropped
     # Need a marker so we can wait till sync is done.
     p.sender(queue(prefix+"x", "configuration"))

 def get_session(self, user, passwd):
     socket = connect(self.broker.host, self.broker.port)
     connection = Connection (sock=socket, username=user, password=passwd,
                              mechanism="PLAIN")
     connection.start()
     return connection.session(str(uuid4()))

def dump_queue(binfo, queue_name, to):
   # Create a client and log in to it.
   child_connection = Connection(sock=connect(str(binfo['ip']), int(binfo['port'])))
   child_connection.start()
   child_session = child_connection.session(str(uuid4()))
   child_session.queue_declare(queue=queue_name, exclusive=True)
   child_session.exchange_bind(exchange='amq.direct', queue=queue_name, binding_key=queue_name)

   print 'Messages queue: ' + queue_name 

   # Create the local queue. Use the queue name as destination name
   dest = queue_name 
   queue = child_session.incoming(dest)

   # Subscribe the local queue to the queue on the server
   child_session.message_subscribe(queue=queue_name, destination=dest, accept_mode=child_session.accept_mode.explicit)
   child_session.message_flow(dest, child_session.credit_unit.message, 0xFFFFFFFFL)
   child_session.message_flow(dest, child_session.credit_unit.byte, 0xFFFFFFFFL)

   # Read responses as they come in and print to the screen.
   message = 0
   count = 0

   while True:
      try:
         message = queue.get(timeout=to)
         count = count + 1
         if count == 1:
            print 'Received first reponse: %s ' % str(time.time())
      except Empty:
         print 'Received %s messages: %s' % (str(count), str(time.time() - to))
         break
#      except qpid.session.Closed:
#         print "Re-establishing"
#         try:
#            child_connection.close()
#         except:
#            pass
#
#         # Give broker time to stablize and accept connections
#         time.sleep(2)
#         child_connection = Connection(sock=connect(str(binfo['ip']), int(binfo['port'])))
#         child_connection.start()
#         child_session = child_connection.session(str(uuid4()))
#         child_session.queue_declare(queue=queue_name, exclusive=True)
#         child_session.exchange_bind(exchange='amq.direct', queue=queue_name, binding_key=queue_name)
#
#         # Create the local queue. Use the queue name as destination name
#         queue = child_session.incoming(dest)
#
#         # Subscribe the local queue to the queue on the server
#         child_session.message_subscribe(queue=queue_name, destination=dest, accept_mode=child_session.accept_mode.explicit)
#         child_session.message_flow(dest, child_session.credit_unit.message, 0xFFFFFFFFL)
#         child_session.message_flow(dest, child_session.credit_unit.byte, 0xFFFFFFFFL)
      except:
         print 'Unexpected exception!'
         break

      if message != 0:
        child_session.message_accept(RangedSet(message.id))

   child_session.close(timeout=10)
   child_connection.close()
   return (0)

def _mangle(addr):
    if addr and addr.startswith("#"):
        return str(uuid4()) + addr
    else:
        return addr

#.........这里部分代码省略.........
    @type sasl_service: str
    @param sasl_service: the service name if needed by the SASL mechanism in use
    @type sasl_min_ssf: int
    @param sasl_min_ssf: the minimum acceptable security strength factor
    @type sasl_max_ssf: int
    @param sasl_max_ssf: the maximum acceptable security strength factor

    @type reconnect: bool
    @param reconnect: enable/disable automatic reconnect
    @type reconnect_timeout: float
    @param reconnect_timeout: total time to attempt reconnect
    @type reconnect_interval_min: float
    @param reconnect_interval_min: minimum interval between reconnect attempts
    @type reconnect_interval_max: float
    @param reconnect_interval_max: maximum interval between reconnect attempts
    @type reconnect_interval: float
    @param reconnect_interval: set both min and max reconnect intervals
    @type reconnect_limit: int
    @param reconnect_limit: limit the total number of reconnect attempts
    @type reconnect_urls: list[str]
    @param reconnect_urls: list of backup hosts specified as urls

    @type address_ttl: float
    @param address_ttl: time until cached address resolution expires

    @type ssl_keyfile: str
    @param ssl_keyfile: file with client's private key (PEM format)
    @type ssl_certfile: str
    @param ssl_certfile: file with client's public (eventually priv+pub) key (PEM format)
    @type ssl_trustfile: str
    @param ssl_trustfile: file trusted certificates to validate the server
    @type ssl_skip_hostname_check: bool
    @param ssl_skip_hostname_check: disable verification of hostname in
    certificate. Use with caution - disabling hostname checking leaves you
    vulnerable to Man-in-the-Middle attacks.

    @rtype: Connection
    @return: a disconnected Connection
    """
        if url is None:
            url = options.get("host")
        if isinstance(url, basestring):
            url = URL(url)
        self.host = url.host
        if options.has_key("transport"):
            self.transport = options.get("transport")
        elif url.scheme == url.AMQP:
            self.transport = "tcp"
        elif url.scheme == url.AMQPS:
            self.transport = "ssl"
        else:
            self.transport = "tcp"
        if self.transport in ("ssl", "tcp+tls"):
            self.port = default(url.port, options.get("port", AMQPS_PORT))
        else:
            self.port = default(url.port, options.get("port", AMQP_PORT))
        self.heartbeat = options.get("heartbeat")
        self.username = default(url.user, options.get("username", None))
        self.password = default(url.password, options.get("password", None))
        self.auth_username = None

        self.sasl_mechanisms = options.get("sasl_mechanisms")
        self.sasl_service = options.get("sasl_service", "qpidd")
        self.sasl_min_ssf = options.get("sasl_min_ssf")
        self.sasl_max_ssf = options.get("sasl_max_ssf")

        self.reconnect = options.get("reconnect", False)
        self.reconnect_timeout = options.get("reconnect_timeout")
        reconnect_interval = options.get("reconnect_interval")
        self.reconnect_interval_min = options.get("reconnect_interval_min", default(reconnect_interval, 1))
        self.reconnect_interval_max = options.get("reconnect_interval_max", default(reconnect_interval, 2 * 60))
        self.reconnect_limit = options.get("reconnect_limit")
        self.reconnect_urls = options.get("reconnect_urls", [])
        self.reconnect_log = options.get("reconnect_log", True)

        self.address_ttl = options.get("address_ttl", 60)
        self.tcp_nodelay = options.get("tcp_nodelay", False)

        self.ssl_keyfile = options.get("ssl_keyfile", None)
        self.ssl_certfile = options.get("ssl_certfile", None)
        self.ssl_trustfile = options.get("ssl_trustfile", None)
        self.ssl_skip_hostname_check = options.get("ssl_skip_hostname_check", False)
        self.client_properties = options.get("client_properties", {})

        self.options = options

        self.id = str(uuid4())
        self.session_counter = 0
        self.sessions = {}
        self._open = False
        self._connected = False
        self._transport_connected = False
        self._lock = RLock()
        self._condition = Condition(self._lock)
        self._waiter = Waiter(self._condition)
        self._modcount = Serial(0)
        self.error = None
        from driver import Driver

        self._driver = Driver(self)

 def testMapUUID(self):
   self.check("map", {"uuid": uuid4()})