メッセージングプログラミングリファレンス

import sys
from qpid.messaging import *

connection = Connection("localhost:5672")

try:
  connection.open()
  session = connection.session()
  
  sender = session.sender("amq.topic")
  receiver = session.receiver("amq.topic")
  
  message = Message("Hello World!")
  sender.send(message)
  
  fetchedmessage = receiver.fetch(timeout=1)
  print fetchedmessage.content
  session.acknowledge()
  
except MessagingError,m:
    print m

connection.close()

using System;
using Org.Apache.Qpid.Messaging;  

namespace Org.Apache.Qpid.Messaging {
    class Program {
        static void Main(string[] args) {
            String broker = args.Length > 0 ? args[0] : "localhost:5672";
            String address = args.Length > 1 ? args[1] : "amq.topic";

            Connection connection = null;
            try {
                connection = new Connection(broker);
                connection.Open();   
                Session session = connection.CreateSession();  

                Receiver receiver = session.CreateReceiver(address);   
                Sender sender = session.CreateSender(address);  

                sender.Send(new Message("Hello world!"));

                Message message = new Message();
                message = receiver.Fetch(DurationConstants.SECOND * 1);   
                Console.WriteLine("{0}", message.GetContentObject());
                session.Acknowledge();   

                connection.Close();   
            } catch (Exception e) {
                Console.WriteLine("Exception {0}.", e);
                if (connection != null)
                    connection.Close();
            }
        }
    }
}

#include <qpid/messaging/Connection.h>
#include <qpid/messaging/Message.h>
#include <qpid/messaging/Receiver.h>
#include <qpid/messaging/Sender.h>
#include <qpid/messaging/Session.h>

#include <iostream>

using namespace qpid::messaging;

int main(int argc, char** argv) {
    std::string broker = argc > 1 ? argv[1] : "localhost:5672";
    std::string address = argc > 2 ? argv[2] : "amq.topic";
    Connection connection(broker); 
    try {
        connection.open();  
        Session session = connection.createSession(); 

        Receiver receiver = session.createReceiver(address); 
        Sender sender = session.createSender(address); 

        sender.send(Message("Hello world!"));

        Message message = receiver.fetch(Duration::SECOND * 1); 
        std::cout << message.getContentObject() << std::endl;
        session.acknowledge(); 
        
        connection.close(); 
        return 0;
    } catch(const std::exception& error) {
        std::cerr << error.what() << std::endl;
        connection.close();
        return 1;   
    }
}

package org.apache.qpid.example.jmsexample.hello;

import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.util.Properties;

public class Hello {

  public Hello() {
  }

  public static void main(String[] args) {
    Hello producer = new Hello();
    producer.runTest();
  }

  private void runTest() {
    try {
      Properties properties = new Properties();
      properties.load(this.getClass().getResourceAsStream("hello.properties")); 
      Context context = new InitialContext(properties);   

      ConnectionFactory connectionFactory 
          = (ConnectionFactory) context.lookup("qpidConnectionfactory"); 
      Connection connection = connectionFactory.createConnection();  
      connection.start();  

      Session session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE); 
      Destination destination = (Destination) context.lookup("topicExchange");  

      MessageProducer messageProducer = session.createProducer(destination);  
      MessageConsumer messageConsumer = session.createConsumer(destination);  

      TextMessage message = session.createTextMessage("Hello world!");
      messageProducer.send(message);

      message = (TextMessage)messageConsumer.receive();    
      System.out.println(message.getText());

      connection.close();  
      context.close();   
    }
    catch (Exception exp) {
      exp.printStackTrace();
    }
  }
}

sender = session.sender("quick-publish; {create: always}")

Sender sender = session.createSender("quick-publish; {create: always}")

Receiver receiver = session.createReceiver('quick-publish');

receiver = session.receiver('quick-publish')

Receiver receiver = session.createReceiver('quick-publish; {mode: browse}');

receiver = session.receiver('quick-publish; {mode: browse}')

Receiver receiver = session.createReceiver("my-own-copies-please; {create: always, node: {type: 'queue'}}");

receiver = session.receiver("my-own-copies-please; {create: always, node: {type: 'queue'}}")

try {
  Receiver receiver = session.createReceiver("my-own-copies-please; {create: always, assert: always, node: {type: 'queue'}}");
} catch(const std::exception& error) {
  std::cerr << error.what() << std::endl;
}

try:
  receiver = session.receiver("my-own-copies-please; {create: always, assert: always, node: {type: 'queue'}}")
except MessagingError m:
  print m

sender = session.sender('engineering;{create: always, node:{type:topic, x-declare:{type:direct}}}')

try:
  sender = session.sender('engineering;{create: always, node:{type:topic, x-declare:{type:direct}}, assert: always}')
except MessagingError, m:
  print m

sender = session.sender('finance;{create:always, node: {type: topic, x-declare: {type: direct}}}')

sender = session.sender('finance/reports')
sender.send('Message to all consumers bound to finance with key reports')

sender = session.sender('finance; {assert: always, node: {type: topic}}')
msg = Message('Message to all consumers bound to finance with key reports')
msg.subject = 'reports'
sender.send(msg)

sender = session.sender('finance; {assert: always, node: {type: topic}}')
msg = Message('Message for reports and records')
    
msg.subject = 'reports'
sender.send(msg)
    
msg.subject = 'records'
sender.send(msg)

Receiver receiver = session.createReceiver("finance/reports")

receiver = session.receiver('finance/reports')

Receiver receiver = session.createReceiver("finance/quick-publish;{link:{name:my-subscription, x-declare:{exclusive:False}}}");

receiver = session.receiver('finance/quick-publish;{link:{name:my-subscription, x-declare:{exclusive:False}}}')

Receiver receiver = session.createReceiver("my-subscription;{create: always, node:{x-bindings: [{exchange: 'finance', key: 'quick-publish'}]}}");

receiver = session.receiver('my-subscription;{create: always, node:{x-bindings: [{exchange: 'finance', key: 'quick-publish'}]}}')

Receiver receiver = session.createReceiver("finance/quick-publish;{node: {capabilities:[shared]}, link: {name: 'my-subscription'}}");

tx = ssn.sender("myfanout; {create: always, node: {type: topic, x-declare: {exchange: myfanout, type: fanout}}}")

import sys
from qpid.messaging import *
con = Connection("localhost:5672")
con.open()
try:
  ssn = con.session()
  tx = ssn.sender("amq.fanout")
  tx.send("Hello to all consumers bound to the amq.fanout exchange")
finally:
  con.close()

rx = receiver("amq.fanout")

rx = receiver("shared-q")

txtopic = ssn.sender("news; {create: always, node: {type: topic}}")

import sys
from qpid.messaging import *
conn = Connection("localhost:5672")
conn.open()
try:
  ssn = conn.session()
  txnews = ssn.sender("news; {create: always, node: {type: topic}}")
  msg = Message("News about Europe") 
  msg.subject = "europe.news"
  txnews.send(msg)
  msg = Message("News about the US")
  msg.subject = "usa.news"
  txnews.send(msg)
finally:
  conn.close()

rxnews = ssn.receiver("news")

rxnews = ssn.receiver("amq.topic/#.news");

txheaders = ssn.sender("headers-match;{create: always, node: {type: topic, x-declare: {exchange: headers-match, type: headers}}}")

import sys
from qpid.messaging import *
conn =  Connection("localhost:5672")
conn.open()
try:
  ssn = conn.session()
  txheaders = ssn.sender("amq.match")
  msg = Message("Headers Exchange message")
  msg.properties['header1'] = 'value1'
  txheaders.send(msg)
finally:
  ssn.close()

tx = ssn.sender("myxml; {create: always, node: {type: topic, x-declare: {exchange: myxml, type: xml}}}")

#!/usr/bin/python
import sys
from qpid.messaging import *

conn = Connection("localhost:5672")
conn.open()
try:
  ssn = conn.session()
  tx = ssn.sender("myxml/weather; {create: always, node: {type: topic, x-declare: {exchange: myxml, type: xml}}}")

  xquerystr = 'let $w := ./weather '
  xquerystr += "return $w/station = 'Raleigh-Durham International Airport (KRDU)' "
  xquerystr += 'and $w/temperature_f > 50 '
  xquerystr += 'and $w/temperature_f - $w/dewpoint > 5 '
  xquerystr += 'and $w/wind_speed_mph > 7 '
  xquerystr += 'and $w/wind_speed_mph < 20'

  rxaddr = 'myxmlq; {create: always, '
  rxaddr += 'link: {x-bindings: [{exchange: myxml, '
  rxaddr += 'key: weather, '
  rxaddr += 'arguments: {xquery: "' + xquerystr + '"'
  rxaddr += '}}]}}'
 
  rx = ssn.receiver(rxaddr)

  msgstr = '<weather>'
  msgstr += '<station>Raleigh-Durham International Airport (KRDU)</station>'
  msgstr += '<wind_speed_mph>16</wind_speed_mph>'
  msgstr += '<temperature_f>70</temperature_f>'
  msgstr += '<dewpoint>35</dewpoint>'
  msgstr += '</weather>'

  msg = Message(msgstr)
        
  tx.send(msg)

  rxmsg = rx.fetch(timeout=1)
  print rxmsg

  ssn.acknowledge()

finally:
  conn.close()

import sys
from qpid.messaging import *     
...        
msg = Message('This is the message content')
msg.content = 'Message content can be assigned like this'
msg.properties['header-key'] = 'value'

tx = ssn.sender('amq.topic')

# msg.subject set by sender for routing purposes
tx.send(msg)

msg.subject = 'Messaging Routing Key can also be manually set'
# beware that this will interfere with sender-object-based routing

import sys
from qpid.messaging import *

def msgfetch(rx):
  try:
    msg = rx.fetch(timeout=1)
  except MessagingError, m:
    msg = m
  return msg
  
connection = Connection("localhost:5672")
connection.open()
try:
  session = connection.session()
  tx = session.sender("browse-acquire-demo;{create:always}")
  rxbrowse1 = session.receiver("browse-acquire-demo;{mode:browse}")
  rxbrowse2 = session.receiver("browse-acquire-demo;{mode:browse}")
  rxbrowse3 = session.receiver("browse-acquire-demo;{mode:browse}")
  rxacquire = session.receiver("browse-acquire-demo")
  
  tx.send("Hello World")
  
  print "\nBrowser 1 saw message:"
  print msgfetch(rxbrowse1)
      
  print "Browser 1 then saw message:"
  print msgfetch(rxbrowse1)

  print "\nBrowser 2 saw message:"
  print msgfetch(rxbrowse2)

  print "Browser 2 then saw message:"
  print msgfetch(rxbrowse2)

  print "\nAcquired message:"
  print msgfetch(rxacquire)
  
  print "\nBrowser 3 saw message:"
  print msgfetch(rxbrowse3)
  
except MessagingError, m:
  print m
finally:
  connection.close()

connection.open()
try:
  session=connection.session()

  rxacquire2 = session.receiver("browse-acquire-demo")
  print "\nAcquirer 2 saw message:"
  print msgfetch(rxacquire2)
except MessagingError, m:
  print m
finally:
  session.acknowledge()
  connection.close()

rxacquire = session.receiver("browse-acquire-demo; {link:{reliability: unreliable}")

import sys
from qpid.messaging import *

def msgfetch(rx):
  try:
    msg = rx.fetch(timeout=1)
  except MessagingError, m:
    msg = m
  return msg
  
linktype=""
while linktype != "R" and linktype !="U":
  response = raw_input("Use (R)eliable or (U)nreliable link [R/U]?")
  linktype = response.upper()

connection = Connection("localhost:5672")
connection.open()
try:
  session = connection.session()
  tx = session.sender("browse-acquire-demo;{create: always}")
  rxbrowse1 = session.receiver("browse-acquire-demo;{mode:browse}")
  rxbrowse2 = session.receiver("browse-acquire-demo;{mode:browse}")
  rxbrowse3 = session.receiver("browse-acquire-demo;{mode:browse}")
  if linktype == "R":
    rxacquire = session.receiver("browse-acquire-demo") 
  else:
    rxacquire = session.receiver("browse-acquire-demo; {link:{reliability:unreliable}}")
  
  tx.send("Hello World")
  
  print "\nBrowser 1 saw message:"
  print msgfetch(rxbrowse1)
      
  print "Browser 1 then saw message:"
  print msgfetch(rxbrowse1)

  print "\nBrowser 2 saw message:"
  print msgfetch(rxbrowse2)

  print "Browser 2 then saw message:"
  print msgfetch(rxbrowse2)

  print "\nAcquired message:"
  print msgfetch(rxacquire)
 
  rxacquire.close()
 
  print "\nBrowser 3 saw message:"
  print msgfetch(rxbrowse3)
  
except MessagingError, m:
  print m
finally:
  connection.close()

connection.open()
try:
  session=connection.session()

  rxacquire2 = session.receiver("browse-acquire-demo")
  print "\nAcquirer 2 saw message:"
  print msgfetch(rxacquire2)  

except MessagingError, m:
  print m
finally:
  session.acknowledge()
  connection.close()

msg = rx.fetch(timeout = 1)
    
if msg.content == "something we don't like":
  ssn.acknowledge(msg, Disposition(REJECTED))
else:
  ssn.acknowledge(msg)

receiver1 = session.receiver(address1)
receiver1.capacity = 10
receiver2 = session.receiver(address)
receiver2.capacity = 10
message = session.next_receiver().fetch()
print message.content
session.acknowledge()

Receiver receiver1 = session.createReceiver(address1);
receiver1.setCapacity(10);
Receiver receiver2 = session.createReceiver(address2);
receiver2.setCapacity(10);

Message message =  session.nextReceiver().fetch();
std::cout << message.getContent() << std::endl;
session.acknowledge(); // acknowledge message receipt

Receiver receiver1 = session.CreateReceiver(address1);
receiver1.SetCapacity(10);
Receiver receiver2 = session.CreateReceiver(address2);
receiver2.SetCapacity(10);

Message message = new Message();
message =  session.NextReceiver().Fetch();
Console.WriteLine("{0}", message.GetContent());
session.Acknowledge();

responsequeue = session.receiver('my-response-queue; {create:always, node:{x-declare:{auto-delete:True}}}')

import sys
from qpid.messaging import *

connection=Connection("localhost:5672")
connection.open()
try:
  session=connection.session()
  tx=session.sender("test-queue; {create:always, node:{x-declare:{auto-delete:True}}}")
  tx.send("test message!")
  x = raw_input("Press Enter to continue")
  tx.send("test message 2")
except MessagingError, m:
  print m
connection.close()

import sys
from qpid.messaging import *

connection=Connection("localhost:5672")
connection.open()
try:
  session=connection.session()
  rx=session.receiver("test-queue")
  print rx.fetch(timeout = 1)
  session.acknowledge()  
except MessagingError,m:
  print m
connection.close()

import sys
from qpid.messaging import *

connection=Connection("localhost:5672")
connection.open()
try:
  session=connection.session()
  tx=session.sender("test-queue; {create:always, node:{x-declare:{auto-delete:True}}}")
  rx=session.receiver("test-queue")
  tx.send("test message!")
  x = raw_input("Press Enter to continue")
  tx.send("test message 2")
  x = raw_input("Press Enter to continue")
except MessagingError, m:
  print m
connection.close()

myLastValueQueue = mySession.sender("stock-ticker;{create:always, node:{type:queue, x-declare:{arguments:{'qpid.last_value_queue_key': 'stock-symbol'}}}}")

import sys
from qpid.messaging import *

connection = Connection("localhost:5672")
connection.open()

session = connection.session()

stockSender = session.sender("stock-ticker;{create:always, node:{type:queue, x-declare:{arguments:{'qpid.last_value_queue_key': 'stock-symbol'}}}}")

msg1 = Message("10")
msg1.properties = {'stock-symbol':'RHT'}

msg2 = Message("10")  
msg2.properties = {'stock-symbol':'JAVA'}

msg3 = Message("10")
msg3.properties = {'stock-symbol':'MSFT'}

msg4 = Message("12")
msg4.properties = {'stock-symbol':'RHT'}

controlSender = session.sender("control-queue;{create:always, node:{type:queue}}")

stockSender.send(msg1)
controlSender.send(msg1)

stockSender.send(msg2)
controlSender.send(msg2)

stockSender.send(msg3)
controlSender.send(msg3)

stockSender.send(msg4)
controlSender.send(msg4)

stockBrowser = session.receiver("stock-ticker; {mode:browse}")
controlBrowser = session.receiver("control-queue; {mode:browse}")

stockBrowser.capacity = 20
controlBrowser.capacity = 20

sleep 10

from time import sleep

while True:
  try:
    msg = stockBrowser.fetch(timeout = 10)
    print msg.properties["stock-symbol"] + ":" + msg.content
  except Empty:
    break

print "Last Value Queue has " + str(stockBrowser.available()) + " messages"

print "\nLast Value Queue messages:"
      
for x in range(stockBrowser.available()):
  try:
    msg = stockBrowser.fetch(timeout = 1)
    print msg.properties["stock-symbol"] + ":" + msg.content
  except MessagingError, m:
    pass
      
print "Control Queue has " + str(controlBrowser.available()) + " messages"
      
print "\nControl Queue messages:"
for x in range(controlBrowser.available()):
  try:
    msg = controlBrowser.fetch(timeout = 1)
    print msg.properties["stock-symbol"] + ":" + msg.content
  except MessagingError, m:
    pass

session.acknowledge()
connection.close()

import sys
from qpid.messaging import *
from time import sleep

connection = Connection("localhost:5672")
try:
  connection.open()
  session = connection.session()

  stockSender = session.sender("stock-ticker;{create:always, node:{type:queue, x-declare:{arguments:{'qpid.last_value_queue_key': 'stock-symbol'}}}}")
  controlSender = session.sender("control-queue;{create:always, node:{type:queue}}")

  stockBrowser = session.receiver("stock-ticker;{mode:browse}")
  controlBrowser = session.receiver("control-queue;{mode:browse}")
  controlBrowser = session.receiver("control-queue")

  msg1 = Message("10")
  msg1.properties = {'stock-symbol':'RHT'}

  msg2 = Message("10")
  msg2.properties = {'stock-symbol':'JAVA'}

  msg3 = Message("10")
  msg3.properties = {'stock-symbol':'MSFT'}

  msg4 = Message("12")
  msg4.properties = {'stock-symbol':'RHT'}

  stockSender.send(msg1)
  controlSender.send(msg1)

  stockSender.send(msg2)
  controlSender.send(msg2)

  stockSender.send(msg3)
  controlSender.send(msg3)

  stockSender.send(msg4)
  controlSender.send(msg4)

  stockBrowser.capacity = 20
  controlBrowser.capacity = 20

  sleep(10)

  print "\nLast Value Queue has " + str(stockBrowser.available()) + " messages"
      
  print "Last Value Queue messages:"
      
  for x in range(stockBrowser.available()):
    try:
      msg = stockBrowser.fetch(timeout = 1)
      print msg.properties["stock-symbol"] + ":" + msg.content
    except MessagingError, m:
      pass
      
  print "\nControl Queue has " + str(controlBrowser.available()) + " messages"
      
  print "Control Queue messages:"
      
    for x in range(controlBrowser.available()):
    try:
      msg = controlBrowser.fetch(timeout = 1)
      print msg.properties["stock-symbol"] + ":" + msg.content
    except MessagingError, m:
      pass

  session.acknowledge()
      
except MessagingError,m:
  print m
finally:
  connection.close()

sender = session.sender('my-queue; {create: always, node:{x-declare:{arguments:{qpid.priorities:10}}}}')

#!/usr/bin/python

import sys
from qpid.messaging import *

connection = Connection("localhost:5672")
connection.open()
try:
  ssn = connection.session()
  
  x = 0
  print "\n"
  while True:
    print "Create queue with 2 or 10 priority levels?"
    x = raw_input()
    if (x == "2") or (x == "10"):
      break

  tx = ssn.sender("nonpriority-demo-queue; {create: always, node: {type: 'queue'}}")
  print "Creating a priority queue with " + x + " priority levels:"
  address =  "priority-demo-queue; {create: always, "
  address = address + "node:{x-declare: {auto-delete:True, "
  address = address + "arguments: {'qpid.priorities': "
  address = address + x + "}}}}"
  print address
  txpriority = ssn.sender(address)

  rx = ssn.receiver('nonpriority-demo-queue')
  rxpriority = ssn.receiver("priority-demo-queue")
  rxbrowse = ssn.receiver("priority-demo-queue; {mode: browse}")

  print "\nPress Enter to continue\n"
  x = raw_input()

  print "First message sent:"
  msg = Message("priority 1")
  msg.priority = 1
  tx.send(msg)
  txpriority.send(msg)
  print msg

  print "Second message sent:"
  msg = Message('priority 4')
  msg.priority = 4
  tx.send(msg)
  txpriority.send(msg)
  print msg

  print "\nPress Enter to continue\n"
  x = raw_input()
  print "BROWSE PRIORITY QUEUE"
  print "First browse in priority queue:"
  print rxbrowse.fetch()

  print "Second browse in priority queue:"
  print rxbrowse.fetch()

  print "\nPress Enter to continue\n"
  x = raw_input()

  print "ACQUIRE PRIORITY QUEUE"
  print "First message in priority queue:"
  print rxpriority.fetch()

  print "Second message in priority queue:"
  print rxpriority.fetch()

  print "\nPress Enter to continue\n"
  x = raw_input()

  print "ACQUIRE NON-PRIORITY QUEUE"
  print "First message in non-priority queue:"
  print rx.fetch()

  print "Second message in non-priority queue:"
  print rx.fetch()

  ssn.acknowledge()
finally:
  connection.close()

Sender sender = session.createSender('my-queue; {create: always, node:{x-declare:{arguments:{qpid.priorities:10, x-qpid-fairshare: 5}}}}')

Sender sender = session.createSender('my-queue; {create: always, node:{x-declare:{arguments:{qpid.priorities:10, x-qpid-fairshare-0: 3, x-qpid-fairshare-1: 5, x-qpid-fairshare-2: 3, x-qpid-fairshare-3: 2, x-qpid-fairshare-4: 4, x-qpid-fairshare-5: 5, x-qpid-fairshare-6: 5, x-qpid-fairshare-7: 3, x-qpid-fairshare-8: 5, x-qpid-fairshare-9: 4, x-qpid-priorities: 10}}}}')

groupedSender = session.sender("my-grouped-msg-queue; {create: always, node: {x-declare: {auto-delete: True, arguments: {'qpid.group_header_key': 'msgGroupID', 'qpid.shared_msg_group': 1}}}}")

Sender groupedSender = session.createSender("my-grouped-msg-queue; {create:always, node: {x-declare: {auto-delete: True, arguments: {'qpid.group_header_key':'msgGroupID', 'qpid.shared_msg_group':1}}}}")

import sys
from qpid.messaging import *

def sendmsg(group, num):
# send the message to the broker and add it to our in-memory representation of the broker queue
  global memoryqueue
  global tx

  msg = Message(group + num)
  msg.properties = {'ourGroupID': group}

  tx.send(msg)
  memoryqueue.append(group + num)

def pullmsg(consumer):
# fetch a message from the broker and print it to the console  
  global counter
  global memoryqueue

  msg = consumers[consumer - 1].fetch(timeout = 1)
  
  print "\nQueued message: " + memoryqueue[counter]
  print "Consumer " + str(consumer) + " got: " + msg.content

  counter +=1
  return msg
  
# Two connections are used to simulate two distinct consumers  
connection = Connection("localhost:5672")
connection2 = Connection("localhost:5672")
connection.open()
connection2.open()

try:
  session = connection.session()
  session2 = connection2.session()
  
  x = raw_input('Enable message grouping [Y/n]?')

  if x == 'N' or x == 'n':
  
    # Create the queue without message groups
    tx = session.sender("test-nogroup-queue; {create: always, node:{x-declare:{auto-delete:True}}}")
    rx1 = session.receiver("test-nogroup-queue")
    rx2 = session2.receiver("test-nogroup-queue")
  
    print "\nMessage grouping is disabled"
    msggroup = False
  
  else:
  
    # Create the queue with message groups enabled
    tx = session.sender("test-group-queue; {create: always, node:{x-declare:{auto-delete: True, arguments: {'qpid.group_header_key': 'ourGroupID', 'qpid.shared_msg_group' : 1}}}}")
    rx1 = session.receiver("test-group-queue")
    rx2 = session2.receiver("test-group-queue")
  
    print "\nMessage grouping is enabled"
    msggroup = True

# Put the receivers in an array so we can use a function to fetch messages  
  consumers = []
  consumers.append(rx1)
  consumers.append(rx2)

  print "Sending interleaved messages from two different groups to the queue..."

# We create an in-memory picture of the queue, to see what order the messages are on the broker
  memoryqueue = []

  sendmsg('A', '1')
  sendmsg('B', '1')
  sendmsg('B', '2')
  sendmsg('A', '2')
  sendmsg('B', '3')
  sendmsg('A', '3')

  counter = 0 
  pullmsg(1)
  pullmsg(2)  
  
  if msggroup:
    print "\nConsumer 1 now owns message group A. Consumer 2 now owns message group B."  

  msgc1 = pullmsg(1)
  msgc2 = pullmsg(2)

  if msggroup:
    print "\nThe consumers will now acknowledge all the messages, or only the last one."
    resp = raw_input('Should they acknowlege all messages? [Y/n]')
  
    if resp == 'N' or resp == 'n':
      print "\nAcknowledging only part of the group. The consumers retain ownership of the group. This is an anti-pattern! See the source code comments for details."
  
      session.acknowledge(msgc1)
      session2.acknowledge(msgc2)
      antipattern = True
  
      # Acknowledging only part of a group is an anti-pattern. Messages are grouped to ensure that a single consumer can deal with the whole group. If this consumer now fails before completing the rest of the group, the unacknowledged messages in the group will be released and redelivered by the broker, but the acknowledged messages in the group are now missing in action!
  
    else:
      print "\nAcknowledging all fetched messages. The consumers will release ownership of the groups."
      session.acknowledge()
      session2.acknowledge()
      antipattern = False 
  
    print "\nPulling more messages from the queue:"

  pullmsg(1)
  pullmsg(2)
  if msggroup:
    if antipattern == False:
      print "\nConsumer 1 now owns message group B. Consumer 2 now owns message group A."
    print "\nSending some more messages to the queue..."
  
  sendmsg('B', '4')
  sendmsg('B', '5')
  sendmsg('A', '4')
  sendmsg('A', '5')
 
  pullmsg(1)
  pullmsg(2)
  pullmsg(1)
  pullmsg(2)

finally:
  connection.close()
  connection2.close()

sender.send(message, sync = False)

sender.send(message, false)

sender.capacity = 20

sender.setCapacity(20)

  if (sender.getAvailable() > 0)
      sender.send(message, false)
   // else drop the message

if sender.available() > 0:
  sender.send(message, sync=False)
else:
  # drop the message

sender.setCapacity(SOME_LARGE_NUMBER)

sender.capacity = SOME_LARGE_NUMBER

sender.setCapacity(MY_CAPACITY); 

// Later
bool resend = true;
while (resend)
{
  if (sender.getAvailable()>0)
  {
    sender.send(message,false);
    resend = false;
  }
  else
  {
    // May wish to do nothing here 
    // or send to log file
    std::cout << "Warning: Capacity \ full. Retry" << std::endl;
   }
}
// Later
if (sender.getUnsettled())
{
    session.sync();
}

snd.capacity = MY_CAPACITY

# Later

resend = True
while (resend):
  if (snd.available()>0):
    snd.send(msg, sync = False)
    resend = False
  else:
    print "Warning: Capacity full" 

# Later   
    if (snd.unsettled()):
      ssn.sync()

sender = session.sender("amq.topic;{link: {'reliability': 'unreliable'}}")

import sys
from qpid.messaging import *

connection = Connection("localhost:5672")

try:
  connection.open()
  session = connection.session()

  linktype=""
  while linktype != "R" and linktype !="U":
    response = raw_input("Use (R)eliable or (U)nreliable link [R/U]? ")
    linktype = response.upper()  
  
  if linktype == "U":
    sender = session.sender("amq.topic;{link: {'reliability': 'unreliable'}}")
  else:
    sender = session.sender("amq.topic")

  message = Message("Hello World:")
  print sender.capacity
  sender.capacity = 5
  for x in range (1000):
    if sender.available() == 0:
      print "Sender is blocking..."
    sender.send("Hello World: " + str(x), sync=False)
    print str(x) +" : " + str(sender.unsettled()) + " : " + str(sender.available()) 
  
except MessagingError,m:
    print m
finally:
  connection.close()

import sys
from qpid.messaging import *
  
connection = Connection("localhost:5672")
connection.open()
ssn = connection.session()
    
prefetchingReceiver = ssn.receiver("testqueue; {create:always}");
prefetchingReceiver.capacity = 100

session.acknowledge()

msg = rx.fetch(timeout = 1)
session.acknowledge(msg)

session.acknowledge(msg, sync = False)

sender = session.sender("amq.topic;{link: {'reliability': 'unreliable'}}")

tx = ssn.sender("my-queue; {create: always, node: {x-declare: {'auto-delete': True, arguments:{'qpid.max_count': 5000, 'qpid.max_size': 204800000}}}}")

tx = ssn.sender("my-queue; {create: always, node: {x-declare: {'auto-delete': True, arguments:{'qpid.max_count': 5000, 'qpid.max_size': 204800, 'qpid.policy_type': 'ring'}}}}")

conn = Connection.establish("localhost:5672")
session = conn.session()
rcv = session.receiver("qmf.default.topic/agent.ind.event.org_apache_qpid_broker.queueThresholdExceeded.#")
while True:
  event = rcv.fetch()
  print "Threshold exceeded on queue %s" % event.content[0]["_values"]["qName"]
  print "     at a depth of %s messages, %s bytes" % (event.content[0]["_values"]["msgDepth"], event.content[0]["_values"]["byteDepth"])
  session.acknowledge()

Sender sender = session.createSender("important-messages; {create:always, node:{durable: True}")

newqueue = session.sender("important-messages; {create:always, node:{durable: True}")

tx = ssn.sender("my-queue;{create: always, node: {durable: True, x-declare: {arguments: {'qpid.file_size': 20, 'qpid.file_count': 12}}}}")

Connection connection = new Connection(broker);
Session session =  connection.createTransactionalSession();
...
if (smellsOk())
   session.Commit();
else 
   session.Rollback();

Connection connection(broker);
Session session =  connection.createTransactionalSession();
...
if (smellsOk())
   session.commit();
else 
   session.rollback();