diff options
Diffstat (limited to 'java/client/example/src')
3 files changed, 223 insertions, 0 deletions
diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/DeclareQueue.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/DeclareQueue.java new file mode 100755 index 0000000000..cb49e9d28f --- /dev/null +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/DeclareQueue.java @@ -0,0 +1,51 @@ +package org.apache.qpid.example.amqpexample.fanout; + +import org.apache.qpidity.nclient.Client; +import org.apache.qpidity.nclient.Connection; +import org.apache.qpidity.nclient.Session; + +/** + * This creates a queue a queue and binds it to the + * amq.direct exchange + * + */ +public class DeclareQueue +{ + + public static void main(String[] args) + { + // Create connection + Connection con = Client.createConnection(); + try + { + con.connect("localhost", 5672, "test", "guest", "guest"); + } + catch(Exception e) + { + System.out.print("Error connecting to broker"); + e.printStackTrace(); + } + + // Create session + Session session = con.createSession(0); + + // declare and bind queue + session.queueDeclare("message_queue", null, null); + session.queueBind("message_queue", "amq.fanout",null, null); + + // confirm completion + session.sync(); + + //cleanup + session.sessionClose(); + try + { + con.close(); + } + catch(Exception e) + { + System.out.print("Error closing broker connection"); + e.printStackTrace(); + } + } +} diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/FannoutProducer.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/FannoutProducer.java new file mode 100755 index 0000000000..1dfc468cd4 --- /dev/null +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/FannoutProducer.java @@ -0,0 +1,62 @@ +package org.apache.qpid.example.amqpexample.fanout; + +import org.apache.qpidity.nclient.Client; +import org.apache.qpidity.nclient.Connection; +import org.apache.qpidity.nclient.Session; +import org.apache.qpidity.transport.DeliveryProperties; + +public class FannoutProducer +{ + /** + * This sends 10 messages to the + * amq.fannout exchange + */ + public static void main(String[] args) + { + // Create connection + Connection con = Client.createConnection(); + try + { + con.connect("localhost", 5672, "test", "guest", "guest"); + } + catch(Exception e) + { + System.out.print("Error connecting to broker"); + e.printStackTrace(); + } + + // Create session + Session session = con.createSession(0); + DeliveryProperties deliveryProps = new DeliveryProperties(); + deliveryProps.setRoutingKey("routing_key"); + + for (int i=0; i<10; i++) + { + session.messageTransfer("amq.fanout", Session.TRANSFER_CONFIRM_MODE_REQUIRED, Session.TRANSFER_ACQUIRE_MODE_PRE_ACQUIRE); + session.header(deliveryProps); + session.data("Message " + i); + session.endData(); + } + + session.messageTransfer("amq.fanout", Session.TRANSFER_CONFIRM_MODE_REQUIRED, Session.TRANSFER_ACQUIRE_MODE_PRE_ACQUIRE); + session.header(deliveryProps); + session.data("That's all, folks!"); + session.endData(); + + // confirm completion + session.sync(); + + //cleanup + session.sessionClose(); + try + { + con.close(); + } + catch(Exception e) + { + System.out.print("Error closing broker connection"); + e.printStackTrace(); + } + } + +} diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java new file mode 100755 index 0000000000..cf12e80b5c --- /dev/null +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java @@ -0,0 +1,110 @@ +package org.apache.qpid.example.amqpexample.fanout; + +import java.nio.ByteBuffer; + +import org.apache.qpidity.api.Message; +import org.apache.qpidity.nclient.Client; +import org.apache.qpidity.nclient.Connection; +import org.apache.qpidity.nclient.Session; +import org.apache.qpidity.nclient.util.MessageListener; +import org.apache.qpidity.nclient.util.MessagePartListenerAdapter; + +/** + * This listens to messages on a queue and terminates + * when it sees the final message + * + */ +public class Listener implements MessageListener +{ + boolean finish = false; + + public void onMessage(Message m) + { + String data = null; + + try + { + ByteBuffer buf = m.readData(); + byte[] b = new byte[buf.remaining()]; + buf.get(b); + data = new String(b); + } + catch(Exception e) + { + System.out.print("Error reading message"); + e.printStackTrace(); + } + + System.out.println("Message: " + data); + + if (data != null && data.equals("That's all, folks!")) + { + finish = true; + } + } + + public boolean isFinished() + { + return finish; + } + + /** + * This sends 10 messages to the + * amq.direct exchange using the + * routing key as "routing_key" + * + */ + public static void main(String[] args) + { + // Create connection + Connection con = Client.createConnection(); + try + { + con.connect("localhost", 5672, "test", "guest", "guest"); + } + catch(Exception e) + { + System.out.print("Error connecting to broker"); + e.printStackTrace(); + } + + // Create session + Session session = con.createSession(0); + + // Create an instance of the listener + Listener listener = new Listener(); + + // create a subscription + session.messageSubscribe("message_queue", + "listener_destination", + Session.TRANSFER_CONFIRM_MODE_NOT_REQUIRED, + Session.TRANSFER_ACQUIRE_MODE_PRE_ACQUIRE, + new MessagePartListenerAdapter(listener), null); + + + // issue credits + session.messageFlow("listener_destination", Session.MESSAGE_FLOW_UNIT_BYTE, Session.MESSAGE_FLOW_MAX_BYTES); + session.messageFlow("listener_destination", Session.MESSAGE_FLOW_UNIT_MESSAGE, 11); + + // confirm completion + session.sync(); + + // check to see if we have received all the messages + while (!listener.isFinished()){} + System.out.println("Shutting down listener for listener_destination"); + session.messageCancel("listener_destination"); + + //cleanup + session.sessionClose(); + try + { + con.close(); + } + catch(Exception e) + { + System.out.print("Error closing broker connection"); + e.printStackTrace(); + } + } + +} |