NonBlockingClientImpl extends NonBlockingClient implements FSMActions, Component, CallbackService { @Override public void breakInboundLinks() { final String methodName = "breakInboundLinks"; logger.entry(this, methodName); undrainedSends = 0; if (pendingDrain) { pendingDrain = false; clientListener.onDrain(callbackService); } for (InternalSend<?> sendRequest : outstandingSends.values()) { if (sendRequest.qos == QOS.AT_MOST_ONCE) { sendRequest.future.setSuccess(null); } else { pendingWork.addLast(sendRequest); } } outstandingSends.clear(); for (Map.Entry<SubscriptionTopic, SubData>entry : subscribedDestinations.entrySet()) { final SubData subData = entry.getValue(); subData.pendingDeliveries.clear(); while (!subData.pending.isEmpty()) { pendingWork.addLast(subData.pending.removeFirst()); } subData.state = SubData.State.BROKEN; } logger.exit(this, methodName); } 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(timeout=5000) public void testBreakInboundLinks() throws InterruptedException { MockComponent engine = new MockComponent(); MockNonBlockingClientListener listener = new MockNonBlockingClientListener(false); EngineConnection engineConnection = new EngineConnection(); NonBlockingClientImpl client = openCommon(engine, listener); OpenRequest openRequest = (OpenRequest)engine.getMessages().get(0); client.tell(new OpenResponse(openRequest, engineConnection), engine); assertEquals(ClientState.STARTED, client.getState()); final Semaphore done = new Semaphore(0); client.subscribe("/kittens", new DestinationAdapter<Void>() {}, new CompletionListener<Void>() { @Override public void onSuccess(NonBlockingClient client, Void context) { done.release(); } @Override public void onError(NonBlockingClient client, Void context, Exception exception) { done.release(); } }, null); client.tell(new SubscribeResponse(engineConnection, new SubscriptionTopic("/kittens", null)), engine); assertTrue("Client failed to subscribe within timeout. ", done.tryAcquire(4, TimeUnit.SECONDS)); client.tell(new DeliveryRequest(null, QOS.AT_LEAST_ONCE, "/kittens", null, null), engine); client.unsubscribe("/kittens", null, null); Thread.sleep(100L); client.breakInboundLinks(); }