NonBlockingClientImpl extends NonBlockingClient implements FSMActions, Component, CallbackService { @Override public <T> NonBlockingClient subscribe(String topicPattern, SubscribeOptions subOptions, DestinationListener<T> destListener, CompletionListener<T> compListener, T context) throws SubscribedException, StoppedException, IllegalArgumentException { final String methodName = "subscribe"; logger.entry(this, methodName, topicPattern, subOptions, destListener, compListener, context); if (topicPattern == null) { final IllegalArgumentException exception = new IllegalArgumentException("Topic pattern cannot be null"); logger.throwing(this, methodName, exception); throw exception; } if (destListener == null) { final IllegalArgumentException exception = new IllegalArgumentException("DestinationListener cannot be null"); logger.throwing(this, methodName, exception); throw exception; } if (subOptions == null) subOptions = defaultSubscribeOptions; final SubscriptionTopic subTopic = new SubscriptionTopic(topicPattern, subOptions.getShareName()); boolean autoConfirm = subOptions.getAutoConfirm() || subOptions.getQOS() == QOS.AT_MOST_ONCE; InternalSubscribe<T> is = new InternalSubscribe<>(this, subTopic, subOptions.getQOS(), subOptions.getCredit(), autoConfirm, Math.round(subOptions.getTtl() / 1000.0), gsonBuilder, destListener, context); tell(is, this); try { is.future.setListener(callbackService, compListener, context); } catch (SubscribedException|StoppedException e) { logger.throwing(this, methodName, e); throw e; } catch (StateException e) { IllegalStateException exception = new IllegalStateException("Unexpected state exception", e); logger.ffdc(methodName, FFDCProbeId.PROBE_004, exception, this); logger.throwing(this, methodName, e); throw exception; } logger.exit(this, methodName, this); return this; } 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 testMessageTtlValuesIntoSubscribe() { class MockClient extends NonBlockingClientImpl { private final Queue<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.add(message); } protected java.util.Queue<Message> getMessages() { return messages; } } MockClient client = new MockClient(new StubEndpointService(), new StubCallbackService(), new MockComponent(), new StubTimerService(), null, null, null, null); final int[] testInputs = { 1, 500, 999, 1001 }; for (final int inputTtl : testInputs) { final SubscribeOptions subOptions = SubscribeOptions.builder() .setTtl(inputTtl).build(); client.subscribe("topicPattern", subOptions, new DestinationAdapter<Object>() { }, null, null); assertEquals( "Expected a single message to have been sent to the mock engine component", 1, client.getMessages().size()); final InternalSubscribe<?> subscribe = (InternalSubscribe<?>) client .getMessages().remove(); assertEquals( "Expected input " + inputTtl + " milliseconds ttl to have been rounded to the nearest second for subscribe - ", (int) Math.round(subOptions.getTtl() / 1000.0), subscribe.ttl); } }
@Test public void testNullValuesIntoSubscribe() { 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.subscribe("topicPattern", null, new DestinationAdapter<Object>(){}, null, null); try { client.subscribe(null, null, new DestinationAdapter<Object>(){}, null, null); throw new AssertionFailedError("Null topic pattern should have thrown an exception"); } catch(IllegalArgumentException e) { } try { client.subscribe("topicPattern", null, null, null, null); throw new AssertionFailedError("Null destination listener should have thrown an exception"); } catch(IllegalArgumentException e) { } }
@Test(expected=StoppedException.class) public void testSubscribeWhileStoppedThrowsException() { NonBlockingClientImpl client = stoppedClient(); client.subscribe("/kittens", new DestinationAdapter<Object>() {}, null, null); }