NonBlockingClientImpl extends NonBlockingClient implements FSMActions, Component, CallbackService { @Override public <T> boolean send(String topic, String data, Map<String, Object> properties, SendOptions sendOptions, CompletionListener<T> listener, T context) throws StoppedException { final String methodName = "send"; logger.entry(this, methodName, topic, data, properties, sendOptions, listener, context); if (data == null) { final IllegalArgumentException exception = new IllegalArgumentException("data cannot be null"); logger.throwing(this, methodName, exception); throw exception; } org.apache.qpid.proton.message.Message protonMsg = Proton.message(); protonMsg.setBody(new AmqpValue(data)); final boolean result = send(topic, protonMsg, properties, sendOptions == null ? defaultSendOptions : sendOptions, listener, context); logger.exit(this, methodName, result); return result; } protected <T> NonBlockingClientImpl(EndpointService endpointService,
            CallbackService callbackService,
            ComponentImpl engine,
            TimerService timerService,
            GsonBuilder gsonBuilder,
            ClientOptions options,
            NonBlockingClientListener<T>listener,
            T context); <T> NonBlockingClientImpl(EndpointService endpointService,
                                     CallbackService callbackService,
                                     NetworkService networkService,
                                     TimerService timerService,
                                     GsonBuilder gsonBuilder,
                                     ClientOptions options,
                                     NonBlockingClientListener<T>listener,
                                     T context); <T> NonBlockingClientImpl(String service, ClientOptions options, NonBlockingClientListener<T> listener, T context); @Override String getId(); @Override String getService(); @Override ClientState getState(); @Override boolean send(String topic, String data, Map<String, Object> properties,
            SendOptions sendOptions, CompletionListener<T> listener, T context); @Override boolean send(String topic, ByteBuffer data, Map<String, Object> properties,
            SendOptions sendOptions, CompletionListener<T> listener, T context); @Override boolean send(String topic, Object json,
            Map<String, Object> properties, SendOptions sendOptions,
            CompletionListener<T> listener, T context); @Override boolean send(String topic, Object json, Type type,
            Map<String, Object> properties, SendOptions sendOptions,
            CompletionListener<T> listener, T context); @Override boolean sendJson(String topic, String json,
            Map<String, Object> properties, SendOptions sendOptions,
            CompletionListener<T> listener, T context); @Override NonBlockingClient start(CompletionListener<T> listener, T context); @Override void stop(CompletionListener<T> listener, T context); @Override NonBlockingClient subscribe(String topicPattern,
            SubscribeOptions subOptions, DestinationListener<T> destListener,
            CompletionListener<T> compListener, T context); @Override NonBlockingClient unsubscribe(String topicPattern, String share, int ttl, CompletionListener<T> listener, T context); @Override NonBlockingClient unsubscribe(String topicPattern, String share, CompletionListener<T> listener, T context); @Override void startTimer(); @Override void openConnection(); @Override void closeConnection(); @Override void cancelTimer(); @Override void requestEndpoint(); @Override void remakeInboundLinks(); @Override void blessEndpoint(); @Override void cleanup(); @Override void failPendingStops(); @Override void succeedPendingStops(); @Override void failPendingStarts(); @Override void succeedPendingStarts(); @Override void processQueuedActions(); @Override void eventStarting(); @Override void eventUserStopping(); @Override void eventSystemStopping(); @Override void eventStopped(); @Override void eventStarted(); @Override void eventRetrying(); @Override void eventRestarted(); @Override void breakInboundLinks(); @Override void tell(Message message, Component self); @Override void run(Runnable runnable, Object orderingCtx, Promise<Void> promise);  }
@Test public void testNullValuesIntoSend() { StubEndpointService endpointService = new StubEndpointService(); StubCallbackService callbackService = new StubCallbackService(); MockComponent component = new MockComponent(); StubTimerService timerService = new StubTimerService(); NonBlockingClientImpl client = new NonBlockingClientImpl(endpointService, callbackService, component, timerService, null, null, null, null); client.send("topic", "data", (Map<String, Object>)null, null, null, null); client.send("topic", ByteBuffer.allocate(1), (Map<String, Object>)null, null, null, null); try { client.send(null, "data", (Map<String, Object>)null, null, null, null); throw new AssertionFailedError("Null topic (send String) should have thrown an exception"); } catch(IllegalArgumentException e) { } try { client.send(null, ByteBuffer.allocate(1), (Map<String, Object>)null, null, null, null); throw new AssertionFailedError("Null topic (send ByteBuffer) should have thrown an exception"); } catch(IllegalArgumentException e) { } try { client.send("topic", (String)null, (Map<String, Object>)null, null, null, null); throw new AssertionFailedError("Null data (send String) should have thrown an exception"); } catch(IllegalArgumentException e) { } try { client.send("topic", (ByteBuffer)null, (Map<String, Object>)null, null, null, null); throw new AssertionFailedError("Null data (send ByteBuffer) should have thrown an exception"); } catch(IllegalArgumentException e) { } }
@Test public void testRoundtripMessageProperties() { StubEndpointService endpointService = new StubEndpointService(); StubCallbackService callbackService = new StubCallbackService(); MockComponent component = new MockComponent(); StubTimerService timerService = new StubTimerService(); class MockClient extends NonBlockingClientImpl { private final LinkedList<Message> messages = new LinkedList<>(); protected <T> MockClient(EndpointService endpointService, CallbackService callbackService, ComponentImpl engine, TimerService timerService, GsonBuilder builder, ClientOptions options, NonBlockingClientListener<T> listener, T context) { super(endpointService, callbackService, engine, timerService, builder, options, listener, context); } @Override public void tell(Message message, Component self) { messages.addLast(message); } protected java.util.List<Message> getMessages() { return messages; } } class TestDestinationListener implements DestinationListener<Void> { protected Map<String, Object> properties = null; @Override public void onMessage(NonBlockingClient client, Void context, Delivery delivery) { properties = delivery.getProperties(); } @Override public void onMalformed(NonBlockingClient client, Void context, MalformedDelivery delivery) { } @Override public void onUnsubscribed(NonBlockingClient client, Void context, String topicPattern, String share, Exception error) { } } MockClient client = new MockClient(endpointService, callbackService, component, timerService, null, null, null, null); HashMap<String, Object> props = new HashMap<>(); props.put("boolean", true); props.put("byte", (byte)0x01); props.put("short", (short)123); props.put("int", 4567); props.put("long", (long)121723); props.put("float", (float)0.1234); props.put("double", 543.1234); props.put("byte[]", new byte[]{1,2,3,4}); props.put("Byte[]", new Byte[]{1,2,3,4}); props.put("string", "this is a string"); client.send("/kittens", "data", props); assertEquals("Expected a single message to have been sent to the mock engine component", 1, client.getMessages().size()); InternalSend<?> send = (InternalSend<?>)client.getMessages().get(0); byte[] msgData = new byte[send.length]; System.arraycopy(send.buf.array(), 0, msgData, 0, send.length); DeliveryRequest dr = new DeliveryRequest(msgData, QOS.AT_MOST_ONCE, "/kittens", null, null); TestDestinationListener destinationListener = new TestDestinationListener(); DestinationListenerWrapper<Void> wrapper = new DestinationListenerWrapper<>(client, new GsonBuilder(), destinationListener, null); wrapper.onDelivery(new SameThreadCallbackService(), dr, QOS.AT_MOST_ONCE, false); assertNotNull("Expected onMessage to have been called with message properties", destinationListener.properties); assertEquals("Expected all message properties to have been round-tripped", props.size(), destinationListener.properties.size()); Map<String, Object> actualProperties = destinationListener.properties; for (Map.Entry<String, Object> expectedProperty : props.entrySet()) { assertTrue("Round-tripped properties should have contained key: "+expectedProperty.getKey(), actualProperties.containsKey(expectedProperty.getKey())); if (expectedProperty.getValue() instanceof Byte[]) { final Byte[] expected = (Byte[]) expectedProperty.getValue(); final byte[] actual = (byte[]) actualProperties.get(expectedProperty.getKey()); for (int i = 0; i < expected.length; i++) { assertTrue("Round-tripped Byte array should match for key: "+expectedProperty.getKey(), expected[i] == actual[i]); } } else if (expectedProperty.getValue() instanceof byte[]) { assertTrue("Round-tripped byte array should match for key: "+expectedProperty.getKey(), Arrays.equals((byte[])expectedProperty.getValue(), (byte[])actualProperties.get(expectedProperty.getKey()))); } else { assertEquals("Ronnd-tripped value should match for key: "+expectedProperty.getKey(), expectedProperty.getValue(), actualProperties.get(expectedProperty.getKey())); } } }
@Test public void testValidPropertyValuesInSend() { StubEndpointService endpointService = new StubEndpointService(); StubCallbackService callbackService = new StubCallbackService(); MockComponent component = new MockComponent(); StubTimerService timerService = new StubTimerService(); NonBlockingClientImpl client = new NonBlockingClientImpl(endpointService, callbackService, component, timerService, null, null, null, null); client.send("topic", "data", getPropertiesMap("null", null)); client.send("topic", "data", getPropertiesMap("boolean", false)); client.send("topic", "data", getPropertiesMap("byte", (byte)3)); client.send("topic", "data", getPropertiesMap("short", (short)3)); client.send("topic", "data", getPropertiesMap("int", 3)); client.send("topic", "data", getPropertiesMap("long", 3L)); client.send("topic", "data", getPropertiesMap("float", (float)3.0)); client.send("topic", "data", getPropertiesMap("double", 3.0)); client.send("topic", "data", getPropertiesMap("byte[]", new byte[0])); client.send("topic", "data", getPropertiesMap("Byte[]", new Byte[0])); client.send("topic", "data", getPropertiesMap("string", "hello")); try { client.send("topic", "data", getPropertiesMap("Object", new Object())); throw new AssertionFailedError("Object property value should have thrown an Exception"); } catch(IllegalArgumentException e) { } try { client.send("topic", "data", getPropertiesMap("char", 'c')); throw new AssertionFailedError("char property value should have thrown an Exception"); } catch(IllegalArgumentException e) { } try { client.send("topic", "data", getPropertiesMap("BigDecimal", new BigDecimal(3))); throw new AssertionFailedError("BigDecimal property value should have thrown an Exception"); } catch(IllegalArgumentException e) { } }
@Test(expected=StoppedException.class) public void testSendWhileStoppedThrowsException() { NonBlockingClientImpl client = stoppedClient(); client.send("/kittens", "data", null); }