diff options
Diffstat (limited to 'java/amqp-1-0-client/example/src/main/java/org/apache/qpid/amqp_1_0/client/Request.java')
| -rw-r--r-- | java/amqp-1-0-client/example/src/main/java/org/apache/qpid/amqp_1_0/client/Request.java | 249 |
1 files changed, 249 insertions, 0 deletions
diff --git a/java/amqp-1-0-client/example/src/main/java/org/apache/qpid/amqp_1_0/client/Request.java b/java/amqp-1-0-client/example/src/main/java/org/apache/qpid/amqp_1_0/client/Request.java new file mode 100644 index 0000000000..6e1d15376c --- /dev/null +++ b/java/amqp-1-0-client/example/src/main/java/org/apache/qpid/amqp_1_0/client/Request.java @@ -0,0 +1,249 @@ +/*
+ *
+ * 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.amqp_1_0.client;
+
+import org.apache.qpid.amqp_1_0.type.AmqpErrorException;
+import org.apache.qpid.amqp_1_0.type.Section;
+import org.apache.qpid.amqp_1_0.type.UnsignedInteger;
+import org.apache.qpid.amqp_1_0.type.UnsignedLong;
+import org.apache.qpid.amqp_1_0.type.messaging.AmqpValue;
+import org.apache.qpid.amqp_1_0.type.messaging.Header;
+import org.apache.qpid.amqp_1_0.type.messaging.Properties;
+import org.apache.commons.cli.*;
+
+import java.util.Arrays;
+
+public class Request extends Util
+{
+ private static final String USAGE_STRING = "request [options] <address> [<content> ...]\n\nOptions:";
+
+ public static void main(String[] args)
+ {
+ new Request(args).run();
+ }
+
+ public Request(String[] args)
+ {
+ super(args);
+ }
+
+ @Override
+ protected boolean hasLinkDurableOption()
+ {
+ return false;
+ }
+
+ @Override
+ protected boolean hasLinkNameOption()
+ {
+ return false;
+ }
+
+ @Override
+ protected boolean hasResponseQueueOption()
+ {
+ return false;
+ }
+
+ @Override
+ protected boolean hasSizeOption()
+ {
+ return true;
+ }
+
+ @Override
+ protected boolean hasBlockOption()
+ {
+ return false;
+ }
+
+ @Override
+ protected boolean hasStdInOption()
+ {
+ return false;
+ }
+
+ @Override
+ protected boolean hasTxnOption()
+ {
+ return true;
+ }
+
+ @Override
+ protected boolean hasModeOption()
+ {
+ return true;
+ }
+
+ @Override
+ protected boolean hasCountOption()
+ {
+ return true;
+ }
+
+ @Override
+ protected boolean hasWindowSizeOption()
+ {
+ return true;
+ }
+
+ public void run()
+ {
+
+ try
+ {
+
+
+ final String queue = getArgs()[0];
+
+ String message = "";
+
+ Connection conn = newConnection();
+ Session session = conn.createSession();
+
+ Connection conn2;
+ Session session2;
+ Receiver responseReceiver;
+
+ if(isUseMultipleConnections())
+ {
+ conn2 = newConnection();
+ session2 = conn2.createSession();
+ responseReceiver = session2.createTemporaryQueueReceiver();
+ }
+ else
+ {
+ conn2 = null;
+ session2 = null;
+ responseReceiver = session.createTemporaryQueueReceiver();
+ }
+
+
+
+
+ responseReceiver.setCredit(UnsignedInteger.valueOf(getWindowSize()), true);
+
+
+
+ Sender s = session.createSender(queue, getWindowSize(), getMode());
+
+ Transaction txn = null;
+
+ if(useTran())
+ {
+ txn = session.createSessionLocalTransaction();
+ }
+
+ int received = 0;
+
+ if(getArgs().length >= 2)
+ {
+ message = getArgs()[1];
+ if(message.length() < getMessageSize())
+ {
+ StringBuilder builder = new StringBuilder(getMessageSize());
+ builder.append(message);
+ for(int x = message.length(); x < getMessageSize(); x++)
+ {
+ builder.append('.');
+ }
+ message = builder.toString();
+ }
+
+ for(int i = 0; i < getCount(); i++)
+ {
+ Properties properties = new Properties();
+ properties.setMessageId(UnsignedLong.valueOf(i));
+ properties.setReplyTo(responseReceiver.getAddress());
+
+ AmqpValue amqpValue = new AmqpValue(message);
+ Section[] sections = { new Header() , properties, amqpValue};
+ final Message message1 = new Message(Arrays.asList(sections));
+
+ s.send(message1, txn);
+
+ Message responseMessage = responseReceiver.receive(false);
+ if(responseMessage != null)
+ {
+ responseReceiver.acknowledge(responseMessage.getDeliveryTag(),txn);
+ received++;
+ }
+ }
+ }
+
+ if(txn != null)
+ {
+ txn.commit();
+ }
+
+
+ while(received < getCount())
+ {
+ Message responseMessage = responseReceiver.receive();
+ responseReceiver.acknowledge(responseMessage.getDeliveryTag());
+ received++;
+ }
+
+
+
+
+ s.close();
+ session.close();
+ conn.close();
+
+ if(session2 != null)
+ {
+ session2.close();
+ conn2.close();
+ }
+ }
+ catch (Connection.ConnectionException e)
+ {
+ e.printStackTrace(); //TODO.
+ }
+ catch (Sender.SenderClosingException e)
+ {
+ e.printStackTrace(); //TODO.
+ }
+ catch (Sender.SenderCreationException e)
+ {
+ e.printStackTrace(); //TODO.
+ }
+ catch (AmqpErrorException e)
+ {
+ e.printStackTrace(); //TODO.
+ }
+
+ }
+
+ protected boolean hasSingleLinkPerConnectionMode()
+ {
+ return true;
+ }
+
+ protected void printUsage(Options options)
+ {
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp(USAGE_STRING, options );
+ }
+
+}
|