DestinationListenerWrapper { protected void onDelivery(final CallbackService callbackService, final DeliveryRequest deliveryRequest, final QOS qos, final boolean autoConfirm) { final String methodName = "onDelivery"; logger.entry(this, methodName, callbackService, deliveryRequest, qos, autoConfirm); callbackService.run(new Runnable() { @Override public void run() { final String methodName = "run"; logger.entry(this, methodName); byte[] data = deliveryRequest.buf; deliveryRequest.buf = null; MalformedDelivery.MalformedReason malformedReason = null; String malformedDescription = null; String malformedMQMDFormat = null; int malformedMQMDCCSID = 0; byte[] payloadBytes = null; String payloadString = null; boolean payloadIsJson = false; org.apache.qpid.proton.message.Message msg = Proton.message(); try { msg.decode(data, 0, data.length); } catch(BufferOverflowException | BufferUnderflowException | DecodeException e) { malformedReason = MalformedDelivery.MalformedReason.PAYLOADNOTAMQP; malformedDescription = "The message could not be decoded because the message data is not a valid AMQP message"; payloadBytes = data; } Map<String, Object> properties = new HashMap<>(); if (malformedReason == null) { Object msgBodyValue = ((AmqpValue)msg.getBody()).getValue(); if (msgBodyValue instanceof Binary) { Binary binaryValue = (Binary)msgBodyValue; if ((binaryValue.getArrayOffset() == 0) && (binaryValue.getArray().length == binaryValue.getLength())) { payloadBytes = binaryValue.getArray(); } else { payloadBytes = new byte[binaryValue.getLength()]; System.arraycopy(binaryValue.getArray(), binaryValue.getArrayOffset(), payloadBytes, 0, binaryValue.getLength()); } } else if (msgBodyValue instanceof String) { payloadString = (String)msgBodyValue; payloadIsJson = "application/json".equalsIgnoreCase(msg.getContentType()); } else { malformedReason = MalformedDelivery.MalformedReason.FORMATNOMAPPING; malformedDescription = "The message payload uses an AMQP format that the MQ Light client cannot process"; payloadBytes = data; } if ((msg.getApplicationProperties() != null) && (msg.getApplicationProperties().getValue() != null)) { Map<?, ?> msgMap = msg.getApplicationProperties().getValue(); for (Map.Entry<?, ?> entry : msgMap.entrySet()) { if (entry.getKey() instanceof String) { Object value = entry.getValue(); if (value == null) { properties.put((String)entry.getKey(), null); } else if (value instanceof Binary) { properties.put((String)entry.getKey(), ((Binary)value).getArray()); } else { for (int i = 0; i < NonBlockingClientImpl.validPropertyValueTypes.length; ++i) { if (NonBlockingClientImpl.validPropertyValueTypes[i].isAssignableFrom(value.getClass())) { properties.put((String)entry.getKey(), value); } } } } } } } String parts[] = new SubscriptionTopic(deliveryRequest.topicPattern).split(); String shareName = parts[1]; String topicPattern = parts[0]; long ttl = 0; String topic = ""; if (malformedReason == null) { try { topic = URI.create(msg.getAddress()).getPath(); } catch(IllegalArgumentException e) { } if (topic == null) topic = ""; else if (topic.startsWith("/")) topic = topic.substring(1); ttl = msg.getTtl(); if (msg.getDeliveryAnnotations() != null) { Map<Symbol, Object> annotations = msg.getDeliveryAnnotations().getValue(); if (annotations.containsKey(malformedConditionSymbol) && annotations.get(malformedConditionSymbol) instanceof Symbol) { String condition = annotations.get(malformedConditionSymbol).toString(); if (condition.equals("FORMATNOMAPPING")) { malformedReason = MalformedDelivery.MalformedReason.FORMATNOMAPPING; } else if (condition.equals("JMSNOMAPPING")) { malformedReason = MalformedDelivery.MalformedReason.JMSNOMAPPING; } else if (condition.equals("PAYLOADENCODING")) { malformedReason = MalformedDelivery.MalformedReason.PAYLOADENCODING; } else if (condition.equals("PAYLOADNOTAMQP")) { malformedReason = MalformedDelivery.MalformedReason.PAYLOADNOTAMQP; } if (malformedReason != null && annotations.containsKey(malformedDescriptionSymbol) && annotations.get(malformedDescriptionSymbol) instanceof String) { malformedDescription = (String)annotations.get(malformedDescriptionSymbol); if (annotations.containsKey(malformedMQMDFormatSymbol) && annotations.get(malformedMQMDFormatSymbol) instanceof String) { malformedMQMDFormat = (String)annotations.get(malformedMQMDFormatSymbol); } if (annotations.containsKey(malformedMQMDCCSIDSymbol) && annotations.get(malformedMQMDCCSIDSymbol) instanceof Integer) { malformedMQMDCCSID = (Integer)annotations.get(malformedMQMDCCSIDSymbol); } } } } } if (payloadBytes != null) { if (malformedReason == null) { BytesDeliveryImpl delivery = new BytesDeliveryImpl(client, qos, shareName, topic, topicPattern, ttl, ByteBuffer.wrap(payloadBytes), properties, autoConfirm ? null : deliveryRequest); listener.onMessage(client, context, delivery); } else { MalformedDeliveryImpl delivery = new MalformedDeliveryImpl(client, qos, shareName, topic, topicPattern, ttl, ByteBuffer.wrap(payloadBytes), properties, autoConfirm ? null : deliveryRequest, malformedReason, malformedDescription, malformedMQMDFormat, malformedMQMDCCSID); listener.onMalformed(client, context, delivery); } } else { if (malformedReason == null) { Delivery delivery; if (payloadIsJson) { delivery = new JsonDeliveryImpl(client, qos, shareName, topic, topicPattern, ttl, payloadString, gsonBuilder, properties, autoConfirm ? null : deliveryRequest); } else { delivery = new StringDeliveryImpl(client, qos, shareName, topic, topicPattern, ttl, payloadString, properties, autoConfirm ? null : deliveryRequest); } listener.onMessage(client, context, delivery); } else { MalformedDeliveryImpl delivery = new MalformedDeliveryImpl(client, qos, shareName, topic, topicPattern, ttl, ByteBuffer.wrap(payloadString.getBytes(Charset.forName("UTF-8"))), properties, autoConfirm ? null : deliveryRequest, malformedReason, malformedDescription, malformedMQMDFormat, malformedMQMDCCSID); listener.onMalformed(client, context, delivery); } } if (autoConfirm) { client.doDelivery(deliveryRequest); } logger.exit(this, methodName); } }, client, new CallbackPromiseImpl(client, true)); logger.exit(this, methodName); } protected  DestinationListenerWrapper(NonBlockingClientImpl client, GsonBuilder gsonBuilder, DestinationListener<T> listener, T context);   }
@Test public void onDeliveryBytes() { StubClient expectedClient = new StubClient(); MockListener listener = new MockListener(MockListener.Method.ON_MESSAGE); Object expectedContext = new Object(); MockCallbackService callbackService = new MockCallbackService(); byte[] expectedData = new byte[] {3, 1, 4, 1, 5, 9}; String expectedTopic = "/topic1"; String expectedTopicPattern = "/#"; long expectedTtl = 12345; QOS expectedQos = QOS.AT_LEAST_ONCE; byte[] msgData = createSerializedProtonMessage(new AmqpValue(new Binary(expectedData)), expectedTopic, expectedTtl, null, null, null); DeliveryRequest request = new DeliveryRequest(msgData, expectedQos, "private:" + expectedTopicPattern, null, null); DestinationListenerWrapper<Object> wrapper = new DestinationListenerWrapper<Object>(expectedClient, new GsonBuilder(), listener, expectedContext); wrapper.onDelivery(callbackService, request, expectedQos, false); assertNotNull("Properties should not have been null!", listener.actualDelivery.getProperties()); assertEquals("Expected no message properties", 0, listener.actualDelivery.getProperties().size()); assertEquals("Expected QOS to match", expectedQos, listener.actualDelivery.getQOS()); assertEquals("Expected share to match", null, listener.actualDelivery.getShare()); assertEquals("Expected topic to match", expectedTopic, listener.actualDelivery.getTopic()); assertEquals("Expected topic pattern to match", expectedTopicPattern, listener.actualDelivery.getTopicPattern()); assertEquals("Expected ttl to match", expectedTtl, listener.actualDelivery.getTtl()); assertEquals("Expected delivery to be of type bytes", Delivery.Type.BYTES, listener.actualDelivery.getType()); BytesDelivery byteDelivery = (BytesDelivery)listener.actualDelivery; assertEquals("Expected same amount of delivery data", expectedData.length, byteDelivery.getData().remaining()); byte[] actualData = new byte[expectedData.length]; byteDelivery.getData().get(actualData); assertArrayEquals("Expected delivery data to match", expectedData, actualData); }
@Test public void onDeliveryString() { StubClient expectedClient = new StubClient(); MockListener listener = new MockListener(MockListener.Method.ON_MESSAGE); Object expectedContext = new Object(); MockCallbackService callbackService = new MockCallbackService(); String expectedData = "message data"; String expectedTopic = "/topic1"; String expectedTopicPattern = "/#"; long expectedTtl = 12345; QOS expectedQos = QOS.AT_LEAST_ONCE; byte[] msgData = createSerializedProtonMessage(new AmqpValue(expectedData), expectedTopic, expectedTtl, null, null, null); DeliveryRequest request = new DeliveryRequest(msgData, expectedQos, "private:" + expectedTopicPattern, null, null); DestinationListenerWrapper<Object> wrapper = new DestinationListenerWrapper<Object>(expectedClient, new GsonBuilder(), listener, expectedContext); wrapper.onDelivery(callbackService, request, expectedQos, false); assertNotNull("Properties should not have been null!", listener.actualDelivery.getProperties()); assertEquals("Expected no message properties", 0, listener.actualDelivery.getProperties().size()); assertEquals("Expected QOS to match", expectedQos, listener.actualDelivery.getQOS()); assertEquals("Expected share to match", null, listener.actualDelivery.getShare()); assertEquals("Expected topic to match", expectedTopic, listener.actualDelivery.getTopic()); assertEquals("Expected topic pattern to match", expectedTopicPattern, listener.actualDelivery.getTopicPattern()); assertEquals("Expected ttl to match", expectedTtl, listener.actualDelivery.getTtl()); assertEquals("Expected delivery to be of type string", Delivery.Type.STRING, listener.actualDelivery.getType()); StringDelivery stringDelivery = (StringDelivery)listener.actualDelivery; assertEquals("Expected data to match", expectedData, stringDelivery.getData()); }
@Test public void malformedBadAMQPMessageData() { StubClient expectedClient = new StubClient(); MockListener listener = new MockListener(MockListener.Method.ON_MALFORMED); Object expectedContext = new Object(); MockCallbackService callbackService = new MockCallbackService(); QOS expectedQos = QOS.AT_LEAST_ONCE; byte[] msgData = "I bet this isn't a valid AMQP message".getBytes(); DeliveryRequest request = new DeliveryRequest(msgData, expectedQos, "private:/malformed", null, null); DestinationListenerWrapper<Object> wrapper = new DestinationListenerWrapper<Object>(expectedClient, new GsonBuilder(), listener, expectedContext); wrapper.onDelivery(callbackService, request, expectedQos, false); assertEquals("Expected delivery to be of type string", Delivery.Type.MALFORMED, listener.actualDelivery.getType()); assertEquals("Expected malformed reason to be PAYLOADNOTAMQP", MalformedDelivery.MalformedReason.PAYLOADNOTAMQP, ((MalformedDelivery)listener.actualDelivery).getReason()); }
@Test public void onDeliveryJson() { StubClient expectedClient = new StubClient(); MockListener listener = new MockListener(MockListener.Method.ON_MESSAGE); Object expectedContext = new Object(); MockCallbackService callbackService = new MockCallbackService(); int[] expectedArray = new int[] {1, 2, 3}; String expectedData = new Gson().toJson(expectedArray); String expectedTopic = "/topic1"; String expectedTopicPattern = "/#"; long expectedTtl = 12345; QOS expectedQos = QOS.AT_LEAST_ONCE; byte[] msgData = createSerializedProtonMessage(new AmqpValue(expectedData), expectedTopic, expectedTtl, null, null, "application/json"); DeliveryRequest request = new DeliveryRequest(msgData, expectedQos, "private:" + expectedTopicPattern, null, null); DestinationListenerWrapper<Object> wrapper = new DestinationListenerWrapper<Object>(expectedClient, new GsonBuilder(), listener, expectedContext); wrapper.onDelivery(callbackService, request, expectedQos, false); assertNotNull("Properties should not have been null!", listener.actualDelivery.getProperties()); assertEquals("Expected no message properties", 0, listener.actualDelivery.getProperties().size()); assertEquals("Expected QOS to match", expectedQos, listener.actualDelivery.getQOS()); assertEquals("Expected share to match", null, listener.actualDelivery.getShare()); assertEquals("Expected topic to match", expectedTopic, listener.actualDelivery.getTopic()); assertEquals("Expected topic pattern to match", expectedTopicPattern, listener.actualDelivery.getTopicPattern()); assertEquals("Expected ttl to match", expectedTtl, listener.actualDelivery.getTtl()); assertEquals("Expected delivery to be of type JSON", Delivery.Type.JSON, listener.actualDelivery.getType()); JsonDelivery jsonDelivery = (JsonDelivery)listener.actualDelivery; assertEquals("Expected data to match", expectedData, jsonDelivery.getRawData()); ArrayList<Double> actual1List = jsonDelivery.getData(Double[].class.getGenericSuperclass()); int[] actual1 = new int[actual1List.size()]; for (int i = 0; i < actual1.length; actual1[i] = actual1List.get(i).intValue(), ++i); int[] actual2 = jsonDelivery.getData(int[].class); assertArrayEquals("Expected deserialized Json to match 1", expectedArray, actual1); assertArrayEquals("Expected deserialized Json to match 2", expectedArray, actual2); JsonElement element = jsonDelivery.getData(); assertTrue(element instanceof JsonArray); JsonArray array = element.getAsJsonArray(); for (int i = 0 ; i < array.size(); ++i) { int actual = array.get(i).getAsInt(); assertEquals("Expected array element #"+i+" to match", expectedArray[i], actual); } }