QPIDIT-41: Re-organization of project directory structure
Project: http://git-wip-us.apache.org/repos/asf/qpid-interop-test/repo Commit: http://git-wip-us.apache.org/repos/asf/qpid-interop-test/commit/83b89fe4 Tree: http://git-wip-us.apache.org/repos/asf/qpid-interop-test/tree/83b89fe4 Diff: http://git-wip-us.apache.org/repos/asf/qpid-interop-test/diff/83b89fe4 Branch: refs/heads/master Commit: 83b89fe4077e33554de5ed593b5fe1c4009e18ad Parents: baed642 Author: Kim van der Riet <[email protected]> Authored: Tue Oct 4 15:17:05 2016 -0400 Committer: Kim van der Riet <[email protected]> Committed: Tue Oct 4 15:17:05 2016 -0400 ---------------------------------------------------------------------- .../amqp_types_test/Receiver.java | 274 ++++++++++ .../amqp_types_test/Sender.java | 260 +++++++++ .../jms_messages_test/Receiver.java | 525 +++++++++++++++++++ .../jms_messages_test/Sender.java | 508 ++++++++++++++++++ .../qpid_interop_test/shim/AmqpReceiver.java | 271 ---------- .../qpid/qpid_interop_test/shim/AmqpSender.java | 260 --------- .../qpid_interop_test/shim/JmsReceiverShim.java | 525 ------------------- .../qpid_interop_test/shim/JmsSenderShim.java | 508 ------------------ shims/qpid-proton-cpp/src/CMakeLists.txt | 75 ++- .../src/qpidit/amqp_types_test/Receiver.cpp | 290 ++++++++++ .../src/qpidit/amqp_types_test/Receiver.hpp | 85 +++ .../src/qpidit/amqp_types_test/Sender.cpp | 368 +++++++++++++ .../src/qpidit/amqp_types_test/Sender.hpp | 104 ++++ .../qpidit/jms_messages_test/JmsDefinitions.hpp | 46 ++ .../src/qpidit/jms_messages_test/Receiver.cpp | 432 +++++++++++++++ .../src/qpidit/jms_messages_test/Receiver.hpp | 111 ++++ .../src/qpidit/jms_messages_test/Sender.cpp | 487 +++++++++++++++++ .../src/qpidit/jms_messages_test/Sender.hpp | 120 +++++ .../src/qpidit/shim/AmqpReceiver.cpp | 290 ---------- .../src/qpidit/shim/AmqpReceiver.hpp | 85 --- .../src/qpidit/shim/AmqpSender.cpp | 368 ------------- .../src/qpidit/shim/AmqpSender.hpp | 104 ---- .../src/qpidit/shim/JmsDefinitions.hpp | 46 -- .../src/qpidit/shim/JmsReceiver.cpp | 432 --------------- .../src/qpidit/shim/JmsReceiver.hpp | 111 ---- .../src/qpidit/shim/JmsSender.cpp | 487 ----------------- .../src/qpidit/shim/JmsSender.hpp | 120 ----- shims/qpid-proton-python/src/JmsReceiverShim.py | 358 ------------- shims/qpid-proton-python/src/JmsSenderShim.py | 390 -------------- .../qpid-proton-python/src/TypesReceiverShim.py | 127 ----- shims/qpid-proton-python/src/TypesSenderShim.py | 155 ------ .../src/amqp_types_test/Receiver.py | 127 +++++ .../src/amqp_types_test/Sender.py | 155 ++++++ .../src/jms_messages_test/Receiver.py | 358 +++++++++++++ .../src/jms_messages_test/Sender.py | 390 ++++++++++++++ src/python/qpid-interop-test/__init__.py | 1 + src/python/qpid-interop-test/amqp_types_test.py | 435 +++++++++++++++ .../qpid-interop-test/jms/jms_message_tests.py | 511 ------------------ .../qpid-interop-test/jms_messages_test.py | 511 ++++++++++++++++++ .../types/simple_type_tests.py | 435 --------------- 40 files changed, 5638 insertions(+), 5607 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/83b89fe4/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/amqp_types_test/Receiver.java ---------------------------------------------------------------------- diff --git a/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/amqp_types_test/Receiver.java b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/amqp_types_test/Receiver.java new file mode 100644 index 0000000..fea263a --- /dev/null +++ b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/amqp_types_test/Receiver.java @@ -0,0 +1,274 @@ +/** + * 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.interop_test.amqp_types_test; + +import java.math.BigDecimal; +import java.util.UUID; +import java.util.Vector; +import javax.jms.BytesMessage; +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.ExceptionListener; +import javax.jms.JMSException; +import javax.jms.MapMessage; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.ObjectMessage; +import javax.jms.Queue; +import javax.jms.Session; +import javax.jms.StreamMessage; +import javax.jms.TextMessage; +import javax.naming.Context; +import javax.naming.InitialContext; +import org.apache.qpid.jms.JmsConnectionFactory; + +public class Receiver { + private static final String USER = "guest"; + private static final String PASSWORD = "guest"; + private static final int TIMEOUT = 1000; + private static final String[] SUPPORTED_AMQP_TYPES = {"null", + "boolean", + "ubyte", + "ushort", + "uint", + "ulong", + "byte", + "short", + "int", + "long", + "float", + "double", + "decimal32", + "decimal64", + "decimal128", + "char", + "timestamp", + "uuid", + "binary", + "string", + "symbol", + "list", + "map", + "array"}; + + public static void main(String[] args) throws Exception { + if (args.length < 4) { + System.out.println("AmqpReceiver: Insufficient number of arguments"); + System.out.println("AmqpReceiver: Expected arguments: broker_address, queue_name, amqp_type, num_test_values"); + System.exit(1); + } + String brokerAddress = "amqp://" + args[0]; + String queueName = args[1]; + String amqpType = args[2]; + int numTestValues = Integer.parseInt(args[3]); + Connection connection = null; + + try { + ConnectionFactory factory = (ConnectionFactory)new JmsConnectionFactory(brokerAddress); + + connection = factory.createConnection(USER, PASSWORD); + connection.setExceptionListener(new MyExceptionListener()); + connection.start(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + Queue queue = session.createQueue(queueName); + + MessageConsumer messageConsumer = session.createConsumer(queue); + + Vector<String> outList = new Vector<String>(); + outList.add(amqpType); + if (isSupportedAmqpType(amqpType)) { + int actualCount = 0; + Message message = null; + for (int i = 1; i <= numTestValues; i++, actualCount++) { + message = messageConsumer.receive(TIMEOUT); + if (message == null) + break; + switch (amqpType) { + case "null": + long bodyLength = ((BytesMessage)message).getBodyLength(); + if (bodyLength == 0L) { + outList.add("None"); + } else { + throw new Exception("AmqpReceiver: JMS BytesMessage size error: Expected 0 bytes, read " + bodyLength); + } + break; + case "boolean": + String bs = String.valueOf(((BytesMessage)message).readBoolean()); + outList.add(Character.toUpperCase(bs.charAt(0)) + bs.substring(1)); + break; + case "ubyte": + byte byteValue = ((BytesMessage)message).readByte(); + short ubyteValue = (short)(byteValue & 0xff); + outList.add(String.valueOf(ubyteValue)); + break; + case "ushort": + { + byte[] byteArray = new byte[2]; + int numBytes = ((BytesMessage)message).readBytes(byteArray); + if (numBytes != 2) { + // TODO: numBytes == -1 means no more bytes in stream - add error message for this case? + throw new Exception("AmqpReceiver: JMS BytesMessage size error: Exptected 2 bytes, read " + numBytes); + } + int ushortValue = 0; + for (int j=0; j<byteArray.length; j++) { + ushortValue = (ushortValue << 8) + (byteArray[j] & 0xff); + } + outList.add(String.valueOf(ushortValue)); + break; + } + case "uint": + { + byte[] byteArray = new byte[4]; + int numBytes = ((BytesMessage)message).readBytes(byteArray); + if (numBytes != 4) { + // TODO: numBytes == -1 means no more bytes in stream - add error message for this case? + throw new Exception("AmqpReceiver: JMS BytesMessage size error: Exptected 4 bytes, read " + numBytes); + } + long uintValue = 0; + for (int j=0; j<byteArray.length; j++) { + uintValue = (uintValue << 8) + (byteArray[j] & 0xff); + } + outList.add(String.valueOf(uintValue)); + break; + } + case "ulong": + case "timestamp": + { + // TODO: Tidy this ugliness up - perhaps use of vector<byte>? + byte[] byteArray = new byte[8]; + int numBytes = ((BytesMessage)message).readBytes(byteArray); + if (numBytes != 8) { + // TODO: numBytes == -1 means no more bytes in stream - add error message for this case? + throw new Exception("AmqpReceiver: JMS BytesMessage size error: Exptected 8 bytes, read " + numBytes); + } + // TODO: shortcut in use here - this byte array should go through a Java type that can represent this as a number - such as BigInteger. + outList.add(String.format("0x%02x%02x%02x%02x%02x%02x%02x%02x", byteArray[0], byteArray[1], + byteArray[2], byteArray[3], byteArray[4], byteArray[5], byteArray[6], byteArray[7])); + break; + } + case "byte": + outList.add(String.valueOf(((BytesMessage)message).readByte())); + break; + case "short": + outList.add(String.valueOf(((BytesMessage)message).readShort())); + break; + case "int": + outList.add(String.valueOf(((BytesMessage)message).readInt())); + break; + case "long": + outList.add(String.valueOf(((BytesMessage)message).readLong())); + break; + case "float": + float f = ((BytesMessage)message).readFloat(); + int i0 = Float.floatToRawIntBits(f); + outList.add(String.format("0x%8s", Integer.toHexString(i0)).replace(' ', '0')); + break; + case "double": + double d = ((BytesMessage)message).readDouble(); + long l = Double.doubleToRawLongBits(d); + outList.add(String.format("0x%16s", Long.toHexString(l)).replace(' ', '0')); + break; + case "decimal32": + BigDecimal bd32 = (BigDecimal)((ObjectMessage)message).getObject(); + outList.add(bd32.toString()); + break; + case "decimal64": + BigDecimal bd64 = (BigDecimal)((ObjectMessage)message).getObject(); + outList.add(bd64.toString()); + break; + case "decimal128": + BigDecimal bd128 = (BigDecimal)((ObjectMessage)message).getObject(); + outList.add(bd128.toString()); + break; + case "char": + outList.add(String.format("%c", ((BytesMessage)message).readChar())); + break; + case "uuid": + UUID uuid = (UUID)((ObjectMessage)message).getObject(); + outList.add(uuid.toString()); + break; + case "binary": + BytesMessage bm = (BytesMessage)message; + int msgLen = (int)bm.getBodyLength(); + byte[] ba = new byte[msgLen]; + if (bm.readBytes(ba) == msgLen) { + outList.add(new String(ba)); + } else { + // TODO: Raise exception or error here: size mismatch + } + break; + case "string": + outList.add(((TextMessage)message).getText()); + break; + case "symbol": + outList.add(((BytesMessage)message).readUTF()); + break; + case "list": + break; + case "map": + break; + case "array": + break; + default: + // Internal error, should never happen if SUPPORTED_AMQP_TYPES matches this case stmt + connection.close(); + throw new Exception("AmqpReceiver: Internal error: unsupported AMQP type \"" + amqpType + "\""); + } + } + } else { + System.out.println("ERROR: AmqpReceiver: AMQP type \"" + amqpType + "\" is not supported"); + connection.close(); + System.exit(1); + } + + connection.close(); + + // No exception, print results + for (int i=0; i<outList.size(); i++) { + System.out.println(outList.get(i)); + } + } catch (Exception exp) { + if (connection != null) + connection.close(); + System.out.println("Caught exception, exiting."); + exp.printStackTrace(System.out); + System.exit(1); + } + } + + protected static boolean isSupportedAmqpType(String amqpType) { + for (String supportedAmqpType: SUPPORTED_AMQP_TYPES) { + if (amqpType.equals(supportedAmqpType)) + return true; + } + return false; + } + + private static class MyExceptionListener implements ExceptionListener { + @Override + public void onException(JMSException exception) { + System.out.println("Connection ExceptionListener fired, exiting."); + exception.printStackTrace(System.out); + System.exit(1); + } + } + + public Receiver(String brokerAddress, String queueName) { + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/83b89fe4/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/amqp_types_test/Sender.java ---------------------------------------------------------------------- diff --git a/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/amqp_types_test/Sender.java b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/amqp_types_test/Sender.java new file mode 100644 index 0000000..7b436de --- /dev/null +++ b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/amqp_types_test/Sender.java @@ -0,0 +1,260 @@ +/** + * 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.interop_test.amqp_types_test; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.math.MathContext; +import java.util.Arrays; +import java.util.UUID; +import javax.jms.BytesMessage; +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.DeliveryMode; +import javax.jms.ExceptionListener; +import javax.jms.JMSException; +import javax.jms.MapMessage; +import javax.jms.Message; +import javax.jms.MessageProducer; +import javax.jms.ObjectMessage; +import javax.jms.Queue; +import javax.jms.Session; +import javax.jms.StreamMessage; +import javax.jms.TextMessage; +import javax.naming.Context; +import javax.naming.InitialContext; +import org.apache.qpid.jms.JmsConnectionFactory; + +public class Sender { + private static final String USER = "guest"; + private static final String PASSWORD = "guest"; + private static final String[] SUPPORTED_AMQP_TYPES = {"null", + "boolean", + "ubyte", + "ushort", + "uint", + "ulong", + "byte", + "short", + "int", + "long", + "float", + "double", + "decimal32", + "decimal64", + "decimal128", + "char", + "timestamp", + "uuid", + "binary", + "string", + "symbol", + "list", + "map", + "array"}; + + public static void main(String[] args) throws Exception { + if (args.length < 4) { + System.out.println("AmqpSender: Insufficient number of arguments"); + System.out.println("AmqpSender: Expected arguments: broker_address, queue_name, amqp_type, test_val, test_val, ..."); + System.exit(1); + } + String brokerAddress = "amqp://" + args[0]; + String queueName = args[1]; + String amqpType = args[2]; + String[] testValueList = Arrays.copyOfRange(args, 3, args.length); // Use remaining args as test values + + try { + ConnectionFactory factory = (ConnectionFactory)new JmsConnectionFactory(brokerAddress); + + Connection connection = factory.createConnection(); + connection.setExceptionListener(new MyExceptionListener()); + connection.start(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + Queue queue = session.createQueue(queueName); + + MessageProducer messageProducer = session.createProducer(queue); + + if (isSupportedAmqpType(amqpType)) { + Message message = null; + for (String testValueStr : testValueList) { + switch (amqpType) { + case "null": + message = session.createBytesMessage(); + break; + case "boolean": + message = session.createBytesMessage(); + ((BytesMessage)message).writeBoolean(Boolean.parseBoolean(testValueStr)); + break; + case "ubyte": + { + byte testValue = (byte)Short.parseShort(testValueStr); + message = session.createBytesMessage(); + ((BytesMessage)message).writeByte(testValue); + break; + } + case "ushort": + { + int testValue = Integer.parseInt(testValueStr); + byte[] byteArray = new byte[2]; + byteArray[0] = (byte)(testValue >> 8); + byteArray[1] = (byte)(testValue); + message = session.createBytesMessage(); + ((BytesMessage)message).writeBytes(byteArray); + break; + } + case "uint": + { + long testValue = Long.parseLong(testValueStr); + byte[] byteArray = new byte[4]; + byteArray[0] = (byte)(testValue >> 24); + byteArray[1] = (byte)(testValue >> 16); + byteArray[2] = (byte)(testValue >> 8); + byteArray[3] = (byte)(testValue); + message = session.createBytesMessage(); + ((BytesMessage)message).writeBytes(byteArray); + break; + } + case "ulong": + { + // TODO: Tidy this ugliness up - perhaps use of vector<byte>? + BigInteger testValue = new BigInteger(testValueStr); + byte[] bigIntArray = testValue.toByteArray(); // may be 1 to 9 bytes depending on number + byte[] byteArray = {0, 0, 0, 0, 0, 0, 0, 0}; + int effectiveBigIntArrayLen = bigIntArray.length > 8 ? 8 : bigIntArray.length; // Cap length at 8 + int bigIntArrayOffs = bigIntArray.length > 8 ? bigIntArray.length - 8 : 0; // Offset when length > 8 + for (int i=0; i<bigIntArray.length && i < 8; i++) + byteArray[8 - effectiveBigIntArrayLen + i] = bigIntArray[bigIntArrayOffs + i]; + message = session.createBytesMessage(); + ((BytesMessage)message).writeBytes(byteArray); + break; + } + case "byte": + message = session.createBytesMessage(); + ((BytesMessage)message).writeByte(Byte.parseByte(testValueStr)); + break; + case "short": + message = session.createBytesMessage(); + ((BytesMessage)message).writeShort(Short.parseShort(testValueStr)); + break; + case "int": + message = session.createBytesMessage(); + ((BytesMessage)message).writeInt(Integer.parseInt(testValueStr)); + break; + case "long": + case "timestamp": + message = session.createBytesMessage(); + ((BytesMessage)message).writeLong(Long.parseLong(testValueStr)); + break; + case "float": + Long i = Long.parseLong(testValueStr.substring(2), 16); + message = session.createBytesMessage(); + ((BytesMessage)message).writeFloat(Float.intBitsToFloat(i.intValue())); + break; + case "double": + Long l1 = Long.parseLong(testValueStr.substring(2, 3), 16) << 60; + Long l2 = Long.parseLong(testValueStr.substring(3), 16); + message = session.createBytesMessage(); + ((BytesMessage)message).writeDouble(Double.longBitsToDouble(l1 | l2)); + break; + case "decimal32": + BigDecimal bd32 = new BigDecimal(testValueStr, MathContext.DECIMAL32); + message = session.createObjectMessage(); + ((ObjectMessage)message).setObject(bd32); + break; + case "decimal64": + BigDecimal bd64 = new BigDecimal(testValueStr, MathContext.DECIMAL64); + message = session.createObjectMessage(); + ((ObjectMessage)message).setObject(bd64); + break; + case "decimal128": + BigDecimal bd128 = new BigDecimal(testValueStr, MathContext.DECIMAL128); + message = session.createObjectMessage(); + ((ObjectMessage)message).setObject(bd128); + break; + case "char": + char c = 0; + if (testValueStr.length() == 1) // Single char + c = testValueStr.charAt(0); + else if (testValueStr.length() == 6) // unicode format + c = (char)Integer.parseInt(testValueStr, 16); + message = session.createBytesMessage(); + ((BytesMessage)message).writeChar(c); + break; + case "uuid": + UUID uuid = UUID.fromString(testValueStr); + message = session.createObjectMessage(); + ((ObjectMessage)message).setObject(uuid); + break; + case "binary": + message = session.createBytesMessage(); + byte[] byteArray = testValueStr.getBytes(); + ((BytesMessage)message).writeBytes(byteArray, 0, byteArray.length); + break; + case "string": + message = session.createTextMessage(testValueStr); + break; + case "symbol": + message = session.createBytesMessage(); + ((BytesMessage)message).writeUTF(testValueStr); + break; + case "list": + break; + case "map": + break; + case "array": + break; + default: + // Internal error, should never happen if SUPPORTED_AMQP_TYPES matches this case stmt + connection.close(); + throw new Exception("AmqpSender: Internal error: unsupported AMQP type \"" + amqpType + "\""); + } + messageProducer.send(message, DeliveryMode.NON_PERSISTENT, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE); + } + } else { + System.out.println("ERROR: AmqpSender: AMQP type \"" + amqpType + "\" is not supported"); + connection.close(); + System.exit(1); + } + + connection.close(); + } catch (Exception exp) { + System.out.println("Caught exception, exiting."); + exp.printStackTrace(System.out); + System.exit(1); + } + } + + protected static boolean isSupportedAmqpType(String amqpType) { + for (String supportedAmqpType: SUPPORTED_AMQP_TYPES) { + if (amqpType.equals(supportedAmqpType)) + return true; + } + return false; + } + + private static class MyExceptionListener implements ExceptionListener { + @Override + public void onException(JMSException exception) { + System.out.println("Connection ExceptionListener fired, exiting."); + exception.printStackTrace(System.out); + System.exit(1); + } + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/83b89fe4/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/jms_messages_test/Receiver.java ---------------------------------------------------------------------- diff --git a/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/jms_messages_test/Receiver.java b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/jms_messages_test/Receiver.java new file mode 100644 index 0000000..29e8ddc --- /dev/null +++ b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/jms_messages_test/Receiver.java @@ -0,0 +1,525 @@ +/** + * 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.interop_test.jms_messages_test; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.StringReader; +import java.io.StringWriter; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Enumeration; +import java.util.List; +import javax.jms.BytesMessage; +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.Destination; +import javax.jms.ExceptionListener; +import javax.jms.JMSException; +import javax.jms.MapMessage; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.ObjectMessage; +import javax.jms.Queue; +import javax.jms.Session; +import javax.jms.StreamMessage; +import javax.jms.TextMessage; +import javax.jms.Topic; +import javax.json.Json; +import javax.json.JsonArray; +import javax.json.JsonArrayBuilder; +import javax.json.JsonObject; +import javax.json.JsonObjectBuilder; +import javax.json.JsonReader; +import javax.json.JsonWriter; +import org.apache.qpid.jms.JmsConnectionFactory; + +public class Receiver { + private static final String USER = "guest"; + private static final String PASSWORD = "guest"; + private static final int TIMEOUT = 1000; + private static final String[] SUPPORTED_JMS_MESSAGE_TYPES = {"JMS_MESSAGE_TYPE", + "JMS_BYTESMESSAGE_TYPE", + "JMS_MAPMESSAGE_TYPE", + "JMS_OBJECTMESSAGE_TYPE", + "JMS_STREAMMESSAGE_TYPE", + "JMS_TEXTMESSAGE_TYPE"}; + private static enum JMS_DESTINATION_TYPE {JMS_QUEUE, JMS_TEMPORARY_QUEUE, JMS_TOPIC, JMS_TEMPORARY_TOPIC}; + + Connection _connection; + Session _session; + Queue _queue; + MessageConsumer _messageConsumer; + JsonObjectBuilder _jsonTestValueMapBuilder; + JsonObjectBuilder _jsonMessageHeaderMapBuilder; + JsonObjectBuilder _jsonMessagePropertiesMapBuilder; + + // args[0]: Broker URL + // args[1]: Queue name + // args[2]: JMS message type + // args[3]: JSON Test parameters containing 2 maps: [testValuesMap, flagMap] + public static void main(String[] args) throws Exception { + if (args.length != 4) { + System.out.println("JmsReceiverShim: Incorrect number of arguments"); + System.out.println("JmsReceiverShim: Expected arguments: broker_address, queue_name, JMS_msg_type, JSON_receive_params"); + System.exit(1); + } + String brokerAddress = "amqp://" + args[0]; + String queueName = args[1]; + String jmsMessageType = args[2]; + if (!isSupportedJmsMessageType(jmsMessageType)) { + System.out.println("ERROR: JmsReceiverShim: Unknown or unsupported JMS message type \"" + jmsMessageType + "\""); + System.exit(1); + } + + JsonReader jsonReader = Json.createReader(new StringReader(args[3])); + JsonArray testParamsList = jsonReader.readArray(); + jsonReader.close(); + + if (testParamsList.size() != 2) { + System.err.println("ERROR: Incorrect number of JSON parameters: Expected 2, got " + Integer.toString(testParamsList.size())); + System.exit(1); + } + + JsonObject numTestValuesMap = testParamsList.getJsonObject(0); + JsonObject flagMap = testParamsList.getJsonObject(1); + + Receiver shim = new Receiver(brokerAddress, queueName); + shim.run(jmsMessageType, numTestValuesMap, flagMap); + } + + public Receiver(String brokerAddress, String queueName) { + try { + _connection = null; + ConnectionFactory factory = (ConnectionFactory)new JmsConnectionFactory(brokerAddress); + _connection = factory.createConnection(USER, PASSWORD); + _connection.setExceptionListener(new MyExceptionListener()); + _connection.start(); + + _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + _queue = _session.createQueue(queueName); + + _messageConsumer = _session.createConsumer(_queue); + + _jsonTestValueMapBuilder = Json.createObjectBuilder(); + _jsonMessageHeaderMapBuilder = Json.createObjectBuilder(); + _jsonMessagePropertiesMapBuilder = Json.createObjectBuilder(); + } catch (Exception exc) { + if (_connection != null) + try { _connection.close(); } catch (JMSException e) {} + System.out.println("Caught exception, exiting."); + exc.printStackTrace(System.out); + System.exit(1); + } + } + + public void run(String jmsMessageType, JsonObject numTestValuesMap, JsonObject flagMap) { + try { + List<String> subTypeKeyList = new ArrayList<String>(numTestValuesMap.keySet()); + Collections.sort(subTypeKeyList); + + Message message = null; + + for (String subType: subTypeKeyList) { + JsonArrayBuilder jasonTestValuesArrayBuilder = Json.createArrayBuilder(); + for (int i=0; i<numTestValuesMap.getJsonNumber(subType).intValue(); ++i) { + message = _messageConsumer.receive(TIMEOUT); + if (message == null) break; + switch (jmsMessageType) { + case "JMS_MESSAGE_TYPE": + processJMSMessage(jasonTestValuesArrayBuilder); + break; + case "JMS_BYTESMESSAGE_TYPE": + processJMSBytesMessage(jmsMessageType, subType, message, jasonTestValuesArrayBuilder); + break; + case "JMS_STREAMMESSAGE_TYPE": + processJMSStreamMessage(jmsMessageType, subType, message, jasonTestValuesArrayBuilder); + break; + case "JMS_MAPMESSAGE_TYPE": + processJMSMapMessage(jmsMessageType, subType, i, message, jasonTestValuesArrayBuilder); + break; + case "JMS_OBJECTMESSAGE_TYPE": + processJMSObjectMessage(subType, message, jasonTestValuesArrayBuilder); + break; + case "JMS_TEXTMESSAGE_TYPE": + processJMSTextMessage(message, jasonTestValuesArrayBuilder); + break; + default: + _connection.close(); + throw new Exception("JmsReceiverShim: Internal error: Unknown or unsupported JMS message type \"" + jmsMessageType + "\""); + } + + processMessageHeaders(message, flagMap); + processMessageProperties(message); + } + _jsonTestValueMapBuilder.add(subType, jasonTestValuesArrayBuilder); + } + _connection.close(); + + System.out.println(jmsMessageType); + StringWriter out = new StringWriter(); + writeJsonObject(_jsonTestValueMapBuilder, out); + out.append('\n'); + writeJsonObject(_jsonMessageHeaderMapBuilder, out); + out.append('\n'); + writeJsonObject(_jsonMessagePropertiesMapBuilder, out); + System.out.println(out.toString()); + } catch (Exception exp) { + try { _connection.close(); } catch (JMSException e) {} + System.out.println("Caught exception, exiting."); + exp.printStackTrace(System.out); + System.exit(1); + } + } + + protected void processJMSMessage(JsonArrayBuilder jasonTestValuesArrayBuilder) { + jasonTestValuesArrayBuilder.addNull(); + } + + protected void processJMSBytesMessage(String jmsMessageType, String subType, Message message, JsonArrayBuilder jasonTestValuesArrayBuilder) throws Exception, JMSException, IOException, ClassNotFoundException { + switch (subType) { + case "boolean": + jasonTestValuesArrayBuilder.add(((BytesMessage)message).readBoolean()?"True":"False"); + break; + case "byte": + jasonTestValuesArrayBuilder.add(formatByte(((BytesMessage)message).readByte())); + break; + case "bytes": + { + byte[] bytesBuff = new byte[65536]; + int numBytesRead = ((BytesMessage)message).readBytes(bytesBuff); + if (numBytesRead >= 0) { + jasonTestValuesArrayBuilder.add(new String(Arrays.copyOfRange(bytesBuff, 0, numBytesRead))); + } else { + // NOTE: For this case, an empty byte array has nothing to return + jasonTestValuesArrayBuilder.add(new String()); + } + } + break; + case "char": + jasonTestValuesArrayBuilder.add(formatChar(((BytesMessage)message).readChar())); + break; + case "double": + long l = Double.doubleToRawLongBits(((BytesMessage)message).readDouble()); + jasonTestValuesArrayBuilder.add(String.format("0x%16s", Long.toHexString(l)).replace(' ', '0')); + break; + case "float": + int i0 = Float.floatToRawIntBits(((BytesMessage)message).readFloat()); + jasonTestValuesArrayBuilder.add(String.format("0x%8s", Integer.toHexString(i0)).replace(' ', '0')); + break; + case "int": + jasonTestValuesArrayBuilder.add(formatInt(((BytesMessage)message).readInt())); + break; + case "long": + jasonTestValuesArrayBuilder.add(formatLong(((BytesMessage)message).readLong())); + break; + case "object": + { + byte[] bytesBuff = new byte[65536]; + int numBytesRead = ((BytesMessage)message).readBytes(bytesBuff); + if (numBytesRead >= 0) { + ByteArrayInputStream bais = new ByteArrayInputStream(Arrays.copyOfRange(bytesBuff, 0, numBytesRead)); + ObjectInputStream ois = new ObjectInputStream(bais); + Object obj = ois.readObject(); + jasonTestValuesArrayBuilder.add(obj.getClass().getName() + ":" + obj.toString()); + } else { + jasonTestValuesArrayBuilder.add("<object error>"); + } + } + break; + case "short": + jasonTestValuesArrayBuilder.add(formatShort(((BytesMessage)message).readShort())); + break; + case "string": + jasonTestValuesArrayBuilder.add(((BytesMessage)message).readUTF()); + break; + default: + throw new Exception("JmsReceiverShim: Unknown subtype for " + jmsMessageType + ": \"" + subType + "\""); + } + } + + protected void processJMSMapMessage(String jmsMessageType, String subType, int count, Message message, JsonArrayBuilder jasonTestValuesArrayBuilder) throws Exception, JMSException { + String name = String.format("%s%03d", subType, count); + switch (subType) { + case "boolean": + jasonTestValuesArrayBuilder.add(((MapMessage)message).getBoolean(name)?"True":"False"); + break; + case "byte": + jasonTestValuesArrayBuilder.add(formatByte(((MapMessage)message).getByte(name))); + break; + case "bytes": + jasonTestValuesArrayBuilder.add(new String(((MapMessage)message).getBytes(name))); + break; + case "char": + jasonTestValuesArrayBuilder.add(formatChar(((MapMessage)message).getChar(name))); + break; + case "double": + long l = Double.doubleToRawLongBits(((MapMessage)message).getDouble(name)); + jasonTestValuesArrayBuilder.add(String.format("0x%16s", Long.toHexString(l)).replace(' ', '0')); + break; + case "float": + int i0 = Float.floatToRawIntBits(((MapMessage)message).getFloat(name)); + jasonTestValuesArrayBuilder.add(String.format("0x%8s", Integer.toHexString(i0)).replace(' ', '0')); + break; + case "int": + jasonTestValuesArrayBuilder.add(formatInt(((MapMessage)message).getInt(name))); + break; + case "long": + jasonTestValuesArrayBuilder.add(formatLong(((MapMessage)message).getLong(name))); + break; + case "object": + Object obj = ((MapMessage)message).getObject(name); + jasonTestValuesArrayBuilder.add(obj.getClass().getName() + ":" + obj.toString()); + break; + case "short": + jasonTestValuesArrayBuilder.add(formatShort(((MapMessage)message).getShort(name))); + break; + case "string": + jasonTestValuesArrayBuilder.add(((MapMessage)message).getString(name)); + break; + default: + throw new Exception("JmsReceiverShim: Unknown subtype for " + jmsMessageType + ": \"" + subType + "\""); + } + } + + protected void processJMSObjectMessage(String subType, Message message, JsonArrayBuilder jasonTestValuesArrayBuilder) throws JMSException { + jasonTestValuesArrayBuilder.add(((ObjectMessage)message).getObject().toString()); + } + + protected void processJMSStreamMessage(String jmsMessageType, String subType, Message message, JsonArrayBuilder jasonTestValuesArrayBuilder) throws Exception, JMSException { + switch (subType) { + case "boolean": + jasonTestValuesArrayBuilder.add(((StreamMessage)message).readBoolean()?"True":"False"); + break; + case "byte": + jasonTestValuesArrayBuilder.add(formatByte(((StreamMessage)message).readByte())); + break; + case "bytes": + byte[] bytesBuff = new byte[65536]; + int numBytesRead = ((StreamMessage)message).readBytes(bytesBuff); + if (numBytesRead >= 0) { + jasonTestValuesArrayBuilder.add(new String(Arrays.copyOfRange(bytesBuff, 0, numBytesRead))); + } else { + System.out.println("StreamMessage.readBytes() returned " + numBytesRead); + jasonTestValuesArrayBuilder.add("<bytes error>"); + } + break; + case "char": + jasonTestValuesArrayBuilder.add(formatChar(((StreamMessage)message).readChar())); + break; + case "double": + long l = Double.doubleToRawLongBits(((StreamMessage)message).readDouble()); + jasonTestValuesArrayBuilder.add(String.format("0x%16s", Long.toHexString(l)).replace(' ', '0')); + break; + case "float": + int i0 = Float.floatToRawIntBits(((StreamMessage)message).readFloat()); + jasonTestValuesArrayBuilder.add(String.format("0x%8s", Integer.toHexString(i0)).replace(' ', '0')); + break; + case "int": + jasonTestValuesArrayBuilder.add(formatInt(((StreamMessage)message).readInt())); + break; + case "long": + jasonTestValuesArrayBuilder.add(formatLong(((StreamMessage)message).readLong())); + break; + case "object": + Object obj = ((StreamMessage)message).readObject(); + jasonTestValuesArrayBuilder.add(obj.getClass().getName() + ":" + obj.toString()); + break; + case "short": + jasonTestValuesArrayBuilder.add(formatShort(((StreamMessage)message).readShort())); + break; + case "string": + jasonTestValuesArrayBuilder.add(((StreamMessage)message).readString()); + break; + default: + throw new Exception("JmsReceiverShim: Unknown subtype for " + jmsMessageType + ": \"" + subType + "\""); + } + } + + protected void processJMSTextMessage(Message message, JsonArrayBuilder jasonTestValuesArrayBuilder) throws JMSException { + jasonTestValuesArrayBuilder.add(((TextMessage)message).getText()); + } + + protected void processMessageHeaders(Message message, JsonObject flagMap) throws Exception, JMSException { + addMessageHeaderString("JMS_TYPE_HEADER", message.getJMSType()); + if (flagMap.containsKey("JMS_CORRELATIONID_AS_BYTES") && flagMap.getBoolean("JMS_CORRELATIONID_AS_BYTES")) { + addMessageHeaderByteArray("JMS_CORRELATIONID_HEADER", message.getJMSCorrelationIDAsBytes()); + } else { + addMessageHeaderString("JMS_CORRELATIONID_HEADER", message.getJMSCorrelationID()); + } + if (flagMap.containsKey("JMS_REPLYTO_AS_TOPIC") && flagMap.getBoolean("JMS_REPLYTO_AS_TOPIC")) { + addMessageHeaderDestination("JMS_REPLYTO_HEADER", JMS_DESTINATION_TYPE.JMS_TOPIC, message.getJMSReplyTo()); + } else { + addMessageHeaderDestination("JMS_REPLYTO_HEADER", JMS_DESTINATION_TYPE.JMS_QUEUE, message.getJMSReplyTo()); + } + } + + protected void addMessageHeaderString(String headerName, String value) { + if (value != null) { + JsonObjectBuilder valueMap = Json.createObjectBuilder(); + valueMap.add("string", value); + _jsonMessageHeaderMapBuilder.add(headerName, valueMap); + } + } + + protected void addMessageHeaderByteArray(String headerName, byte[] value) { + if (value != null) { + JsonObjectBuilder valueMap = Json.createObjectBuilder(); + valueMap.add("bytes", new String(value)); + _jsonMessageHeaderMapBuilder.add(headerName, valueMap); + } + } + + protected void addMessageHeaderDestination(String headerName, JMS_DESTINATION_TYPE destinationType, Destination destination) throws Exception { + if (destination != null) { + JsonObjectBuilder valueMap = Json.createObjectBuilder(); + switch (destinationType) { + case JMS_QUEUE: + valueMap.add("queue", ((Queue)destination).getQueueName()); + break; + case JMS_TOPIC: + valueMap.add("topic", ((Topic)destination).getTopicName()); + break; + default: + throw new Exception("Internal error: JMSDestination type not supported"); + } + _jsonMessageHeaderMapBuilder.add(headerName, valueMap); + } + } + + protected void processMessageProperties(Message message) throws Exception, JMSException { + Enumeration<String> propertyNames = message.getPropertyNames(); + while (propertyNames.hasMoreElements()) { + JsonObjectBuilder valueMap = Json.createObjectBuilder(); + String propertyName = propertyNames.nextElement(); + int underscoreIndex = propertyName.indexOf('_'); + if (underscoreIndex >= 0) { + String propType = propertyName.substring(0, underscoreIndex); + switch (propType) { + case "boolean": + valueMap.add(propType, message.getBooleanProperty(propertyName) ? "True" : "False"); + _jsonMessagePropertiesMapBuilder.add(propertyName, valueMap); + break; + case "byte": + valueMap.add(propType, formatByte(message.getByteProperty(propertyName))); + _jsonMessagePropertiesMapBuilder.add(propertyName, valueMap); + break; + case "double": + long l = Double.doubleToRawLongBits(message.getDoubleProperty(propertyName)); + valueMap.add(propType, String.format("0x%16s", Long.toHexString(l)).replace(' ', '0')); + _jsonMessagePropertiesMapBuilder.add(propertyName, valueMap); + break; + case "float": + int i = Float.floatToRawIntBits(message.getFloatProperty(propertyName)); + valueMap.add(propType, String.format("0x%8s", Integer.toHexString(i)).replace(' ', '0')); + _jsonMessagePropertiesMapBuilder.add(propertyName, valueMap); + break; + case "int": + valueMap.add(propType, formatInt(message.getIntProperty(propertyName))); + _jsonMessagePropertiesMapBuilder.add(propertyName, valueMap); + break; + case "long": + valueMap.add(propType, formatLong(message.getLongProperty(propertyName))); + _jsonMessagePropertiesMapBuilder.add(propertyName, valueMap); + break; + case "short": + valueMap.add(propType, formatShort(message.getShortProperty(propertyName))); + _jsonMessagePropertiesMapBuilder.add(propertyName, valueMap); + break; + case "string": + valueMap.add(propType, message.getStringProperty(propertyName)); + _jsonMessagePropertiesMapBuilder.add(propertyName, valueMap); + break; + default: + ; // Ignore any other property the broker may add + } + } else { + // TODO: handle other non-test properties that might exist here + } + } + } + + protected static void writeJsonObject(JsonObjectBuilder builder, StringWriter out) { + JsonWriter jsonWriter = Json.createWriter(out); + jsonWriter.writeObject(builder.build()); + jsonWriter.close(); + } + + protected static String formatByte(byte b) { + boolean neg = false; + if (b < 0) { + neg = true; + b = (byte)-b; + } + return String.format("%s0x%x", neg?"-":"", b); + } + + protected static String formatChar(char c) { + if (Character.isLetterOrDigit(c)) { + return String.format("%c", c); + } + char[] ca = {c}; + return new String(ca); + } + + protected static String formatInt(int i) { + boolean neg = false; + if (i < 0) { + neg = true; + i = -i; + } + return String.format("%s0x%x", neg?"-":"", i); + } + + protected static String formatLong(long l) { + boolean neg = false; + if (l < 0) { + neg = true; + l = -l; + } + return String.format("%s0x%x", neg?"-":"", l); + } + + protected static String formatShort(int s) { + boolean neg = false; + if (s < 0) { + neg = true; + s = -s; + } + return String.format("%s0x%x", neg?"-":"", s); + } + + protected static boolean isSupportedJmsMessageType(String jmsMessageType) { + for (String supportedJmsMessageType: SUPPORTED_JMS_MESSAGE_TYPES) { + if (jmsMessageType.equals(supportedJmsMessageType)) + return true; + } + return false; + } + + private static class MyExceptionListener implements ExceptionListener { + @Override + public void onException(JMSException exception) { + System.out.println("Connection ExceptionListener fired, exiting."); + exception.printStackTrace(System.out); + System.exit(1); + } + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/83b89fe4/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/jms_messages_test/Sender.java ---------------------------------------------------------------------- diff --git a/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/jms_messages_test/Sender.java b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/jms_messages_test/Sender.java new file mode 100644 index 0000000..0a0539c --- /dev/null +++ b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/jms_messages_test/Sender.java @@ -0,0 +1,508 @@ +/** + * 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.interop_test.jms_messages_test; + +import java.io.Serializable; +import java.io.StringReader; +import java.lang.reflect.Constructor; +import java.lang.reflect.InvocationTargetException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import javax.jms.BytesMessage; +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.DeliveryMode; +import javax.jms.ExceptionListener; +import javax.jms.JMSException; +import javax.jms.MapMessage; +import javax.jms.Message; +import javax.jms.MessageProducer; +import javax.jms.ObjectMessage; +import javax.jms.Queue; +import javax.jms.Session; +import javax.jms.StreamMessage; +import javax.jms.TextMessage; +import javax.json.Json; +import javax.json.JsonArray; +import javax.json.JsonObject; +import javax.json.JsonReader; +import org.apache.qpid.jms.JmsConnectionFactory; + +public class Sender { + private static final String USER = "guest"; + private static final String PASSWORD = "guest"; + private static final String[] SUPPORTED_JMS_MESSAGE_TYPES = {"JMS_MESSAGE_TYPE", + "JMS_BYTESMESSAGE_TYPE", + "JMS_MAPMESSAGE_TYPE", + "JMS_OBJECTMESSAGE_TYPE", + "JMS_STREAMMESSAGE_TYPE", + "JMS_TEXTMESSAGE_TYPE"}; + Connection _connection; + Session _session; + Queue _queue; + MessageProducer _messageProducer; + int _msgsSent; + + + // args[0]: Broker URL + // args[1]: Queue name + // args[2]: JMS message type + // args[3]: JSON Test parameters containing 3 maps: [testValueMap, testHeadersMap, testPropertiesMap] + public static void main(String[] args) throws Exception { + if (args.length != 4) { + System.out.println("JmsSenderShim: Incorrect number of arguments"); + System.out.println("JmsSenderShim: Expected arguments: broker_address, queue_name, JMS_msg_type, JSON_send_params"); + System.exit(1); + } + String brokerAddress = "amqp://" + args[0]; + String queueName = args[1]; + String jmsMessageType = args[2]; + if (!isSupportedJmsMessageType(jmsMessageType)) { + System.err.println("ERROR: JmsSender: Unknown or unsupported JMS message type \"" + jmsMessageType + "\""); + System.exit(1); + } + + JsonReader jsonReader = Json.createReader(new StringReader(args[3])); + JsonArray testParamsList = jsonReader.readArray(); + jsonReader.close(); + + if (testParamsList.size() != 3) { + System.err.println("ERROR: Incorrect number of JSON parameters: Expected 3, got " + Integer.toString(testParamsList.size())); + System.err.println(" JSON parameters found: \"" + testParamsList + "\""); + System.exit(1); + } + JsonObject testValuesMap = testParamsList.getJsonObject(0); + JsonObject testHeadersMap = testParamsList.getJsonObject(1); + JsonObject testPropertiesMap = testParamsList.getJsonObject(2); + + Sender shim = new Sender(brokerAddress, queueName); + shim.runTests(jmsMessageType, testValuesMap, testHeadersMap, testPropertiesMap); + } + + public Sender(String brokerAddress, String queueName) { + try { + ConnectionFactory factory = (ConnectionFactory)new JmsConnectionFactory(brokerAddress); + + _connection = factory.createConnection(); + _connection.setExceptionListener(new MyExceptionListener()); + _connection.start(); + + _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + _queue = _session.createQueue(queueName); + + _messageProducer = _session.createProducer(_queue); + + _msgsSent = 0; + } catch (Exception exp) { + System.out.println("Caught exception, exiting."); + exp.printStackTrace(System.out); + System.exit(1); + } + } + + public void runTests(String jmsMessageType, JsonObject testValuesMap, JsonObject testHeadersMap, JsonObject testPropertiesMap) throws Exception { + List<String> testValuesKeyList = new ArrayList<String>(testValuesMap.keySet()); + Collections.sort(testValuesKeyList); + for (String key: testValuesKeyList) { + JsonArray testValues = testValuesMap.getJsonArray(key); + for (int i=0; i<testValues.size(); ++i) { + String testValue = ""; + if (!testValues.isNull(i)) { + testValue = testValues.getJsonString(i).getString(); + } + + // Send message + Message msg = createMessage(jmsMessageType, key, testValue, i); + addMessageHeaders(msg, testHeadersMap); + addMessageProperties(msg, testPropertiesMap); + _messageProducer.send(msg, DeliveryMode.NON_PERSISTENT, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE); + _msgsSent++; + } + } + _connection.close(); + } + + protected Message createMessage(String jmsMessageType, String key, String testValue, int i) throws Exception { + Message message = null; + switch (jmsMessageType) { + case "JMS_MESSAGE_TYPE": + message = createJMSMessage(key, testValue); + break; + case "JMS_BYTESMESSAGE_TYPE": + message = createJMSBytesMessage(key, testValue); + break; + case "JMS_MAPMESSAGE_TYPE": + message = createJMSMapMessage(key, testValue, i); + break; + case "JMS_OBJECTMESSAGE_TYPE": + message = createJMSObjectMessage(key, testValue); + break; + case "JMS_STREAMMESSAGE_TYPE": + message = createJMSStreamMessage(key, testValue); + break; + case "JMS_TEXTMESSAGE_TYPE": + message = createTextMessage(testValue); + break; + default: + throw new Exception("Internal exception: Unexpected JMS message type \"" + jmsMessageType + "\""); + } + return message; + } + + + protected void addMessageHeaders(Message msg, JsonObject testHeadersMap) throws Exception, JMSException { + List<String> testHeadersKeyList = new ArrayList<String>(testHeadersMap.keySet()); + for (String key: testHeadersKeyList) { + JsonObject subMap = testHeadersMap.getJsonObject(key); + List<String> subMapKeyList = new ArrayList<String>(subMap.keySet()); + String subMapKey = subMapKeyList.get(0); // There is always only one entry in map + String subMapVal = subMap.getString(subMapKey); + switch (key) { + case "JMS_TYPE_HEADER": + if (subMapKey.compareTo("string") == 0) { + msg.setJMSType(subMapVal); + } else { + throw new Exception("Internal exception: Invalid message header type \"" + subMapKey + "\" for message header \"" + key + "\""); + } + break; + case "JMS_CORRELATIONID_HEADER": + if (subMapKey.compareTo("string") == 0) { + msg.setJMSCorrelationID(subMapVal); + } else if (subMapKey.compareTo("bytes") == 0) { + msg.setJMSCorrelationIDAsBytes(subMapVal.getBytes()); + } else { + throw new Exception("Internal exception: Invalid message header type \"" + subMapKey + "\" for message header \"" + key + "\""); + } + break; + case "JMS_REPLYTO_HEADER": + switch (subMapKey) { + case "queue": + msg.setJMSReplyTo(_session.createQueue(subMapVal)); + break; + case "temp_queue": + msg.setJMSReplyTo(_session.createTemporaryQueue()); + break; + case "topic": + msg.setJMSReplyTo(_session.createTopic(subMapVal)); + break; + case "temp_topic": + msg.setJMSReplyTo(_session.createTemporaryTopic()); + break; + default: + throw new Exception("Internal exception: Invalid message header type \"" + subMapKey + "\" for message header \"" + key + "\""); + } + break; + default: + throw new Exception("Internal exception: Unknown or unsupported message header \"" + key + "\""); + } + } + } + + protected void addMessageProperties(Message msg, JsonObject testPropertiesMap) throws Exception, JMSException { + List<String> testPropertiesKeyList = new ArrayList<String>(testPropertiesMap.keySet()); + for (String key: testPropertiesKeyList) { + JsonObject subMap = testPropertiesMap.getJsonObject(key); + List<String> subMapKeyList = new ArrayList<String>(subMap.keySet()); + String subMapKey = subMapKeyList.get(0); // There is always only one entry in map + String subMapVal = subMap.getString(subMapKey); + switch (subMapKey) { + case "boolean": + msg.setBooleanProperty(key, Boolean.parseBoolean(subMapVal)); + break; + case "byte": + msg.setByteProperty(key, Byte.decode(subMapVal)); + break; + case "double": + Long l1 = Long.parseLong(subMapVal.substring(2, 3), 16) << 60; + Long l2 = Long.parseLong(subMapVal.substring(3), 16); + msg.setDoubleProperty(key, Double.longBitsToDouble(l1 | l2)); + break; + case "float": + Long i = Long.parseLong(subMapVal.substring(2), 16); + msg.setFloatProperty(key, Float.intBitsToFloat(i.intValue())); + break; + case "int": + msg.setIntProperty(key, Integer.decode(subMapVal)); + break; + case "long": + msg.setLongProperty(key, Long.decode(subMapVal)); + break; + case "short": + msg.setShortProperty(key, Short.decode(subMapVal)); + break; + case "string": + msg.setStringProperty(key, subMapVal); + break; + default: + throw new Exception("Internal exception: Unknown or unsupported message property type \"" + subMapKey + "\""); + } + } + } + + protected Message createJMSMessage(String testValueType, String testValue) throws Exception, JMSException { + if (testValueType.compareTo("none") != 0) { + throw new Exception("Internal exception: Unexpected JMS message sub-type \"" + testValueType + "\""); + } + if (testValue.length() > 0) { + throw new Exception("Internal exception: Unexpected JMS message value \"" + testValue + "\" for sub-type \"" + testValueType + "\""); + } + return _session.createMessage(); + } + + protected BytesMessage createJMSBytesMessage(String testValueType, String testValue) throws Exception, JMSException { + BytesMessage message = _session.createBytesMessage(); + switch (testValueType) { + case "boolean": + message.writeBoolean(Boolean.parseBoolean(testValue)); + break; + case "byte": + message.writeByte(Byte.decode(testValue)); + break; + case "bytes": + message.writeBytes(testValue.getBytes()); + break; + case "char": + if (testValue.length() == 1) { // Char format: "X" or "\xNN" + message.writeChar(testValue.charAt(0)); + } else { + throw new Exception("JmsSenderShim.createJMSBytesMessage() Malformed char string: \"" + testValue + "\" of length " + testValue.length()); + } + break; + case "double": + Long l1 = Long.parseLong(testValue.substring(2, 3), 16) << 60; + Long l2 = Long.parseLong(testValue.substring(3), 16); + message.writeDouble(Double.longBitsToDouble(l1 | l2)); + break; + case "float": + Long i = Long.parseLong(testValue.substring(2), 16); + message.writeFloat(Float.intBitsToFloat(i.intValue())); + break; + case "int": + message.writeInt(Integer.decode(testValue)); + break; + case "long": + message.writeLong(Long.decode(testValue)); + break; + case "object": + Object obj = (Object)createObject(testValue); + message.writeObject(obj); + break; + case "short": + message.writeShort(Short.decode(testValue)); + break; + case "string": + message.writeUTF(testValue); + break; + default: + throw new Exception("Internal exception: Unexpected JMS message sub-type \"" + testValueType + "\""); + } + return message; + } + + protected MapMessage createJMSMapMessage(String testValueType, String testValue, int testValueNum) throws Exception, JMSException { + MapMessage message = _session.createMapMessage(); + String name = String.format("%s%03d", testValueType, testValueNum); + switch (testValueType) { + case "boolean": + message.setBoolean(name, Boolean.parseBoolean(testValue)); + break; + case "byte": + message.setByte(name, Byte.decode(testValue)); + break; + case "bytes": + message.setBytes(name, testValue.getBytes()); + break; + case "char": + if (testValue.length() == 1) { // Char format: "X" + message.setChar(name, testValue.charAt(0)); + } else if (testValue.length() == 6) { // Char format: "\xNNNN" + message.setChar(name, (char)Integer.parseInt(testValue.substring(2), 16)); + } else { + throw new Exception("JmsSenderShim.createJMSMapMessage() Malformed char string: \"" + testValue + "\""); + } + break; + case "double": + Long l1 = Long.parseLong(testValue.substring(2, 3), 16) << 60; + Long l2 = Long.parseLong(testValue.substring(3), 16); + message.setDouble(name, Double.longBitsToDouble(l1 | l2)); + break; + case "float": + Long i = Long.parseLong(testValue.substring(2), 16); + message.setFloat(name, Float.intBitsToFloat(i.intValue())); + break; + case "int": + message.setInt(name, Integer.decode(testValue)); + break; + case "long": + message.setLong(name, Long.decode(testValue)); + break; + case "object": + Object obj = (Object)createObject(testValue); + message.setObject(name, obj); + break; + case "short": + message.setShort(name, Short.decode(testValue)); + break; + case "string": + message.setString(name, testValue); + break; + default: + throw new Exception("Internal exception: Unexpected JMS message sub-type \"" + testValueType + "\""); + } + return message; + } + + protected ObjectMessage createJMSObjectMessage(String className, String testValue) throws Exception, JMSException { + Serializable obj = createJavaObject(className, testValue); + if (obj == null) { + // TODO: Handle error here + System.out.println("JmsSenderShim.createJMSObjectMessage: obj == null"); + return null; + } + ObjectMessage message = _session.createObjectMessage(); + message.setObject(obj); + return message; + } + + protected StreamMessage createJMSStreamMessage(String testValueType, String testValue) throws Exception, JMSException { + StreamMessage message = _session.createStreamMessage(); + switch (testValueType) { + case "boolean": + message.writeBoolean(Boolean.parseBoolean(testValue)); + break; + case "byte": + message.writeByte(Byte.decode(testValue)); + break; + case "bytes": + message.writeBytes(testValue.getBytes()); + break; + case "char": + if (testValue.length() == 1) { // Char format: "X" + message.writeChar(testValue.charAt(0)); + } else if (testValue.length() == 6) { // Char format: "\xNNNN" + message.writeChar((char)Integer.parseInt(testValue.substring(2), 16)); + } else { + throw new Exception("JmsSenderShim.createJMSStreamMessage() Malformed char string: \"" + testValue + "\""); + } + break; + case "double": + Long l1 = Long.parseLong(testValue.substring(2, 3), 16) << 60; + Long l2 = Long.parseLong(testValue.substring(3), 16); + message.writeDouble(Double.longBitsToDouble(l1 | l2)); + break; + case "float": + Long i = Long.parseLong(testValue.substring(2), 16); + message.writeFloat(Float.intBitsToFloat(i.intValue())); + break; + case "int": + message.writeInt(Integer.decode(testValue)); + break; + case "long": + message.writeLong(Long.decode(testValue)); + break; + case "object": + Object obj = (Object)createObject(testValue); + message.writeObject(obj); + break; + case "short": + message.writeShort(Short.decode(testValue)); + break; + case "string": + message.writeString(testValue); + break; + default: + throw new Exception("JmsSenderShim.createJMSStreamMessage(): Unexpected JMS message sub-type \"" + testValueType + "\""); + } + return message; + } + + protected static Serializable createJavaObject(String className, String testValue) throws Exception { + Serializable obj = null; + try { + Class<?> c = Class.forName(className); + if (className.compareTo("java.lang.Character") == 0) { + Constructor ctor = c.getConstructor(char.class); + if (testValue.length() == 1) { + // Use first character of string + obj = (Serializable)ctor.newInstance(testValue.charAt(0)); + } else if (testValue.length() == 4 || testValue.length() == 6) { + // Format '\xNN' or '\xNNNN' + obj = (Serializable)ctor.newInstance((char)Integer.parseInt(testValue.substring(2), 16)); + } else { + throw new Exception("JmsSenderShim.createJavaObject(): Malformed char string: \"" + testValue + "\""); + } + } else { + // Use string constructor + Constructor ctor = c.getConstructor(String.class); + obj = (Serializable)ctor.newInstance(testValue); + } + } + catch (ClassNotFoundException e) { + e.printStackTrace(System.out); + } + catch (NoSuchMethodException e) { + e.printStackTrace(System.out); + } + catch (InstantiationException e) { + e.printStackTrace(System.out); + } + catch (IllegalAccessException e) { + e.printStackTrace(System.out); + } + catch (InvocationTargetException e) { + e.printStackTrace(System.out); + } + return obj; + } + + // value has format "classname:ctorstrvalue" + protected static Serializable createObject(String value) throws Exception { + Serializable obj = null; + int colonIndex = value.indexOf(":"); + if (colonIndex >= 0) { + String className = value.substring(0, colonIndex); + String testValue = value.substring(colonIndex+1); + obj = createJavaObject(className, testValue); + } else { + throw new Exception("createObject(): Malformed value string"); + } + return obj; + } + + protected TextMessage createTextMessage(String valueStr) throws JMSException { + return _session.createTextMessage(valueStr); + } + + protected static boolean isSupportedJmsMessageType(String jmsMessageType) { + for (String supportedJmsMessageType: SUPPORTED_JMS_MESSAGE_TYPES) { + if (jmsMessageType.equals(supportedJmsMessageType)) + return true; + } + return false; + } + + private static class MyExceptionListener implements ExceptionListener { + @Override + public void onException(JMSException exception) { + System.out.println("Connection ExceptionListener fired, exiting."); + exception.printStackTrace(System.out); + System.exit(1); + } + } +} \ No newline at end of file --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
