Author: rajith
Date: Wed Jan 16 14:32:23 2008
New Revision: 612598
URL: http://svn.apache.org/viewvc?rev=612598&view=rev
Log:
This example is written using the AMQP Java API and interoperates with the
python and c++ client
Added:
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/DeclareQueue.java
(with props)
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/FannoutProducer.java
(with props)
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java
(with props)
Added:
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/DeclareQueue.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/DeclareQueue.java?rev=612598&view=auto
==============================================================================
---
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/DeclareQueue.java
(added)
+++
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/DeclareQueue.java
Wed Jan 16 14:32:23 2008
@@ -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();
+ }
+ }
+}
Propchange:
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/DeclareQueue.java
------------------------------------------------------------------------------
svn:executable = *
Added:
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/FannoutProducer.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/FannoutProducer.java?rev=612598&view=auto
==============================================================================
---
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/FannoutProducer.java
(added)
+++
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/FannoutProducer.java
Wed Jan 16 14:32:23 2008
@@ -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();
+ }
+ }
+
+}
Propchange:
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/FannoutProducer.java
------------------------------------------------------------------------------
svn:executable = *
Added:
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java?rev=612598&view=auto
==============================================================================
---
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java
(added)
+++
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java
Wed Jan 16 14:32:23 2008
@@ -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();
+ }
+ }
+
+}
Propchange:
incubator/qpid/trunk/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java
------------------------------------------------------------------------------
svn:executable = *
