From 248f1fe188fe2307b9dcf2c87a83b653eaa1920c Mon Sep 17 00:00:00 2001 From: "Rafael H. Schloming" Date: Sat, 26 Dec 2009 12:42:57 +0000 Subject: synchronized with trunk except for ruby dir git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qpid.rnr@893970 13f79535-47bb-0310-9956-ffa450edef68 --- java/client/example/src/main/java/README.txt | 9 +- .../example/amqpexample/direct/DeclareQueue.java | 50 ++-- .../example/amqpexample/direct/DirectProducer.java | 85 +++---- .../qpid/example/amqpexample/direct/Listener.java | 114 ++++----- .../example/amqpexample/fanout/DeclareQueue.java | 50 ++-- .../amqpexample/fanout/FannoutProducer.java | 50 ++-- .../qpid/example/amqpexample/fanout/Listener.java | 108 ++++----- .../example/amqpexample/headers/DeclareQueue.java | 74 ++++++ .../qpid/example/amqpexample/headers/Listener.java | 107 +++++++++ .../qpid/example/amqpexample/headers/Producer.java | 88 +++++++ .../example/amqpexample/lvq/DeclareLVQueue.java | 64 +++++ .../qpid/example/amqpexample/lvq/Listener.java | 115 +++++++++ .../qpid/example/amqpexample/lvq/Producer.java | 69 ++++++ .../example/amqpexample/pubsub/TopicListener.java | 129 +++++----- .../example/amqpexample/pubsub/TopicPublisher.java | 51 ++-- .../apache/qpid/example/jmsexample/direct/verify | 23 +- .../qpid/example/jmsexample/direct/verify_cpp_java | 2 +- .../qpid/example/jmsexample/direct/verify_java_cpp | 2 +- .../example/jmsexample/direct/verify_java_python | 4 +- .../example/jmsexample/direct/verify_python_java | 4 +- .../apache/qpid/example/jmsexample/fanout/verify | 23 +- .../qpid/example/jmsexample/fanout/verify_cpp_java | 2 +- .../qpid/example/jmsexample/fanout/verify_java_cpp | 2 +- .../example/jmsexample/fanout/verify_java_python | 4 +- .../example/jmsexample/fanout/verify_python_java | 4 +- .../apache/qpid/example/jmsexample/pubsub/verify | 23 +- .../qpid/example/jmsexample/pubsub/verify_cpp_java | 2 +- .../qpid/example/jmsexample/pubsub/verify_java_cpp | 2 +- .../example/jmsexample/pubsub/verify_java_python | 4 +- .../example/jmsexample/pubsub/verify_python_java | 4 +- .../qpid/example/jmsexample/requestResponse/verify | 23 +- .../jmsexample/requestResponse/verify_cpp_java | 2 +- .../jmsexample/requestResponse/verify_java_cpp | 4 +- .../jmsexample/requestResponse/verify_java_python | 4 +- .../jmsexample/requestResponse/verify_python_java | 4 +- .../qpid/example/publisher/FileMessageFactory.java | 6 +- .../example/publisher/MultiMessageDispatcher.java | 141 +++++++++++ .../apache/qpid/example/publisher/Publisher.java | 47 +++- .../qpid/example/publisher/TopicPublisher.java | 59 +++++ .../apache/qpid/example/simple/reqresp/Client.java | 263 +++++++++++++++++++++ .../apache/qpid/example/simple/reqresp/Server.java | 236 ++++++++++++++++++ java/client/example/src/main/java/runSample.sh | 22 +- 42 files changed, 1689 insertions(+), 390 deletions(-) create mode 100644 java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/DeclareQueue.java create mode 100644 java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Listener.java create mode 100644 java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Producer.java create mode 100644 java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/DeclareLVQueue.java create mode 100644 java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Listener.java create mode 100644 java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Producer.java create mode 100644 java/client/example/src/main/java/org/apache/qpid/example/publisher/MultiMessageDispatcher.java create mode 100644 java/client/example/src/main/java/org/apache/qpid/example/publisher/TopicPublisher.java create mode 100644 java/client/example/src/main/java/org/apache/qpid/example/simple/reqresp/Client.java create mode 100644 java/client/example/src/main/java/org/apache/qpid/example/simple/reqresp/Server.java (limited to 'java/client/example/src/main') diff --git a/java/client/example/src/main/java/README.txt b/java/client/example/src/main/java/README.txt index c8fe6a76ad..c99c3fddec 100644 --- a/java/client/example/src/main/java/README.txt +++ b/java/client/example/src/main/java/README.txt @@ -3,15 +3,16 @@ variables, QPID_HOME and QPID_SAMPLE. If not the default values will be used. QPID_HOME --------- -This is the distribution directory. You would have set the QPID_HOME when you -started the Qpid Java broker. +This is the directory that contains the QPID distribution. If you are running the Qpid +Java broker on the same machine as the examples, you have already set QPID_HOME to this +directory. default: /usr/share/java/ QPID_SAMPLE ----------- -This is the parent directory of the directory in which you find the runSample.sh +This is the parent directory of the 'java' directory in which you find 'runSample.sh' (Ex:- $QPID_SRC_HOME/java/client/example/src/main) -default: /usr/share/doc/rhm-0.2 +default: $PWD diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/DeclareQueue.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/DeclareQueue.java index b16a7fa7c3..38073cb7f2 100755 --- a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/DeclareQueue.java +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/DeclareQueue.java @@ -1,8 +1,28 @@ package org.apache.qpid.example.amqpexample.direct; +/* + * + * 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. + * + */ + -import org.apache.qpid.nclient.Client; -import org.apache.qpid.nclient.Connection; -import org.apache.qpid.nclient.Session; +import org.apache.qpid.transport.Connection; +import org.apache.qpid.transport.Session; /** * This creates a queue a queue and binds it to the @@ -15,16 +35,8 @@ 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(); - } + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); // Create session Session session = con.createSession(0); @@ -37,15 +49,7 @@ public class DeclareQueue session.sync(); //cleanup - session.sessionDetach(session.getName()); - try - { - con.close(); - } - catch(Exception e) - { - System.out.print("Error closing broker connection"); - e.printStackTrace(); - } + session.close(); + con.close(); } } diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/DirectProducer.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/DirectProducer.java index 2793e567ea..2234eb22da 100755 --- a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/DirectProducer.java +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/DirectProducer.java @@ -1,65 +1,44 @@ package org.apache.qpid.example.amqpexample.direct; +/* + * + * 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. + * + */ + import java.nio.ByteBuffer; import org.apache.qpid.api.Message; -import org.apache.qpid.nclient.Client; -import org.apache.qpid.nclient.Connection; -import org.apache.qpid.nclient.Session; -import org.apache.qpid.nclient.util.MessageListener; +import org.apache.qpid.transport.Connection; import org.apache.qpid.transport.DeliveryProperties; import org.apache.qpid.transport.Header; import org.apache.qpid.transport.MessageAcceptMode; import org.apache.qpid.transport.MessageAcquireMode; +import org.apache.qpid.transport.Session; -public class DirectProducer implements MessageListener +public class DirectProducer { - 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; - } 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(); - } + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); // Create session Session session = con.createSession(0); @@ -81,16 +60,8 @@ public class DirectProducer implements MessageListener session.sync(); //cleanup - session.sessionDetach(session.getName()); - try - { - con.close(); - } - catch(Exception e) - { - System.out.print("Error closing broker connection"); - e.printStackTrace(); - } + session.close(); + con.close(); } } diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/Listener.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/Listener.java index 081fbf2521..1ac3e85f7a 100755 --- a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/Listener.java +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/Listener.java @@ -1,115 +1,105 @@ package org.apache.qpid.example.amqpexample.direct; +/* + * + * 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. + * + */ -import java.nio.ByteBuffer; -import org.apache.qpid.api.Message; -import org.apache.qpid.nclient.Client; -import org.apache.qpid.nclient.Connection; -import org.apache.qpid.nclient.Session; -import org.apache.qpid.nclient.util.MessageListener; -import org.apache.qpid.nclient.util.MessagePartListenerAdapter; +import java.nio.ByteBuffer; +import org.apache.qpid.transport.Connection; +import org.apache.qpid.transport.MessageAcceptMode; +import org.apache.qpid.transport.MessageAcquireMode; import org.apache.qpid.transport.MessageCreditUnit; +import org.apache.qpid.transport.MessageTransfer; +import org.apache.qpid.transport.Session; +import org.apache.qpid.transport.SessionException; +import org.apache.qpid.transport.SessionListener; /** * This listens to messages on a queue and terminates * when it sees the final message * */ -public class Listener implements MessageListener +public class Listener implements SessionListener { - boolean finish = false; - public void onMessage(Message m) + public void opened(Session ssn) {} + + public void resumed(Session ssn) {} + + public void message(Session ssn, MessageTransfer xfr) { - 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; - } + System.out.println("Message: " + xfr); } - public boolean isFinished() + public void exception(Session ssn, SessionException exc) { - return finish; + exc.printStackTrace(); } + public void closed(Session ssn) {} + /** * This sends 10 messages to the * amq.direct exchange using the * routing key as "routing_key" * */ - public static void main(String[] args) + public static void main(String[] args) throws InterruptedException { // 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(); - } + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); // Create session Session session = con.createSession(0); // Create an instance of the listener Listener listener = new Listener(); + session.setSessionListener(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); + MessageAcceptMode.NONE, + MessageAcquireMode.PRE_ACQUIRED, + null, 0, null); // issue credits // XXX - session.messageFlow("listener_destination", MessageCreditUnit.BYTE, Session.MESSAGE_FLOW_MAX_BYTES); + session.messageFlow("listener_destination", MessageCreditUnit.BYTE, Session.UNLIMITED_CREDIT); session.messageFlow("listener_destination", MessageCreditUnit.MESSAGE, 11); // confirm completion session.sync(); - // check to see if we have received all the messages - while (!listener.isFinished()){} + // wait to receive all the messages + System.out.println("Waiting 100 seconds for messages from listener_destination"); + Thread.sleep(100*1000); System.out.println("Shutting down listener for listener_destination"); session.messageCancel("listener_destination"); //cleanup - session.sessionDetach(session.getName()); - - try - { - con.close(); - } - catch(Exception e) - { - System.out.print("Error closing broker connection"); - e.printStackTrace(); - } + session.close(); + con.close(); } } 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 index 9af5f21e66..9c3ec2fb3b 100755 --- 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 @@ -1,8 +1,28 @@ package org.apache.qpid.example.amqpexample.fanout; +/* + * + * 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. + * + */ + -import org.apache.qpid.nclient.Client; -import org.apache.qpid.nclient.Connection; -import org.apache.qpid.nclient.Session; +import org.apache.qpid.transport.Connection; +import org.apache.qpid.transport.Session; /** * This creates a queue a queue and binds it to the @@ -15,16 +35,8 @@ 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(); - } + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); // Create session Session session = con.createSession(0); @@ -37,15 +49,7 @@ public class DeclareQueue session.sync(); //cleanup - session.sessionDetach(session.getName()); - try - { - con.close(); - } - catch(Exception e) - { - System.out.print("Error closing broker connection"); - e.printStackTrace(); - } + session.close(); + con.close(); } } 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 index 41038d3e53..39d34713c6 100755 --- 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 @@ -1,12 +1,32 @@ package org.apache.qpid.example.amqpexample.fanout; +/* + * + * 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. + * + */ -import org.apache.qpid.nclient.Client; -import org.apache.qpid.nclient.Connection; -import org.apache.qpid.nclient.Session; + +import org.apache.qpid.transport.Connection; import org.apache.qpid.transport.DeliveryProperties; import org.apache.qpid.transport.Header; import org.apache.qpid.transport.MessageAcceptMode; import org.apache.qpid.transport.MessageAcquireMode; +import org.apache.qpid.transport.Session; public class FannoutProducer { @@ -17,16 +37,8 @@ public class FannoutProducer 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(); - } + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); // Create session Session session = con.createSession(0); @@ -47,16 +59,8 @@ public class FannoutProducer session.sync(); //cleanup - session.sessionDetach(session.getName()); - try - { - con.close(); - } - catch(Exception e) - { - System.out.print("Error closing broker connection"); - e.printStackTrace(); - } + session.close(); + con.close(); } } 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 index 7942ce5ca3..21f9c43cd2 100755 --- 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 @@ -1,115 +1,105 @@ package org.apache.qpid.example.amqpexample.fanout; +/* + * + * 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. + * + */ -import java.nio.ByteBuffer; -import org.apache.qpid.api.Message; -import org.apache.qpid.nclient.Client; -import org.apache.qpid.nclient.Connection; -import org.apache.qpid.nclient.Session; -import org.apache.qpid.nclient.util.MessageListener; -import org.apache.qpid.nclient.util.MessagePartListenerAdapter; +import java.nio.ByteBuffer; +import org.apache.qpid.transport.Connection; import org.apache.qpid.transport.MessageAcceptMode; import org.apache.qpid.transport.MessageAcquireMode; import org.apache.qpid.transport.MessageCreditUnit; +import org.apache.qpid.transport.MessageTransfer; +import org.apache.qpid.transport.Session; +import org.apache.qpid.transport.SessionException; +import org.apache.qpid.transport.SessionListener; /** * This listens to messages on a queue and terminates * when it sees the final message * */ -public class Listener implements MessageListener +public class Listener implements SessionListener { - boolean finish = false; - public void onMessage(Message m) + public void opened(Session ssn) {} + + public void resumed(Session ssn) {} + + public void message(Session ssn, MessageTransfer xfr) { - 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; - } + System.out.println("Message: " + xfr); } - public boolean isFinished() + public void exception(Session ssn, SessionException exc) { - return finish; + exc.printStackTrace(); } + public void closed(Session ssn) {} + /** * This sends 10 messages to the * amq.direct exchange using the * routing key as "routing_key" * */ - public static void main(String[] args) + public static void main(String[] args) throws InterruptedException { // 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(); - } + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); // Create session Session session = con.createSession(0); // Create an instance of the listener Listener listener = new Listener(); + session.setSessionListener(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); + MessageAcceptMode.NONE, + MessageAcquireMode.PRE_ACQUIRED, + null, 0, null); // issue credits // XXX - session.messageFlow("listener_destination", MessageCreditUnit.BYTE, Session.MESSAGE_FLOW_MAX_BYTES); + session.messageFlow("listener_destination", MessageCreditUnit.BYTE, Session.UNLIMITED_CREDIT); session.messageFlow("listener_destination", MessageCreditUnit.MESSAGE, 11); // confirm completion session.sync(); // check to see if we have received all the messages - while (!listener.isFinished()){} + System.out.println("Waiting 100 seconds for messages from listener_destination"); + Thread.sleep(100*1000); System.out.println("Shutting down listener for listener_destination"); session.messageCancel("listener_destination"); //cleanup - session.sessionDetach(session.getName()); - try - { - con.close(); - } - catch(Exception e) - { - System.out.print("Error closing broker connection"); - e.printStackTrace(); - } + session.close(); + con.close(); } } diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/DeclareQueue.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/DeclareQueue.java new file mode 100644 index 0000000000..6fd73a59c4 --- /dev/null +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/DeclareQueue.java @@ -0,0 +1,74 @@ +package org.apache.qpid.example.amqpexample.headers; +/* + * + * 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. + * + */ + + +import org.apache.qpid.transport.Connection; +import org.apache.qpid.transport.Session; + +import java.util.Map; +import java.util.HashMap; + + +public class DeclareQueue +{ + + /** + * Creates 2 queues and bind them to an headers exchange. One queue receives messages with both + * properties H1 and H2 and the other queue receives messages with either one of those properties. + */ + public static void main(String[] args) + { + // Create connection + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); + + // Create session + Session session = con.createSession(0); + + // declare and bind queues + session.queueDeclare("headers_queue_any", null, null); + session.queueDeclare("headers_queue_all", null, null); + // we need to declare the header: name, type, alternate exchange + session.exchangeDeclare("test.headers", "headers", "amq.direct", null); + // The matching algorithm is controlled by 'x-match' property + // 'x-match' can take one of two values, + // (i) 'all' implies that all the other pairs must match the headers + // property of a message for that message to be routed (i.e. an AND match) + // (ii) 'any' implies that the message should be routed if any of the + // fields in the headers property match one of the fields in the arguments table (i.e. an OR match) + Map arguments = new HashMap(); + arguments.put("x-match", "any"); + arguments.put("h1", "v1"); + arguments.put("h2", "v2"); + session.exchangeBind("headers_queue_any", "test.headers", "useless", arguments); + arguments = new HashMap(); + arguments.put("x-match", "all"); + arguments.put("h1", "v1"); + arguments.put("h2", "v2"); + session.exchangeBind("headers_queue_all", "test.headers", "useless", arguments); + // confirm completion + session.sync(); + //cleanup + session.close(); + con.close(); + } +} diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Listener.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Listener.java new file mode 100644 index 0000000000..dff49228a1 --- /dev/null +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Listener.java @@ -0,0 +1,107 @@ +/* + * + * 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. + * + */ +package org.apache.qpid.example.amqpexample.headers; + + +import org.apache.qpid.transport.*; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +public class Listener implements SessionListener +{ + private static CountDownLatch _countDownLatch = new CountDownLatch(1); + + public void opened(Session ssn) {} + + public void resumed(Session ssn) {} + + public void message(Session ssn, MessageTransfer xfr) + { + String body = xfr.getBodyString(); + System.out.println("Message: " + body); + if ( body.equals("That's all, folks!")) + { + System.out.println("Received final message"); + _countDownLatch.countDown(); + } + } + + public void exception(Session ssn, SessionException exc) + { + exc.printStackTrace(); + } + + public void closed(Session ssn) {} + + /** + * Receives messages from queue ANY and then ALL + */ + public static void main(String[] args) throws InterruptedException + { + // Create connection + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); + + // Create session + Session session = con.createSession(0); + // we expect to receive all the messages + Consume(session, "headers_queue_any"); + // we expect to receive only messages that have both properties set. + Consume(session, "headers_queue_all"); + + //cleanup + session.close(); + con.close(); + } + + private static void Consume(Session session, String queueName) throws InterruptedException + { + System.out.println("Consuming messages for queue " + queueName); + _countDownLatch = new CountDownLatch(1); + // Create an instance of the listener + Listener listener = new Listener(); + session.setSessionListener(listener); + + // create a subscription + session.messageSubscribe(queueName, + "listener_destination", + MessageAcceptMode.NONE, + MessageAcquireMode.PRE_ACQUIRED, + null, 0, null); + + + // issue credits + session.messageFlow("listener_destination", MessageCreditUnit.BYTE, Session.UNLIMITED_CREDIT); + session.messageFlow("listener_destination", MessageCreditUnit.MESSAGE, 100); + // confirm completion + session.sync(); + + // wait to receive all the messages + System.out.println("Waiting 100 seconds for messages from queue " + queueName); + + _countDownLatch.await(30, TimeUnit.SECONDS); + System.out.println("Shutting down listener for " + queueName); + System.out.println("========================================="); + session.messageCancel("listener_destination"); + } + +} diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Producer.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Producer.java new file mode 100644 index 0000000000..a54069889a --- /dev/null +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Producer.java @@ -0,0 +1,88 @@ +/* + * + * 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. + * + */ +package org.apache.qpid.example.amqpexample.headers; + +import org.apache.qpid.transport.*; +import java.util.Map; +import java.util.HashMap; + + +public class Producer +{ + /** + * Sends 10 messages with a single property and 10 messages + * with 2 properties to a headers exchange. + */ + public static void main(String[] args) + { + // Create connection + org.apache.qpid.transport.Connection con = new org.apache.qpid.transport.Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); + + // Create session + org.apache.qpid.transport.Session session = con.createSession(0); + DeliveryProperties deliveryProps = new DeliveryProperties(); + + // set message headers + MessageProperties messageProperties = new MessageProperties(); + Map messageHeaders = new HashMap(); + // set the message property + messageHeaders.put("h1", "v1"); + messageProperties.setApplicationHeaders(messageHeaders); + Header header = new Header(deliveryProps, messageProperties); + + for (int i=0; i<10; i++) + { + session.messageTransfer("test.headers", MessageAcceptMode.EXPLICIT,MessageAcquireMode.PRE_ACQUIRED, + header, + "Message H1: " + i); + } + + // set message headers + messageProperties = new MessageProperties(); + messageHeaders = new HashMap(); + // set the message properties + messageHeaders.put("h1", "v1"); + messageHeaders.put("h2", "v2"); + messageProperties.setApplicationHeaders(messageHeaders); + header = new Header(deliveryProps, messageProperties); + + for (int i=0; i<10; i++) + { + session.messageTransfer("test.headers", MessageAcceptMode.EXPLICIT,MessageAcquireMode.PRE_ACQUIRED, + header, + "Message H1 and H2: " + i); + } + + + session.messageTransfer("test.headers", MessageAcceptMode.EXPLICIT,MessageAcquireMode.PRE_ACQUIRED, + header, + "That's all, folks!" ); + + // confirm completion + session.sync(); + + //cleanup + session.close(); + con.close(); + } + +} diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/DeclareLVQueue.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/DeclareLVQueue.java new file mode 100644 index 0000000000..86a0f362ad --- /dev/null +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/DeclareLVQueue.java @@ -0,0 +1,64 @@ +package org.apache.qpid.example.amqpexample.lvq; +/* + * + * 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. + * + */ + + +import org.apache.qpid.transport.Connection; +import org.apache.qpid.transport.Session; + +import java.util.Map; +import java.util.HashMap; + +/** + * This creates a queue a LVQueue with key test and binds it to the + * amq.direct exchange + * + */ +public class DeclareLVQueue +{ + + public static void main(String[] args) + { + // Create connection + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); + + // Create session + Session session = con.createSession(0); + + // declare and bind queue + Map arguments = new HashMap(); + // We use a lvq + arguments.put("qpid.last_value_queue", true); + // We want this queue to use the key test + arguments.put("qpid.LVQ_key", "test"); + session.queueDeclare("message_queue", null, arguments); + session.exchangeBind("message_queue", "amq.direct", "routing_key", null); + + // confirm completion + session.sync(); + + //cleanup + session.close(); + con.close(); + } +} + diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Listener.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Listener.java new file mode 100644 index 0000000000..e17d3eef9f --- /dev/null +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Listener.java @@ -0,0 +1,115 @@ +package org.apache.qpid.example.amqpexample.lvq; +/* + * + * 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. + * + */ + + +import org.apache.qpid.transport.Connection; +import org.apache.qpid.transport.MessageAcceptMode; +import org.apache.qpid.transport.MessageAcquireMode; +import org.apache.qpid.transport.MessageCreditUnit; +import org.apache.qpid.transport.MessageTransfer; +import org.apache.qpid.transport.Session; +import org.apache.qpid.transport.SessionException; +import org.apache.qpid.transport.SessionListener; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +/** + * This listens to messages on a queue and terminates + * when it sees the final message + * + */ +public class Listener implements SessionListener +{ + private static CountDownLatch _countDownLatch = new CountDownLatch(1); + + public void opened(Session ssn) {} + + public void resumed(Session ssn) {} + + public void message(Session ssn, MessageTransfer xfr) + { + String body = xfr.getBodyString(); + System.out.println("Message: " + body); + if ( body.equals("That's all, folks!")) + { + System.out.println("Received final message"); + _countDownLatch.countDown(); + } + } + + public void exception(Session ssn, SessionException exc) + { + exc.printStackTrace(); + } + + public void closed(Session ssn) {} + + /** + * This sends 10 messages to the + * amq.direct exchange using the + * routing key as "routing_key" + * + * @param args + */ + public static void main(String[] args) throws InterruptedException + { + // Create connection + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); + + // Create session + Session session = con.createSession(0); + + // Create an instance of the listener + Listener listener = new Listener(); + session.setSessionListener(listener); + + // create a subscription + session.messageSubscribe("message_queue", + "listener_destination", + MessageAcceptMode.NONE, + MessageAcquireMode.PRE_ACQUIRED, + null, 0, null); + + + // issue credits + // XXX + session.messageFlow("listener_destination", MessageCreditUnit.BYTE, Session.UNLIMITED_CREDIT); + session.messageFlow("listener_destination", MessageCreditUnit.MESSAGE, 11); + + // confirm completion + session.sync(); + + // wait to receive all the messages + System.out.println("Waiting 100 seconds for messages from listener_destination"); + + _countDownLatch.await(30, TimeUnit.SECONDS); + System.out.println("Shutting down listener for listener_destination"); + session.messageCancel("listener_destination"); + + //cleanup + session.close(); + con.close(); + } + +} diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Producer.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Producer.java new file mode 100644 index 0000000000..482e6a6b11 --- /dev/null +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Producer.java @@ -0,0 +1,69 @@ +package org.apache.qpid.example.amqpexample.lvq; +/* + * + * 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. + * + */ + +import java.util.Map; +import java.util.HashMap; +import org.apache.qpid.transport.*; + +public class Producer +{ + + public static void main(String[] args) + { + // Create connection + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); + + // Create session + Session session = con.createSession(0); + DeliveryProperties deliveryProps = new DeliveryProperties(); + deliveryProps.setRoutingKey("routing_key"); + + // set message headers + MessageProperties messageProperties = new MessageProperties(); + Map messageHeaders = new HashMap(); + // set the message key + messageHeaders.put("qpid.LVQ_key", "test"); + messageProperties.setApplicationHeaders(messageHeaders); + + Header header = new Header(deliveryProps, messageProperties); + + for (int i=0; i<10; i++) + { + session.messageTransfer("amq.direct", MessageAcceptMode.EXPLICIT,MessageAcquireMode.PRE_ACQUIRED, + header, + "Message " + i); + } + + session.messageTransfer("amq.direct", MessageAcceptMode.EXPLICIT,MessageAcquireMode.PRE_ACQUIRED, + header, + "That's all, folks!"); + + // confirm completion + session.sync(); + + //cleanup + session.close(); + con.close(); + } + +} diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicListener.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicListener.java index 2e4edfbe6f..dd9307ca84 100755 --- a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicListener.java +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicListener.java @@ -1,64 +1,73 @@ package org.apache.qpid.example.amqpexample.pubsub; +/* + * + * 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. + * + */ + import java.nio.ByteBuffer; -import org.apache.qpid.api.Message; -import org.apache.qpid.nclient.Client; -import org.apache.qpid.nclient.Connection; -import org.apache.qpid.nclient.Session; -import org.apache.qpid.nclient.util.MessageListener; -import org.apache.qpid.nclient.util.MessagePartListenerAdapter; +import org.apache.qpid.transport.Connection; +import org.apache.qpid.transport.DeliveryProperties; +import org.apache.qpid.transport.MessageAcceptMode; +import org.apache.qpid.transport.MessageAcquireMode; import org.apache.qpid.transport.MessageCreditUnit; +import org.apache.qpid.transport.MessageTransfer; import org.apache.qpid.transport.Option; +import org.apache.qpid.transport.Session; +import org.apache.qpid.transport.SessionException; +import org.apache.qpid.transport.SessionListener; -public class TopicListener implements MessageListener +public class TopicListener implements SessionListener { - boolean finish = false; - int count = 0; - public void onMessage(Message m) + public void opened(Session ssn) {} + + public void resumed(Session ssn) {} + + public void message(Session ssn, MessageTransfer xfr) + { + DeliveryProperties dp = xfr.getHeader().get(DeliveryProperties.class); + System.out.println("Message: " + xfr + " with routing_key " + dp.getRoutingKey()); + } + + public void exception(Session ssn, SessionException exc) { - 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 + " with routing_key " + m.getDeliveryProperties().getRoutingKey()); - - if (data != null && data.equals("That's all, folks!")) - { - count++; - if (count == 4){ - finish = true; - } - } + exc.printStackTrace(); } + public void closed(Session ssn) {} + public void prepareQueue(Session session,String queueName,String bindingKey) { session.queueDeclare(queueName, null, null, Option.EXCLUSIVE, Option.AUTO_DELETE); session.exchangeBind(queueName, "amq.topic", bindingKey, null); session.exchangeBind(queueName, "amq.topic", "control", null); - session.messageSubscribe(queueName,queueName, - Session.TRANSFER_CONFIRM_MODE_NOT_REQUIRED, - Session.TRANSFER_ACQUIRE_MODE_PRE_ACQUIRE, - new MessagePartListenerAdapter(this), - null, Option.NONE); + session.messageSubscribe(queueName, queueName, + MessageAcceptMode.NONE, + MessageAcquireMode.PRE_ACQUIRED, + null, 0, null); // issue credits // XXX: need to be able to set to null - session.messageFlow(queueName, MessageCreditUnit.BYTE, Session.MESSAGE_FLOW_MAX_BYTES); + session.messageFlow(queueName, MessageCreditUnit.BYTE, Session.UNLIMITED_CREDIT); session.messageFlow(queueName, MessageCreditUnit.MESSAGE, 24); } @@ -67,30 +76,18 @@ public class TopicListener implements MessageListener session.messageCancel(dest); } - public boolean isFinished() - { - return finish; - } - - public static void main(String[] args) + public static void main(String[] args) throws InterruptedException { // 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(); - } + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); // Create session Session session = con.createSession(0); // Create an instance of the listener TopicListener listener = new TopicListener(); + session.setSessionListener(listener); listener.prepareQueue(session,"usa", "usa.#"); listener.prepareQueue(session,"europe", "europe.#"); @@ -99,25 +96,19 @@ public class TopicListener implements MessageListener // 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"); + + System.out.println("Waiting 100 seconds for messages"); + Thread.sleep(100*1000); + + System.out.println("Shutting down listeners"); listener.cancelSubscription(session,"usa"); listener.cancelSubscription(session,"europe"); listener.cancelSubscription(session,"news"); listener.cancelSubscription(session,"weather"); //cleanup - session.sessionDetach(session.getName()); - try - { - con.close(); - } - catch(Exception e) - { - System.out.print("Error closing broker connection"); - e.printStackTrace(); - } + session.close(); + con.close(); } } diff --git a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicPublisher.java b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicPublisher.java index caec886642..facf08eeca 100755 --- a/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicPublisher.java +++ b/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicPublisher.java @@ -1,15 +1,36 @@ package org.apache.qpid.example.amqpexample.pubsub; +/* + * + * 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. + * + */ -import org.apache.qpid.nclient.Client; -import org.apache.qpid.nclient.Connection; -import org.apache.qpid.nclient.Session; + +import org.apache.qpid.transport.Connection; import org.apache.qpid.transport.DeliveryProperties; import org.apache.qpid.transport.Header; import org.apache.qpid.transport.MessageAcceptMode; import org.apache.qpid.transport.MessageAcquireMode; +import org.apache.qpid.transport.Session; public class TopicPublisher { + public void publishMessages(Session session, String routing_key) { // Set the routing key once, we'll use the same routing key for all @@ -35,16 +56,8 @@ public class TopicPublisher 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(); - } + Connection con = new Connection(); + con.connect("localhost", 5672, "test", "guest", "guest",false); // Create session Session session = con.createSession(0); @@ -61,15 +74,7 @@ public class TopicPublisher session.sync(); //cleanup - session.sessionDetach(session.getName()); - try - { - con.close(); - } - catch(Exception e) - { - System.out.print("Error closing broker connection"); - e.printStackTrace(); - } + session.close(); + con.close(); } } diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify index 0541e490b6..7f81a3a57b 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify @@ -1,14 +1,33 @@ +# +# 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. +# + # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify # The JMS producer doesn't create qeueues so utilising the c++ declare_queues cpp=$CPP/direct direct_consumer_java() { -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Consumer +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Consumer } direct_producer_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Producer +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Producer } clients $cpp/declare_queues direct_producer_java direct_consumer_java diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_cpp_java b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_cpp_java index d581c4c1aa..a22162e075 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_cpp_java +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_cpp_java @@ -2,7 +2,7 @@ cpp=$CPP/direct direct_consumer_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Consumer +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Consumer } clients $cpp/declare_queues $cpp/direct_producer direct_consumer_java diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_java_cpp b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_java_cpp index 573cac6986..dc4b349808 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_java_cpp +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_java_cpp @@ -2,7 +2,7 @@ cpp=$CPP/direct direct_producer_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Producer +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Producer } clients $cpp/declare_queues direct_producer_java $cpp/listener diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_java_python b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_java_python index 61c033e969..befa34d650 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_java_python +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_java_python @@ -1,8 +1,8 @@ # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify -py=$PYTHON_EXAMPLES/direct +py=$PYTHON/direct direct_producer_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Producer +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Producer } clients $py/declare_queues.py direct_producer_java $py/direct_consumer.py diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_python_java b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_python_java index 4182331f3f..b22b44b9a6 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_python_java +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/verify_python_java @@ -1,8 +1,8 @@ # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify -py=$PYTHON_EXAMPLES/direct +py=$PYTHON/direct direct_consumer_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Consumer +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.direct.Consumer } clients $py/declare_queues.py $py/direct_producer.py direct_consumer_java diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify index 2edc1ced02..98c866da99 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify @@ -1,13 +1,32 @@ +# +# 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. +# + # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify # The JMS producer doesn't create qeueues so utilising the c++ declare_queues cpp=$CPP/fanout fanout_listener_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Listener $1 +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Listener $1 } fanout_producer_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Producer +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Producer } background "can receive messages" fanout_listener_java fanoutQueue1 diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_cpp_java b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_cpp_java index de057ea3b1..ab8d37a0d8 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_cpp_java +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_cpp_java @@ -3,7 +3,7 @@ cpp=$CPP/fanout fanout_listener_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Listener $1 +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Listener $1 } background "can receive messages" fanout_listener_java fanoutQueue1 diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_cpp b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_cpp index dab6114572..df923e6354 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_cpp +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_cpp @@ -3,7 +3,7 @@ cpp=$CPP/fanout fanout_producer_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Producer +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Producer } background "Listening" $cpp/listener diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_python b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_python index 1641d88354..5f8701882d 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_python +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_python @@ -1,9 +1,9 @@ # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify # The JMS producer doesn't create qeueues so utilising the c++ declare_queues -py=$PYTHON_EXAMPLES/fanout +py=$PYTHON/fanout fanout_producer_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Producer +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Producer } background "Subscribed" $py/fanout_consumer.py diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_python_java b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_python_java index 0f05663985..72f263fd3d 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_python_java +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_python_java @@ -1,9 +1,9 @@ # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify # The JMS producer doesn't create qeueues so utilising the c++ declare_queues -py=$PYTHON_EXAMPLES/fanout +py=$PYTHON/fanout fanout_listener_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Listener $1 +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.fanout.Listener $1 } background "can receive messages" fanout_listener_java fanoutQueue1 diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify index cceb565dfc..363af252ad 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify @@ -1,12 +1,31 @@ +# +# 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. +# + # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify cpp=$CPP/pub-sub topic_listener_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Listener +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Listener } topic_publisher_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Publisher +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Publisher } background "can receive messages" topic_listener_java diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_cpp_java b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_cpp_java index 9276b3e21b..e73c164b77 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_cpp_java +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_cpp_java @@ -2,7 +2,7 @@ cpp=$CPP/pub-sub topic_listener_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Listener +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Listener } background "can receive messages" topic_listener_java diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_cpp b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_cpp index af22b3b82c..0b877566d3 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_cpp +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_cpp @@ -2,7 +2,7 @@ cpp=$CPP/pub-sub topic_publisher_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Publisher +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Publisher } background "Listening" $cpp/topic_listener diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_python b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_python index 3758e0f014..1340fe79eb 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_python +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_python @@ -1,8 +1,8 @@ # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify -py=$PYTHON_EXAMPLES/pubsub +py=$PYTHON/pubsub topic_publisher_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Publisher +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Publisher } background "Queues created" $py/topic_subscriber.py diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_python_java b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_python_java index c2b516f376..b7fba2b3e0 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_python_java +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_python_java @@ -1,8 +1,8 @@ # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify -py=$PYTHON_EXAMPLES/pubsub +py=$PYTHON/pubsub topic_listener_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Listener +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.pubsub.Listener } background "can receive messages" topic_listener_java diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify index daf84c8ff9..c6caa7239e 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify @@ -1,12 +1,31 @@ +# +# 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. +# + # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify cpp=$CPP/pub-sub client_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Client +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Client } server_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Server +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Server } background "can receive messages" server_java diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_cpp_java b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_cpp_java index 6ef1b3b7e3..c0e788e373 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_cpp_java +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_cpp_java @@ -3,7 +3,7 @@ cpp=$CPP/request-response client_java() { -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Client +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Client } background "Waiting" $cpp/server diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_cpp b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_cpp index 2513eb691b..14a8c28000 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_cpp +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_cpp @@ -2,11 +2,11 @@ cpp=$CPP/request-response server_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Server +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Server } background "can receive messages" server_java clients $cpp/client -#ps -ao pid,cmd | awk '/qpid-client-incubating-M3.jar/{ print $1 }' | xargs -r kill +#ps -ao pid,cmd | awk '/qpid-client-.jar/{ print $1 }' | xargs -r kill kill %% outputs "$cpp/client.out | remove_uuid" "server_java.out | remove_uuid" diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_python b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_python index 0760952527..2d2ec2fc04 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_python +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_python @@ -1,8 +1,8 @@ # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify -py=$PYTHON_EXAMPLES/request-response +py=$PYTHON/request-response server_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Server +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Server } background "can receive messages" server_java diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_python_java b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_python_java index 6ea526e914..bcedf168e3 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_python_java +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_python_java @@ -1,8 +1,8 @@ # See https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/bin/verify -py=$PYTHON_EXAMPLES/request-response +py=$PYTHON/request-response client_java(){ -java -Dlog4j.configuration=file://"$JAVA"/log4j.xml -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Client +java -Dlog4j.configuration=$LOG4J -cp "$CLASSPATH" org.apache.qpid.example.jmsexample.requestResponse.Client } background "Request server running" $py/server.py diff --git a/java/client/example/src/main/java/org/apache/qpid/example/publisher/FileMessageFactory.java b/java/client/example/src/main/java/org/apache/qpid/example/publisher/FileMessageFactory.java index 1240284a56..04339b2498 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/publisher/FileMessageFactory.java +++ b/java/client/example/src/main/java/org/apache/qpid/example/publisher/FileMessageFactory.java @@ -19,7 +19,7 @@ package org.apache.qpid.example.publisher; -import org.apache.qpid.example.shared.FileUtils; +import org.apache.qpid.util.FileUtils; import org.apache.qpid.example.shared.Statics; import java.io.*; @@ -42,10 +42,10 @@ public class FileMessageFactory try { _filename = filename; - _payload = FileUtils.getFileContent(filename); + _payload = FileUtils.readFileAsString(filename); _session = session; } - catch (IOException e) + catch (Exception e) { MessageFactoryException mfe = new MessageFactoryException(e.toString(), e); throw mfe; diff --git a/java/client/example/src/main/java/org/apache/qpid/example/publisher/MultiMessageDispatcher.java b/java/client/example/src/main/java/org/apache/qpid/example/publisher/MultiMessageDispatcher.java new file mode 100644 index 0000000000..a92efe99ac --- /dev/null +++ b/java/client/example/src/main/java/org/apache/qpid/example/publisher/MultiMessageDispatcher.java @@ -0,0 +1,141 @@ +/* + * + * 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. + * + */ +package org.apache.qpid.example.publisher; + +import java.io.File; + +import javax.jms.JMSException; +import javax.jms.TextMessage; + + +import org.apache.qpid.example.shared.FileUtils; +import org.apache.qpid.example.shared.Statics; +import org.slf4j.LoggerFactory; +import org.slf4j.Logger; + +/** + * Class that sends parameterised number of message files to the Publisher + * Must set properties for host in properties file or uses in vm broker + */ +public class MultiMessageDispatcher +{ + + protected static final Logger _logger = LoggerFactory.getLogger(FileMessageDispatcher.class); + + protected static Publisher _publisher = null; + + /** + * To use this main method you need to specify a path or file to use for input + * This class then uses file contents from the dir/file specified to generate + * messages to publish + * Intended to be a very simple way to get going with publishing using the broker + * @param args - must specify one value, the path to file(s) for publisher + */ + public static void main(String[] args) + { + + // Check command line args ok - must provide a path or file for us to dispatch + if (args.length < 2) + { + System.out.println("Usage: MultiMessageDispatcher " + ""); + } + else + { + boolean topicPublisher = true; + + try + { + // publish message(s) + topicPublisher = new Boolean(args[1]).booleanValue(); + publish(new Integer(args[0]).intValue(),topicPublisher); + + // Move payload file(s) to archive location as no error + FileUtils.moveFileToNewDir(args[0], System.getProperties().getProperty(Statics.ARCHIVE_PATH)); + } + catch (Exception e) + { + // log error and exit + _logger.error("Error trying to dispatch message: " + e); + System.exit(1); + } + finally + { + + cleanup(topicPublisher); + } + } + + if (_logger.isDebugEnabled()) + { + _logger.debug("Finished dispatching message"); + } + + System.exit(0); + } + + /** + * Publish the content of a file or files from a directory as messages + * @param numMessages - from main args + * @throws javax.jms.JMSException + * @throws org.apache.qpid.example.publisher.MessageFactoryException - if cannot create message from file content + */ + public static void publish(int numMessages, boolean topicPublisher) throws JMSException, MessageFactoryException + { + { + // Send the message generated from the payload using the _publisher + getPublisher(topicPublisher).sendMessage(numMessages); + } + } + + /** + * Cleanup before exit + */ + public static void cleanup(boolean topicPublisher) + { + if (getPublisher(topicPublisher) != null) + { + getPublisher(topicPublisher).cleanup(); + } + } + + /** + * @return A Publisher instance + */ + private static Publisher getPublisher(boolean topic) + { + if (_publisher != null) + { + return _publisher; + } + + if (!topic) + { + // Create a _publisher + _publisher = new Publisher(); + } + else + { + _publisher = new TopicPublisher(); + } + return _publisher; + } + +} \ No newline at end of file diff --git a/java/client/example/src/main/java/org/apache/qpid/example/publisher/Publisher.java b/java/client/example/src/main/java/org/apache/qpid/example/publisher/Publisher.java index 87fc543dbe..b5f44557a4 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/publisher/Publisher.java +++ b/java/client/example/src/main/java/org/apache/qpid/example/publisher/Publisher.java @@ -20,13 +20,7 @@ package org.apache.qpid.example.publisher; import org.apache.qpid.client.AMQConnectionFactory; -import javax.jms.JMSException; -import javax.jms.Message; -import javax.jms.DeliveryMode; -import javax.jms.Queue; -import javax.jms.MessageProducer; -import javax.jms.Connection; -import javax.jms.Session; +import javax.jms.*; import javax.naming.InitialContext; @@ -50,7 +44,7 @@ public class Publisher protected String _name = "Publisher"; - protected Queue _destination; + protected Destination _destination; protected static final String _defaultDestinationDir = "/tmp"; @@ -70,6 +64,17 @@ public class Publisher AMQConnectionFactory cf = (AMQConnectionFactory) ctx.lookup("local"); _connection = cf.createConnection(); + _connection.setExceptionListener(new ExceptionListener() + { + public void onException(JMSException jmse) + { + // The connection may have broken invoke reconnect code if available. + // The connection may have broken invoke reconnect code if available. + System.err.println("ExceptionListener caught: " + jmse); + //System.exit(0); + } + }); + //create a transactional session _session = _connection.createSession(true, Session.AUTO_ACKNOWLEDGE); @@ -92,6 +97,28 @@ public class Publisher } } + /** + * Creates and sends the number of messages specified in the param + */ + public void sendMessage(int numMessages) + { + try + { + TextMessage txtMessage = _session.createTextMessage("msg"); + for (int i=0;i