You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2016/08/31 21:37:06 UTC
[2/2] qpid-interop-test git commit: QPIDIT-33: JMS headers added to
tests;
JMS properties still to be completed when proton C++ API issues are fixed.
QPIDIT-33: JMS headers added to tests; JMS properties still to be completed when proton C++ API issues are fixed.
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/8876c8fc
Tree: http://git-wip-us.apache.org/repos/asf/qpid-interop-test/tree/8876c8fc
Diff: http://git-wip-us.apache.org/repos/asf/qpid-interop-test/diff/8876c8fc
Branch: refs/heads/master
Commit: 8876c8fcbfe2af76bbb40e6ad029b0322374205c
Parents: 3545a9f
Author: Kim van der Riet <kp...@apache.org>
Authored: Wed Aug 31 17:36:27 2016 -0400
Committer: Kim van der Riet <kp...@apache.org>
Committed: Wed Aug 31 17:36:27 2016 -0400
----------------------------------------------------------------------
.../qpid_interop_test/shim/JmsReceiverShim.java | 534 ++++++++++++-------
.../qpid_interop_test/shim/JmsSenderShim.java | 257 ++++++---
shims/qpid-proton-cpp/CMakeLists.txt | 3 +-
shims/qpid-proton-cpp/src/CMakeLists.txt | 7 +-
.../qpid-proton-cpp/src/qpidit/QpidItErrors.cpp | 25 +
.../qpid-proton-cpp/src/qpidit/QpidItErrors.hpp | 21 +
.../src/qpidit/shim/AmqpReceiver.cpp | 19 +-
.../src/qpidit/shim/AmqpReceiver.hpp | 5 +-
.../src/qpidit/shim/AmqpSender.cpp | 17 +-
.../src/qpidit/shim/AmqpSender.hpp | 3 +-
.../src/qpidit/shim/JmsDefinitions.hpp | 46 ++
.../src/qpidit/shim/JmsReceiver.cpp | 131 ++++-
.../src/qpidit/shim/JmsReceiver.hpp | 23 +-
.../src/qpidit/shim/JmsSender.cpp | 141 ++++-
.../src/qpidit/shim/JmsSender.hpp | 25 +-
shims/qpid-proton-python/src/JmsReceiverShim.py | 73 ++-
shims/qpid-proton-python/src/JmsSenderShim.py | 95 +++-
src/python/qpid-interop-test/.gitignore | 2 +
.../qpid-interop-test/jms/jms_message_tests.py | 266 +++++++--
src/python/qpid-interop-test/test_type_map.py | 7 +
.../types/simple_type_tests.py | 12 +-
21 files changed, 1354 insertions(+), 358 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/shim/JmsReceiverShim.java
----------------------------------------------------------------------
diff --git a/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/shim/JmsReceiverShim.java b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/shim/JmsReceiverShim.java
index 1c257de..bd73492 100644
--- a/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/shim/JmsReceiverShim.java
+++ b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/shim/JmsReceiverShim.java
@@ -17,16 +17,19 @@
package org.apache.qpid.interop_test.shim;
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;
@@ -37,7 +40,9 @@ 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;
@@ -55,15 +60,24 @@ public class JmsReceiverShim {
"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 number map
+ // 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: Insufficient number of arguments");
- System.out.println("JmsReceiverShim: Expected arguments: broker_address, queue_name, amqp_type, num_test_values");
+ 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];
@@ -75,221 +89,379 @@ public class JmsReceiverShim {
}
JsonReader jsonReader = Json.createReader(new StringReader(args[3]));
- JsonObject numTestValuesMap = jsonReader.readObject();
+ JsonArray testParamsList = jsonReader.readArray();
jsonReader.close();
- Connection connection = null;
+ 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);
+
+ JmsReceiverShim shim = new JmsReceiverShim(brokerAddress, queueName);
+ shim.run(jmsMessageType, numTestValuesMap, flagMap);
+ }
+ public JmsReceiverShim(String brokerAddress, String queueName) {
try {
+ _connection = null;
ConnectionFactory factory = (ConnectionFactory)new JmsConnectionFactory(brokerAddress);
+ _connection = factory.createConnection(USER, PASSWORD);
+ _connection.setExceptionListener(new MyExceptionListener());
+ _connection.start();
- connection = factory.createConnection(USER, PASSWORD);
- connection.setExceptionListener(new MyExceptionListener());
- connection.start();
+ _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ _queue = _session.createQueue(queueName);
- Queue queue = session.createQueue(queueName);
+ _messageConsumer = _session.createConsumer(_queue);
- MessageConsumer 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);
- List<String> keyList = new ArrayList<String>(numTestValuesMap.keySet());
- Collections.sort(keyList);
-
Message message = null;
- JsonObjectBuilder job = Json.createObjectBuilder();
- for (String key: keyList) {
- JsonArrayBuilder jab = Json.createArrayBuilder();
- for (int i=0; i<numTestValuesMap.getJsonNumber(key).intValue(); ++i) {
- message = messageConsumer.receive(TIMEOUT);
+
+ 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":
- jab.addNull();
+ processJMSMessage(jasonTestValuesArrayBuilder);
break;
case "JMS_BYTESMESSAGE_TYPE":
- switch (key) {
- case "boolean":
- jab.add(((BytesMessage)message).readBoolean()?"True":"False");
- break;
- case "byte":
- jab.add(formatByte(((BytesMessage)message).readByte()));
- break;
- case "bytes":
- {
- byte[] bytesBuff = new byte[65536];
- int numBytesRead = ((BytesMessage)message).readBytes(bytesBuff);
- if (numBytesRead >= 0) {
- jab.add(new String(Arrays.copyOfRange(bytesBuff, 0, numBytesRead)));
- } else {
- // NOTE: For this case, an empty byte array has nothing to return
- jab.add(new String());
- }
- }
- break;
- case "char":
- jab.add(formatChar(((BytesMessage)message).readChar()));
- break;
- case "double":
- long l = Double.doubleToRawLongBits(((BytesMessage)message).readDouble());
- jab.add(String.format("0x%16s", Long.toHexString(l)).replace(' ', '0'));
- break;
- case "float":
- int i0 = Float.floatToRawIntBits(((BytesMessage)message).readFloat());
- jab.add(String.format("0x%8s", Integer.toHexString(i0)).replace(' ', '0'));
- break;
- case "int":
- jab.add(formatInt(((BytesMessage)message).readInt()));
- break;
- case "long":
- jab.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();
- jab.add(obj.getClass().getName() + ":" + obj.toString());
- } else {
- jab.add("<object error>");
- }
- }
- break;
- case "short":
- jab.add(formatShort(((BytesMessage)message).readShort()));
- break;
- case "string":
- jab.add(((BytesMessage)message).readUTF());
- break;
- default:
- throw new Exception("JmsReceiverShim: Unknown subtype for " + jmsMessageType + ": \"" + key + "\"");
- }
+ processJMSBytesMessage(jmsMessageType, subType, message, jasonTestValuesArrayBuilder);
break;
case "JMS_STREAMMESSAGE_TYPE":
- switch (key) {
- case "boolean":
- jab.add(((StreamMessage)message).readBoolean()?"True":"False");
- break;
- case "byte":
- jab.add(formatByte(((StreamMessage)message).readByte()));
- break;
- case "bytes":
- byte[] bytesBuff = new byte[65536];
- int numBytesRead = ((StreamMessage)message).readBytes(bytesBuff);
- if (numBytesRead >= 0) {
- jab.add(new String(Arrays.copyOfRange(bytesBuff, 0, numBytesRead)));
- } else {
- System.out.println("StreamMessage.readBytes() returned " + numBytesRead);
- jab.add("<bytes error>");
- }
- break;
- case "char":
- jab.add(formatChar(((StreamMessage)message).readChar()));
- break;
- case "double":
- long l = Double.doubleToRawLongBits(((StreamMessage)message).readDouble());
- jab.add(String.format("0x%16s", Long.toHexString(l)).replace(' ', '0'));
- break;
- case "float":
- int i0 = Float.floatToRawIntBits(((StreamMessage)message).readFloat());
- jab.add(String.format("0x%8s", Integer.toHexString(i0)).replace(' ', '0'));
- break;
- case "int":
- jab.add(formatInt(((StreamMessage)message).readInt()));
- break;
- case "long":
- jab.add(formatLong(((StreamMessage)message).readLong()));
- break;
- case "object":
- Object obj = ((StreamMessage)message).readObject();
- jab.add(obj.getClass().getName() + ":" + obj.toString());
- break;
- case "short":
- jab.add(formatShort(((StreamMessage)message).readShort()));
- break;
- case "string":
- jab.add(((StreamMessage)message).readString());
- break;
- default:
- throw new Exception("JmsReceiverShim: Unknown subtype for " + jmsMessageType + ": \"" + key + "\"");
- }
+ processJMSStreamMessage(jmsMessageType, subType, message, jasonTestValuesArrayBuilder);
break;
case "JMS_MAPMESSAGE_TYPE":
- String name = String.format("%s%03d", key, i);
- switch (key) {
- case "boolean":
- jab.add(((MapMessage)message).getBoolean(name)?"True":"False");
- break;
- case "byte":
- jab.add(formatByte(((MapMessage)message).getByte(name)));
- break;
- case "bytes":
- jab.add(new String(((MapMessage)message).getBytes(name)));
- break;
- case "char":
- jab.add(formatChar(((MapMessage)message).getChar(name)));
- break;
- case "double":
- long l = Double.doubleToRawLongBits(((MapMessage)message).getDouble(name));
- jab.add(String.format("0x%16s", Long.toHexString(l)).replace(' ', '0'));
- break;
- case "float":
- int i0 = Float.floatToRawIntBits(((MapMessage)message).getFloat(name));
- jab.add(String.format("0x%8s", Integer.toHexString(i0)).replace(' ', '0'));
- break;
- case "int":
- jab.add(formatInt(((MapMessage)message).getInt(name)));
- break;
- case "long":
- jab.add(formatLong(((MapMessage)message).getLong(name)));
- break;
- case "object":
- Object obj = ((MapMessage)message).getObject(name);
- jab.add(obj.getClass().getName() + ":" + obj.toString());
- break;
- case "short":
- jab.add(formatShort(((MapMessage)message).getShort(name)));
- break;
- case "string":
- jab.add(((MapMessage)message).getString(name));
- break;
- default:
- throw new Exception("JmsReceiverShim: Unknown subtype for " + jmsMessageType + ": \"" + key + "\"");
- }
+ processJMSMapMessage(jmsMessageType, subType, i, message, jasonTestValuesArrayBuilder);
break;
case "JMS_OBJECTMESSAGE_TYPE":
- jab.add(((ObjectMessage)message).getObject().toString());
+ processJMSObjectMessage(subType, message, jasonTestValuesArrayBuilder);
break;
case "JMS_TEXTMESSAGE_TYPE":
- jab.add(((TextMessage)message).getText());
+ processJMSTextMessage(message, jasonTestValuesArrayBuilder);
break;
default:
- connection.close();
+ _connection.close();
throw new Exception("JmsReceiverShim: Internal error: Unknown or unsupported JMS message type \"" + jmsMessageType + "\"");
}
+
+ processMessageHeaders(message, flagMap);
+ processMessageProperties(message);
}
- job.add(key, jab);
+ _jsonTestValueMapBuilder.add(subType, jasonTestValuesArrayBuilder);
}
- connection.close();
-
+ _connection.close();
+
System.out.println(jmsMessageType);
StringWriter out = new StringWriter();
- JsonWriter jsonWriter = Json.createWriter(out);
- jsonWriter.writeObject(job.build());
- jsonWriter.close();
- System.out.println(out.toString());
+ writeJsonObject(_jsonTestValueMapBuilder, out);
+ out.append('\n');
+ writeJsonObject(_jsonMessageHeaderMapBuilder, out);
+ out.append('\n');
+ writeJsonObject(_jsonMessagePropertiesMapBuilder, out);
+ System.out.println(out.toString());
} catch (Exception exp) {
- if (connection != null)
- connection.close();
+ 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) {
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/shim/JmsSenderShim.java
----------------------------------------------------------------------
diff --git a/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/shim/JmsSenderShim.java b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/shim/JmsSenderShim.java
index 2c6281e..78bedfc 100644
--- a/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/shim/JmsSenderShim.java
+++ b/shims/qpid-jms/src/main/java/org/apache/qpid/qpid_interop_test/shim/JmsSenderShim.java
@@ -52,78 +52,62 @@ public class JmsSenderShim {
"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 value map
+ // 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: Insufficient number of arguments");
- System.out.println("JmsSenderShim: Expected arguments: broker_address, queue_name, amqp_type, test_val, test_val, ...");
+ 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.out.println("ERROR: JmsSender: Unknown or unsupported JMS message type \"" + jmsMessageType + "\"");
+ System.err.println("ERROR: JmsSender: Unknown or unsupported JMS message type \"" + jmsMessageType + "\"");
System.exit(1);
}
JsonReader jsonReader = Json.createReader(new StringReader(args[3]));
- JsonObject testValuesMap = jsonReader.readObject();
+ 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.exit(1);
+ }
+ JsonObject testValuesMap = testParamsList.getJsonObject(0);
+ JsonObject testHeadersMap = testParamsList.getJsonObject(1);
+ JsonObject testPropertiesMap = testParamsList.getJsonObject(2);
+
+ JmsSenderShim shim = new JmsSenderShim(brokerAddress, queueName);
+ shim.runTests(jmsMessageType, testValuesMap, testHeadersMap, testPropertiesMap);
+ }
+
+ public JmsSenderShim(String brokerAddress, String queueName) {
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);
+ _connection = factory.createConnection();
+ _connection.setExceptionListener(new MyExceptionListener());
+ _connection.start();
- Queue queue = session.createQueue(queueName);
+ _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- MessageProducer messageProducer = session.createProducer(queue);
+ _queue = _session.createQueue(queueName);
- Message message = null;
- List<String> keyList = new ArrayList<String>(testValuesMap.keySet());
- Collections.sort(keyList);
- for (String key: keyList) {
- JsonArray testValues = testValuesMap.getJsonArray(key);
- for (int i=0; i<testValues.size(); ++i) {
- String testValue = "";
- if (!testValues.isNull(i)) {
- testValue = testValues.getJsonString(i).getString();
- }
- switch (jmsMessageType) {
- case "JMS_MESSAGE_TYPE":
- message = createMessage(session, key, testValue);
- break;
- case "JMS_BYTESMESSAGE_TYPE":
- message = createBytesMessage(session, key, testValue);
- break;
- case "JMS_MAPMESSAGE_TYPE":
- message = createMapMessage(session, key, testValue, i);
- break;
- case "JMS_OBJECTMESSAGE_TYPE":
- message = createObjectMessage(session, key, testValue);
- break;
- case "JMS_STREAMMESSAGE_TYPE":
- message = createStreamMessage(session, key, testValue);
- break;
- case "JMS_TEXTMESSAGE_TYPE":
- message = createTextMessage(session, testValue);
- break;
- default:
- throw new Exception("Internal exception: Unexpected JMS message type \"" + jmsMessageType + "\"");
- }
- messageProducer.send(message, DeliveryMode.NON_PERSISTENT, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE);
- }
- }
- connection.close();
+ _messageProducer = _session.createProducer(_queue);
+
+ _msgsSent = 0;
} catch (Exception exp) {
System.out.println("Caught exception, exiting.");
exp.printStackTrace(System.out);
@@ -131,18 +115,157 @@ public class JmsSenderShim {
}
}
- protected static Message createMessage(Session session, String testValueType, String testValue) throws Exception, JMSException {
+ 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();
+ return _session.createMessage();
}
- protected static BytesMessage createBytesMessage(Session session, String testValueType, String testValue) throws Exception, JMSException {
- BytesMessage message = session.createBytesMessage();
+ protected BytesMessage createJMSBytesMessage(String testValueType, String testValue) throws Exception, JMSException {
+ BytesMessage message = _session.createBytesMessage();
switch (testValueType) {
case "boolean":
message.writeBoolean(Boolean.parseBoolean(testValue));
@@ -157,7 +280,7 @@ public class JmsSenderShim {
if (testValue.length() == 1) { // Char format: "X" or "\xNN"
message.writeChar(testValue.charAt(0));
} else {
- throw new Exception("JmsSenderShim.createBytesMessage() Malformed char string: \"" + testValue + "\" of length " + testValue.length());
+ throw new Exception("JmsSenderShim.createJMSBytesMessage() Malformed char string: \"" + testValue + "\" of length " + testValue.length());
}
break;
case "double":
@@ -191,8 +314,8 @@ public class JmsSenderShim {
return message;
}
- protected static MapMessage createMapMessage(Session session, String testValueType, String testValue, int testValueNum) throws Exception, JMSException {
- MapMessage message = session.createMapMessage();
+ 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":
@@ -210,7 +333,7 @@ public class JmsSenderShim {
} else if (testValue.length() == 6) { // Char format: "\xNNNN"
message.setChar(name, (char)Integer.parseInt(testValue.substring(2), 16));
} else {
- throw new Exception("JmsSenderShim.createMapMessage() Malformed char string: \"" + testValue + "\"");
+ throw new Exception("JmsSenderShim.createJMSMapMessage() Malformed char string: \"" + testValue + "\"");
}
break;
case "double":
@@ -244,20 +367,20 @@ public class JmsSenderShim {
return message;
}
- protected static ObjectMessage createObjectMessage(Session session, String className, String testValue) throws Exception, JMSException {
+ 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("createObjectMessage: obj == null");
+ System.out.println("JmsSenderShim.createJMSObjectMessage: obj == null");
return null;
}
- ObjectMessage message = session.createObjectMessage();
+ ObjectMessage message = _session.createObjectMessage();
message.setObject(obj);
return message;
}
- protected static StreamMessage createStreamMessage(Session session, String testValueType, String testValue) throws Exception, JMSException {
- StreamMessage message = session.createStreamMessage();
+ protected StreamMessage createJMSStreamMessage(String testValueType, String testValue) throws Exception, JMSException {
+ StreamMessage message = _session.createStreamMessage();
switch (testValueType) {
case "boolean":
message.writeBoolean(Boolean.parseBoolean(testValue));
@@ -274,7 +397,7 @@ public class JmsSenderShim {
} else if (testValue.length() == 6) { // Char format: "\xNNNN"
message.writeChar((char)Integer.parseInt(testValue.substring(2), 16));
} else {
- throw new Exception("JmsSenderShim.createStreamMessage() Malformed char string: \"" + testValue + "\"");
+ throw new Exception("JmsSenderShim.createJMSStreamMessage() Malformed char string: \"" + testValue + "\"");
}
break;
case "double":
@@ -303,7 +426,7 @@ public class JmsSenderShim {
message.writeString(testValue);
break;
default:
- throw new Exception("Internal exception: Unexpected JMS message sub-type \"" + testValueType + "\"");
+ throw new Exception("JmsSenderShim.createJMSStreamMessage(): Unexpected JMS message sub-type \"" + testValueType + "\"");
}
return message;
}
@@ -321,7 +444,7 @@ public class JmsSenderShim {
// Format '\xNN' or '\xNNNN'
obj = (Serializable)ctor.newInstance((char)Integer.parseInt(testValue.substring(2), 16));
} else {
- throw new Exception("JmsSenderShim.createStreamMessage() Malformed char string: \"" + testValue + "\"");
+ throw new Exception("JmsSenderShim.createJavaObject(): Malformed char string: \"" + testValue + "\"");
}
} else {
// Use string constructor
@@ -361,8 +484,8 @@ public class JmsSenderShim {
return obj;
}
- protected static TextMessage createTextMessage(Session session, String valueStr) throws JMSException {
- return session.createTextMessage(valueStr);
+ protected TextMessage createTextMessage(String valueStr) throws JMSException {
+ return _session.createTextMessage(valueStr);
}
protected static boolean isSupportedJmsMessageType(String jmsMessageType) {
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-proton-cpp/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/shims/qpid-proton-cpp/CMakeLists.txt b/shims/qpid-proton-cpp/CMakeLists.txt
index e8c7e89..fb00a9d 100644
--- a/shims/qpid-proton-cpp/CMakeLists.txt
+++ b/shims/qpid-proton-cpp/CMakeLists.txt
@@ -21,8 +21,7 @@ project (qpid-interop-test-cpp-shims)
cmake_minimum_required(VERSION 2.6)
-set(PROTON_DIR "$ENV{HOME}/qpid-proton" CACHE PATH "Proton directory")
-set(PROTON_BUILD_DIR_NAME "build" CACHE STRING "Proton build directory")
+set(PROTON_INSTALL_DIR "/usr/local/qpid-proton" CACHE PATH "Proton install directory")
add_subdirectory(src)
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-proton-cpp/src/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/shims/qpid-proton-cpp/src/CMakeLists.txt b/shims/qpid-proton-cpp/src/CMakeLists.txt
index 724176a..06e62a0 100644
--- a/shims/qpid-proton-cpp/src/CMakeLists.txt
+++ b/shims/qpid-proton-cpp/src/CMakeLists.txt
@@ -17,12 +17,11 @@
# under the License.
#
-message("PROTON_DIR=${PROTON_DIR} PROTON_BUILD_DIR_NAME=${PROTON_BUILD_DIR_NAME}")
-include_directories(${PROTON_DIR}/proton-c/include)
-include_directories(${PROTON_DIR}/proton-c/bindings/cpp/include)
+message("PROTON_INSTALL_DIR=${PROTON_INSTALL_DIR}")
+include_directories(${PROTON_INSTALL_DIR}/include)
include_directories(/usr/include/jsoncpp)
include_directories(${CMAKE_CURRENT_SOURCE_DIR})
-link_directories(${PROTON_DIR}/${PROTON_BUILD_DIR_NAME}/proton-c/bindings/cpp)
+link_directories(${PROTON_INSTALL_DIR}/lib64)
#set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
# --- Common files and libs ---
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-proton-cpp/src/qpidit/QpidItErrors.cpp
----------------------------------------------------------------------
diff --git a/shims/qpid-proton-cpp/src/qpidit/QpidItErrors.cpp b/shims/qpid-proton-cpp/src/qpidit/QpidItErrors.cpp
index 4802787..9d7b821 100644
--- a/shims/qpid-proton-cpp/src/qpidit/QpidItErrors.cpp
+++ b/shims/qpid-proton-cpp/src/qpidit/QpidItErrors.cpp
@@ -172,6 +172,23 @@ namespace qpidit
UnknownAmqpTypeError::~UnknownAmqpTypeError() throw() {}
+ // --- UnknownJmsDestinationTypeError ---
+
+ UnknownJmsDestinationTypeError::UnknownJmsDestinationTypeError(const std::string& jmsDestinationType) :
+ std::runtime_error(MSG("Unknown JMS destination type \"" << jmsDestinationType << "\""))
+ {}
+
+ UnknownJmsDestinationTypeError::~UnknownJmsDestinationTypeError() throw() {}
+
+
+ // --- UnknownJmsHeaderTypeError ---
+
+ UnknownJmsHeaderTypeError::UnknownJmsHeaderTypeError(const std::string& jmsHeaderType) :
+ std::runtime_error(MSG("Unknown or incorrect JMS header type \"" << jmsHeaderType << "\""))
+ {}
+
+ UnknownJmsHeaderTypeError::~UnknownJmsHeaderTypeError() throw() {}
+
// --- UnknownJmsMessageSubTypeError ---
UnknownJmsMessageSubTypeError::UnknownJmsMessageSubTypeError(const std::string& jmsMessageSubType) :
@@ -190,6 +207,14 @@ namespace qpidit
UnknownJmsMessageTypeError::~UnknownJmsMessageTypeError() throw() {}
+ // --- UnknownJmsPropertyTypeError ---
+
+ UnknownJmsPropertyTypeError::UnknownJmsPropertyTypeError(const std::string& jmsPropertyType) :
+ std::runtime_error(MSG("Unknown JMS property type \"" << jmsPropertyType << "\""))
+ {}
+
+ UnknownJmsPropertyTypeError::~UnknownJmsPropertyTypeError() throw() {}
+
// --- UnsupportedAmqpTypeError ---
UnsupportedAmqpTypeError::UnsupportedAmqpTypeError(const std::string& amqpType) :
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-proton-cpp/src/qpidit/QpidItErrors.hpp
----------------------------------------------------------------------
diff --git a/shims/qpid-proton-cpp/src/qpidit/QpidItErrors.hpp b/shims/qpid-proton-cpp/src/qpidit/QpidItErrors.hpp
index 3632af6..45b0c22 100644
--- a/shims/qpid-proton-cpp/src/qpidit/QpidItErrors.hpp
+++ b/shims/qpid-proton-cpp/src/qpidit/QpidItErrors.hpp
@@ -140,6 +140,20 @@ namespace qpidit
virtual ~UnknownAmqpTypeError() throw();
};
+ class UnknownJmsDestinationTypeError: public std::runtime_error
+ {
+ public:
+ explicit UnknownJmsDestinationTypeError(const std::string& jmsDestinationType);
+ virtual ~UnknownJmsDestinationTypeError() throw();
+ };
+
+ class UnknownJmsHeaderTypeError: public std::runtime_error
+ {
+ public:
+ explicit UnknownJmsHeaderTypeError(const std::string& jmsHeaderType);
+ virtual ~UnknownJmsHeaderTypeError() throw();
+ };
+
class UnknownJmsMessageSubTypeError: public std::runtime_error
{
public:
@@ -154,6 +168,13 @@ namespace qpidit
virtual ~UnknownJmsMessageTypeError() throw();
};
+ class UnknownJmsPropertyTypeError: public std::runtime_error
+ {
+ public:
+ explicit UnknownJmsPropertyTypeError(const std::string& jmsPropertyType);
+ virtual ~UnknownJmsPropertyTypeError() throw();
+ };
+
class UnsupportedAmqpTypeError: public std::runtime_error
{
public:
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-proton-cpp/src/qpidit/shim/AmqpReceiver.cpp
----------------------------------------------------------------------
diff --git a/shims/qpid-proton-cpp/src/qpidit/shim/AmqpReceiver.cpp b/shims/qpid-proton-cpp/src/qpidit/shim/AmqpReceiver.cpp
index 0d4a8c0..7680a77 100644
--- a/shims/qpid-proton-cpp/src/qpidit/shim/AmqpReceiver.cpp
+++ b/shims/qpid-proton-cpp/src/qpidit/shim/AmqpReceiver.cpp
@@ -27,6 +27,7 @@
#include <proton/default_container.hpp>
#include "proton/delivery.hpp"
#include "proton/receiver.hpp"
+#include "proton/transport.hpp"
#include "qpidit/QpidItErrors.hpp"
namespace qpidit
@@ -55,7 +56,7 @@ namespace qpidit
}
void AmqpReceiver::on_message(proton::delivery &d, proton::message &m) {
- if (m.id().get<uint64_t>() < _received) return; // ignore duplicate
+ if (proton::get<uint64_t>(m.id()) < _received) return; // ignore duplicate
if (_received < _expected) {
if (_amqpType.compare("null") == 0) {
checkMessageType(m, proton::NULL_TYPE);
@@ -155,19 +156,23 @@ namespace qpidit
}
void AmqpReceiver::on_connection_error(proton::connection &c) {
- std::cerr << "AmqpReceiver:on_connection_error()" << std::endl;
+ std::cerr << "AmqpReceiver::on_connection_error(): " << c.error() << std::endl;
}
- void AmqpReceiver::on_sender_error(proton::sender& l) {
- std::cerr << "AmqpReceiver:on_sender_error()" << std::endl;
+ void AmqpReceiver::on_receiver_error(proton::receiver& r) {
+ std::cerr << "AmqpReceiver::on_receiver_error(): " << r.error() << std::endl;
+ }
+
+ void AmqpReceiver::on_session_error(proton::session &s) {
+ std::cerr << "AmqpReceiver::on_session_error(): " << s.error() << std::endl;
}
void AmqpReceiver::on_transport_error(proton::transport &t) {
- std::cerr << "AmqpReceiver:on_transport_error()" << std::endl;
+ std::cerr << "AmqpReceiver::on_transport_error(): " << t.error() << std::endl;
}
- void AmqpReceiver::on_unhandled_error(const proton::error_condition &c) {
- std::cerr << "AmqpReceiver:on_unhandled_error() name=" << c.name() << " description=" << c.description() << std::endl;
+ void AmqpReceiver::on_error(const proton::error_condition &ec) {
+ std::cerr << "AmqpReceiver::on_error(): " << ec << std::endl;
}
// protected
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-proton-cpp/src/qpidit/shim/AmqpReceiver.hpp
----------------------------------------------------------------------
diff --git a/shims/qpid-proton-cpp/src/qpidit/shim/AmqpReceiver.hpp b/shims/qpid-proton-cpp/src/qpidit/shim/AmqpReceiver.hpp
index 64909f9..fdd5287 100644
--- a/shims/qpid-proton-cpp/src/qpidit/shim/AmqpReceiver.hpp
+++ b/shims/qpid-proton-cpp/src/qpidit/shim/AmqpReceiver.hpp
@@ -49,9 +49,10 @@ namespace qpidit
void on_message(proton::delivery &d, proton::message &m);
void on_connection_error(proton::connection &c);
- void on_sender_error(proton::sender& l);
+ void on_receiver_error(proton::receiver& r);
+ void on_session_error(proton::session &s);
void on_transport_error(proton::transport &t);
- void on_unhandled_error(const proton::error_condition &c);
+ void on_error(const proton::error_condition &c);
protected:
static void checkMessageType(const proton::message& msg, proton::type_id msgType);
static Json::Value& getMap(Json::Value& jsonMap, const proton::value& val);
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-proton-cpp/src/qpidit/shim/AmqpSender.cpp
----------------------------------------------------------------------
diff --git a/shims/qpid-proton-cpp/src/qpidit/shim/AmqpSender.cpp b/shims/qpid-proton-cpp/src/qpidit/shim/AmqpSender.cpp
index b256551..a81e235 100644
--- a/shims/qpid-proton-cpp/src/qpidit/shim/AmqpSender.cpp
+++ b/shims/qpid-proton-cpp/src/qpidit/shim/AmqpSender.cpp
@@ -27,6 +27,7 @@
#include "proton/default_container.hpp"
#include "proton/sender.hpp"
#include "proton/tracker.hpp"
+#include "proton/transport.hpp"
namespace qpidit
{
@@ -78,19 +79,23 @@ namespace qpidit
}
void AmqpSender::on_connection_error(proton::connection &c) {
- std::cerr << "AmqpSender:on_connection_error()" << std::endl;
+ std::cerr << "AmqpSender::on_connection_error(): " << c.error() << std::endl;
}
- void AmqpSender::on_sender_error(proton::sender& l) {
- std::cerr << "AmqpSender:on_sender_error()" << std::endl;
+ void AmqpSender::on_sender_error(proton::sender &s) {
+ std::cerr << "AmqpSender::on_sender_error(): " << s.error() << std::endl;
+ }
+
+ void AmqpSender::on_session_error(proton::session &s) {
+ std::cerr << "AmqpSender::on_session_error(): " << s.error() << std::endl;
}
void AmqpSender::on_transport_error(proton::transport &t) {
- std::cerr << "AmqpSender:on_transport_error()" << std::endl;
+ std::cerr << "AmqpSender::on_transport_error(): " << t.error() << std::endl;
}
- void AmqpSender::on_unhandled_error(const proton::error_condition &c) {
- std::cerr << "AmqpSender:on_unhandled_error()" << " name=" << c.name() << " description=" << c.description() << std::endl;
+ void AmqpSender::on_error(const proton::error_condition &ec) {
+ std::cerr << "AmqpSender::on_error(): " << ec << std::endl;
}
// protected
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-proton-cpp/src/qpidit/shim/AmqpSender.hpp
----------------------------------------------------------------------
diff --git a/shims/qpid-proton-cpp/src/qpidit/shim/AmqpSender.hpp b/shims/qpid-proton-cpp/src/qpidit/shim/AmqpSender.hpp
index 0d4a87a..8ad4e52 100644
--- a/shims/qpid-proton-cpp/src/qpidit/shim/AmqpSender.hpp
+++ b/shims/qpid-proton-cpp/src/qpidit/shim/AmqpSender.hpp
@@ -51,9 +51,10 @@ namespace qpidit
void on_transport_close(proton::transport &t);
void on_connection_error(proton::connection &c);
+ void on_session_error(proton::session &s);
void on_sender_error(proton::sender& s);
void on_transport_error(proton::transport &t);
- void on_unhandled_error(const proton::error_condition &c);
+ void on_error(const proton::error_condition &c);
protected:
proton::message& setMessage(proton::message& msg, const Json::Value& testValue);
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-proton-cpp/src/qpidit/shim/JmsDefinitions.hpp
----------------------------------------------------------------------
diff --git a/shims/qpid-proton-cpp/src/qpidit/shim/JmsDefinitions.hpp b/shims/qpid-proton-cpp/src/qpidit/shim/JmsDefinitions.hpp
new file mode 100644
index 0000000..0a1f8cd
--- /dev/null
+++ b/shims/qpid-proton-cpp/src/qpidit/shim/JmsDefinitions.hpp
@@ -0,0 +1,46 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#ifndef SRC_QPIDIT_SHIM_JMSDEFINTIONS_HPP_
+#define SRC_QPIDIT_SHIM_JMSDEFINTIONS_HPP_
+
+namespace qpidit
+{
+ namespace shim
+ {
+ typedef enum {JMS_QUEUE = 0,
+ JMS_TOPIC,
+ JMS_TMEP_QUEUE,
+ JMS_TEMP_TOPIC}
+ jmsDestinationType_t;
+
+ typedef enum {JMS_MESSAGE_TYPE=0,
+ JMS_OBJECTMESSAGE_TYPE,
+ JMS_MAPMESSAGE_TYPE,
+ JMS_BYTESMESSAGE_TYPE,
+ JMS_STREAMMESSAGE_TYPE,
+ JMS_TEXTMESSAGE_TYPE}
+ jmsMessageType_t;
+
+ }
+}
+
+#endif /* SRC_QPIDIT_SHIM_JMSDEFINTIONS_HPP_ */
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-proton-cpp/src/qpidit/shim/JmsReceiver.cpp
----------------------------------------------------------------------
diff --git a/shims/qpid-proton-cpp/src/qpidit/shim/JmsReceiver.cpp b/shims/qpid-proton-cpp/src/qpidit/shim/JmsReceiver.cpp
index e5c41d9..c1396a4 100644
--- a/shims/qpid-proton-cpp/src/qpidit/shim/JmsReceiver.cpp
+++ b/shims/qpid-proton-cpp/src/qpidit/shim/JmsReceiver.cpp
@@ -26,13 +26,13 @@
#include "proton/connection.hpp"
#include "proton/default_container.hpp"
#include "proton/delivery.hpp"
+#include "proton/transport.hpp"
#include "qpidit/QpidItErrors.hpp"
namespace qpidit
{
namespace shim
{
- typedef enum {JMS_MESSAGE_TYPE=0, JMS_OBJECTMESSAGE_TYPE, JMS_MAPMESSAGE_TYPE, JMS_BYTESMESSAGE_TYPE, JMS_STREAMMESSAGE_TYPE, JMS_TEXTMESSAGE_TYPE} jmsMessageType_t;
//static
proton::symbol JmsReceiver::s_jmsMessageTypeAnnotationKey("x-opt-jms-msg-type");
std::map<std::string, int8_t>JmsReceiver::s_jmsMessageTypeAnnotationValues = initializeJmsMessageTypeAnnotationMap();
@@ -40,16 +40,20 @@ namespace qpidit
JmsReceiver::JmsReceiver(const std::string& brokerUrl,
const std::string& jmsMessageType,
- const Json::Value& testNumberMap):
+ const Json::Value& testNumberMap,
+ const Json::Value& flagMap):
_brokerUrl(brokerUrl),
_jmsMessageType(jmsMessageType),
_testNumberMap(testNumberMap),
+ _flagMap(flagMap),
_subTypeList(testNumberMap.getMemberNames()),
_subTypeIndex(0),
_expected(getTotalNumExpectedMsgs(testNumberMap)),
_received(0UL),
_receivedSubTypeList(Json::arrayValue),
- _receivedValueMap(Json::objectValue)
+ _receivedValueMap(Json::objectValue),
+ _receivedHeadersMap(Json::objectValue),
+ _receivedPropertiesMap(Json::objectValue)
{}
JmsReceiver::~JmsReceiver() {}
@@ -58,6 +62,14 @@ namespace qpidit
return _receivedValueMap;
}
+ Json::Value& JmsReceiver::getReceivedHeadersMap() {
+ return _receivedHeadersMap;
+ }
+
+ Json::Value& JmsReceiver::getReceivedPropertiesMap() {
+ return _receivedPropertiesMap;
+ }
+
void JmsReceiver::on_container_start(proton::container &c) {
c.open_receiver(_brokerUrl);
}
@@ -86,6 +98,10 @@ namespace qpidit
default:;
// TODO: handle error - no known JMS message type
}
+
+ processMessageHeaders(m);
+ processMessageProperties(m);
+
std::string subType(_subTypeList[_subTypeIndex]);
// Increment the subtype if the required number of messages have been received
if (_receivedSubTypeList.size() >= _testNumberMap[subType].asInt() &&
@@ -102,6 +118,26 @@ namespace qpidit
}
}
+ void JmsReceiver::on_connection_error(proton::connection &c) {
+ std::cerr << "JmsReceiver::on_connection_error(): " << c.error() << std::endl;
+ }
+
+ void JmsReceiver::on_receiver_error(proton::receiver& r) {
+ std::cerr << "JmsReceiver::on_receiver_error(): " << r.error() << std::endl;
+ }
+
+ void JmsReceiver::on_session_error(proton::session &s) {
+ std::cerr << "JmsReceiver::on_session_error(): " << s.error() << std::endl;
+ }
+
+ void JmsReceiver::on_transport_error(proton::transport &t) {
+ std::cerr << "JmsReceiver::on_transport_error(): " << t.error() << std::endl;
+ }
+
+ void JmsReceiver::on_error(const proton::error_condition &ec) {
+ std::cerr << "JmsReceiver::on_error(): " << ec << std::endl;
+ }
+
//static
uint32_t JmsReceiver::getTotalNumExpectedMsgs(const Json::Value testNumberMap) {
uint32_t total(0UL);
@@ -261,6 +297,72 @@ namespace qpidit
_receivedSubTypeList.append(Json::Value(msg.body().get<std::string>()));
}
+ void JmsReceiver::processMessageHeaders(const proton::message& msg) {
+ addMessageHeaderString("JMS_TYPE_HEADER", msg.subject());
+ if (_flagMap.isMember("JMS_CORRELATIONID_AS_BYTES") && _flagMap["JMS_CORRELATIONID_AS_BYTES"].asBool()) {
+ addMessageHeaderByteArray("JMS_CORRELATIONID_HEADER", proton::get<proton::binary>(msg.correlation_id()));
+ } else {
+ try {
+ addMessageHeaderString("JMS_CORRELATIONID_HEADER", proton::get<std::string>(msg.correlation_id()));
+ } catch (const std::exception& e) {} // TODO: UGLY, how do you check if there _is_ a correlation id?
+ }
+
+ std::string reply_to = msg.reply_to();
+ // Some brokers prepend 'queue://' and 'topic://' to reply_to addresses, strip these when present
+ if (_flagMap.isMember("JMS_REPLYTO_AS_TOPIC") && _flagMap["JMS_REPLYTO_AS_TOPIC"].asBool()) {
+ if (reply_to.find("topic://") == 0) {
+ addMessageHeaderDestination("JMS_REPLYTO_HEADER", JMS_TOPIC, reply_to.substr(8));
+ } else {
+ addMessageHeaderDestination("JMS_REPLYTO_HEADER", JMS_TOPIC, reply_to);
+ }
+ } else {
+ if (reply_to.find("queue://") == 0) {
+ addMessageHeaderDestination("JMS_REPLYTO_HEADER", JMS_QUEUE, reply_to.substr(8));
+ } else {
+ addMessageHeaderDestination("JMS_REPLYTO_HEADER", JMS_QUEUE, reply_to);
+ }
+ }
+ }
+
+ void JmsReceiver::addMessageHeaderString(const char* headerName, const std::string& value) {
+ if (!value.empty()) { // TODO: Remove this test when PROTON-1288 is fixed as empty strings are allowed in headers
+ Json::Value valueMap(Json::objectValue);
+ valueMap["string"] = value;
+ _receivedHeadersMap[headerName] = valueMap;
+ }
+ }
+
+ void JmsReceiver::addMessageHeaderByteArray(const std::string& headerName, const proton::binary ba) {
+ if (!ba.empty()) { // TODO: Remove this test when PROTON-1288 is fixed as empty binaries are allowed in headers
+ Json::Value valueMap(Json::objectValue);
+ valueMap["bytes"] = std::string(ba);
+ _receivedHeadersMap[headerName] = valueMap;
+ }
+ }
+
+ void JmsReceiver::addMessageHeaderDestination(const std::string& headerName, jmsDestinationType_t dt, const std::string& d) {
+ if (!d.empty()) {
+ Json::Value valueMap(Json::objectValue);
+ switch (dt) {
+ case JMS_QUEUE:
+ valueMap["queue"] = d;
+ break;
+ case JMS_TOPIC:
+ valueMap["topic"] = d;
+ break;
+ default:
+ ; // TODO: Handle error: remaining JMS destinations not handled.
+ }
+ _receivedHeadersMap[headerName] = valueMap;
+ }
+ }
+
+ void JmsReceiver::processMessageProperties(const proton::message& msg) {
+ // TODO: Add this function when PROTON-1284 is fixed
+// std::map<proton::value, proton::value> props;
+// msg.properties().value() >> props;
+ }
+
//static
std::map<std::string, int8_t> JmsReceiver::initializeJmsMessageTypeAnnotationMap() {
std::map<std::string, int8_t> m;
@@ -281,7 +383,7 @@ namespace qpidit
* Args: 1: Broker address (ip-addr:port)
* 2: Queue name
* 3: JMS message type
- * 4: JSON string of map containing number of test values to receive for each type/subtype
+ * 4: JSON Test parameters containing 2 maps: [testValuesMap, flagMap]
*/
int main(int argc, char** argv) {
/*
@@ -297,21 +399,24 @@ int main(int argc, char** argv) {
std::ostringstream oss;
oss << argv[1] << "/" << argv[2];
-// try {
- Json::Value testNumberMap;
+ try {
+ Json::Value testParams;
Json::Reader jsonReader;
- if (not jsonReader.parse(argv[4], testNumberMap, false)) {
+ if (not jsonReader.parse(argv[4], testParams, false)) {
throw qpidit::JsonParserError(jsonReader);
}
- qpidit::shim::JmsReceiver receiver(oss.str(), argv[3], testNumberMap);
+ qpidit::shim::JmsReceiver receiver(oss.str(), argv[3], testParams[0], testParams[1]);
proton::default_container(receiver).run();
+ Json::FastWriter fw;
std::cout << argv[3] << std::endl;
- std::cout << receiver.getReceivedValueMap();
-// } catch (const std::exception& e) {
-// std::cerr << "JmsReceiver error: " << e.what() << std::endl;
-// exit(1);
-// }
+ std::cout << fw.write(receiver.getReceivedValueMap());
+ std::cout << fw.write(receiver.getReceivedHeadersMap());
+ std::cout << fw.write(receiver.getReceivedPropertiesMap());
+ } catch (const std::exception& e) {
+ std::cerr << "JmsReceiver error: " << e.what() << std::endl;
+ exit(1);
+ }
exit(0);
}
http://git-wip-us.apache.org/repos/asf/qpid-interop-test/blob/8876c8fc/shims/qpid-proton-cpp/src/qpidit/shim/JmsReceiver.hpp
----------------------------------------------------------------------
diff --git a/shims/qpid-proton-cpp/src/qpidit/shim/JmsReceiver.hpp b/shims/qpid-proton-cpp/src/qpidit/shim/JmsReceiver.hpp
index ff12ac9..23c60fc 100644
--- a/shims/qpid-proton-cpp/src/qpidit/shim/JmsReceiver.hpp
+++ b/shims/qpid-proton-cpp/src/qpidit/shim/JmsReceiver.hpp
@@ -26,6 +26,7 @@
#include <json/value.h>
#include "proton/messaging_handler.hpp"
#include "proton/types.hpp"
+#include "qpidit/shim/JmsDefinitions.hpp"
#include <sstream>
namespace qpidit
@@ -42,19 +43,33 @@ namespace qpidit
const std::string _brokerUrl;
const std::string _jmsMessageType;
const Json::Value _testNumberMap;
+ const Json::Value _flagMap;
Json::Value::Members _subTypeList;
int _subTypeIndex;
uint32_t _expected;
uint32_t _received;
Json::Value _receivedSubTypeList;
Json::Value _receivedValueMap;
+ Json::Value _receivedHeadersMap;
+ Json::Value _receivedPropertiesMap;
public:
- JmsReceiver(const std::string& brokerUrl, const std::string& jmsMessageType, const Json::Value& testNumberMap);
+ JmsReceiver(const std::string& brokerUrl,
+ const std::string& jmsMessageType,
+ const Json::Value& testNumberMap,
+ const Json::Value& flagMap);
virtual ~JmsReceiver();
Json::Value& getReceivedValueMap();
+ Json::Value& getReceivedHeadersMap();
+ Json::Value& getReceivedPropertiesMap();
void on_container_start(proton::container &c);
void on_message(proton::delivery &d, proton::message &m);
+ void on_connection_error(proton::connection &c);
+ void on_receiver_error(proton::receiver& r);
+ void on_session_error(proton::session &s);
+ void on_transport_error(proton::transport &t);
+ void on_error(const proton::error_condition &c);
+
static uint32_t getTotalNumExpectedMsgs(const Json::Value testNumberMap);
protected:
@@ -65,6 +80,12 @@ namespace qpidit
void receiveJmsStreamMessage(const proton::message& msg);
void receiveJmsTextMessage(const proton::message& msg);
+ void processMessageHeaders(const proton::message& msg);
+ void addMessageHeaderString(const char* headerName, const std::string& value);
+ void addMessageHeaderByteArray(const std::string& headerName, const proton::binary ba);
+ void addMessageHeaderDestination(const std::string& headerName, jmsDestinationType_t dt, const std::string& d);
+ void processMessageProperties(const proton::message& msg);
+
static std::map<std::string, int8_t> initializeJmsMessageTypeAnnotationMap();
// Format signed numbers in negative hex format if signedFlag is true, ie -0xNNNN, positive numbers in 0xNNNN format
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org