Java 类org.springframework.util.concurrent.ListenableFutureCallback 实例源码

public void send(String topic, LifecycleEvent message) {
    // the KafkaTemplate provides asynchronous send methods returning a Future
    ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, message);

    // register a callback with the listener to receive the result of the send asynchronously
    future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {

        @Override
        public void onSuccess(SendResult<String, Object> result) {
            LOGGER.info("sent message='{}' with offset={}", message,
                    result.getRecordMetadata().offset());
        }

        @Override
        public void onFailure(Throwable ex) {
            LOGGER.error("unable to send message='{}'", message, ex);
        }
    });
}

@GetMapping(value="/webSyncDept/{id}.json", produces ="application/json", headers = {"Accept=text/xml, application/json"})
public WebAsyncTask<Department> websyncDeptList(@PathVariable("id") Integer id){

    Callable<Department> callable = new Callable<Department>() {
        public Department call() throws Exception {

             ListenableFuture<Department> listenFuture = departmentServiceImpl.findAllFirstById(id);
             listenFuture.addCallback(new ListenableFutureCallback<Department>(){

                @Override
                public void onSuccess(Department dept) {
                    result = dept;
                }

                @Override
                public void onFailure(Throwable arg0) {
                    result = new Department();
                }

             });
             return result;
          }
    };
    return new WebAsyncTask<Department>(500, callable);
}

@RequestMapping(method = RequestMethod.GET, path = "/createCell")
public void createCell(@NotNull @RequestParam String name, @NotNull @RequestParam CellType cellType) throws ExecutionException, InterruptedException {
    JSONObject cell = new JSONObject();
    cell.put("name", name);
    cell.put("type", cellType);

    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, cell.toString());
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
        @Override
        public void onSuccess(SendResult<String, String> result) {
            System.out.println("success");
        }

        @Override
        public void onFailure(Throwable ex) {
            System.out.println("failed");
        }
    });
}

@Bean
@ServiceActivator(inputChannel = "pubSubOutputChannel")
public MessageHandler messageSender(PubSubTemplate pubsubTemplate) {
    PubSubMessageHandler adapter =
            new PubSubMessageHandler(pubsubTemplate, "exampleTopic");
    adapter.setPublishCallback(new ListenableFutureCallback<String>() {
        @Override
        public void onFailure(Throwable ex) {
            LOGGER.info("There was an error sending the message.");
        }

        @Override
        public void onSuccess(String result) {
            LOGGER.info("Message was sent successfully.");
        }
    });

    return adapter;
}

@Test
public void testPublishCallback() {
    ListenableFutureCallback<String> callbackSpy = spy(new ListenableFutureCallback<String>() {
        @Override
        public void onFailure(Throwable ex) {

        }

        @Override
        public void onSuccess(String result) {

        }
    });

    this.adapter.setPublishCallback(callbackSpy);

    this.adapter.handleMessage(this.message);

    verify(callbackSpy, times(1)).onSuccess(eq("benfica"));
}

private void sendSystemSubscriptions() {
    int i = 0;
    for (String destination : getSystemSubscriptions().keySet()) {
        StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.SUBSCRIBE);
        accessor.setSubscriptionId(String.valueOf(i++));
        accessor.setDestination(destination);
        if (logger.isDebugEnabled()) {
            logger.debug("Subscribing to " + destination + " on \"system\" connection.");
        }
        TcpConnection<byte[]> conn = getTcpConnection();
        if (conn != null) {
            MessageHeaders headers = accessor.getMessageHeaders();
            conn.send(MessageBuilder.createMessage(EMPTY_PAYLOAD, headers)).addCallback(
                    new ListenableFutureCallback<Void>() {
                        public void onSuccess(Void result) {
                        }
                        public void onFailure(Throwable ex) {
                            String error = "Failed to subscribe in \"system\" session.";
                            handleTcpConnectionFailure(error, ex);
                        }
                    });
        }
    }
}

@Override
public void handleReturnValue(Object returnValue, MethodParameter returnType,
        ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {

    if (returnValue == null) {
        mavContainer.setRequestHandled(true);
        return;
    }

    final DeferredResult<Object> deferredResult = new DeferredResult<Object>();
    WebAsyncUtils.getAsyncManager(webRequest).startDeferredResultProcessing(deferredResult, mavContainer);

    ListenableFuture<?> future = (ListenableFuture<?>) returnValue;
    future.addCallback(new ListenableFutureCallback<Object>() {
        @Override
        public void onSuccess(Object result) {
            deferredResult.setResult(result);
        }
        @Override
        public void onFailure(Throwable ex) {
            deferredResult.setErrorResult(ex);
        }
    });
}

@Override
protected void openConnection() {
    if (logger.isInfoEnabled()) {
        logger.info("Connecting to WebSocket at " + getUri());
    }

    ListenableFuture<WebSocketSession> future =
            this.client.doHandshake(this.webSocketHandler, this.headers, getUri());

    future.addCallback(new ListenableFutureCallback<WebSocketSession>() {
        @Override
        public void onSuccess(WebSocketSession result) {
            webSocketSession = result;
            logger.info("Successfully connected");
        }
        @Override
        public void onFailure(Throwable ex) {
            logger.error("Failed to connect", ex);
        }
    });
}

@Override
public ListenableFuture<WebSocketSession> connect(TransportRequest request, WebSocketHandler handler) {
    final SettableListenableFuture<WebSocketSession> future = new SettableListenableFuture<WebSocketSession>();
    WebSocketClientSockJsSession session = new WebSocketClientSockJsSession(request, handler, future);
    handler = new ClientSockJsWebSocketHandler(session);
    request.addTimeoutTask(session.getTimeoutTask());

    URI url = request.getTransportUrl();
    WebSocketHttpHeaders headers = new WebSocketHttpHeaders(request.getHandshakeHeaders());
    if (logger.isDebugEnabled()) {
        logger.debug("Starting WebSocket session url=" + url);
    }
    this.webSocketClient.doHandshake(handler, headers, url).addCallback(
            new ListenableFutureCallback<WebSocketSession>() {
                @Override
                public void onSuccess(WebSocketSession webSocketSession) {
                    // WebSocket session ready, SockJS Session not yet
                }
                @Override
                public void onFailure(Throwable ex) {
                    future.setException(ex);
                }
            });
    return future;
}

@Test
public void infoRequestFailure() throws Exception {
    TestClientHandler handler = new TestClientHandler();
    this.testFilter.sendErrorMap.put("/info", 500);
    CountDownLatch latch = new CountDownLatch(1);
    initSockJsClient(createWebSocketTransport());
    this.sockJsClient.doHandshake(handler, this.baseUrl + "/echo").addCallback(
            new ListenableFutureCallback<WebSocketSession>() {
                @Override
                public void onSuccess(WebSocketSession result) {
                }

                @Override
                public void onFailure(Throwable ex) {
                    latch.countDown();
                }
            }
    );
    assertTrue(latch.await(5000, TimeUnit.MILLISECONDS));
}

@Test
public void connectFailure() throws Exception {
    final HttpServerErrorException expected = new HttpServerErrorException(HttpStatus.INTERNAL_SERVER_ERROR);
    RestOperations restTemplate = mock(RestOperations.class);
    given(restTemplate.execute((URI) any(), eq(HttpMethod.POST), any(), any())).willThrow(expected);

    final CountDownLatch latch = new CountDownLatch(1);
    connect(restTemplate).addCallback(
            new ListenableFutureCallback<WebSocketSession>() {
                @Override
                public void onSuccess(WebSocketSession result) {
                }
                @Override
                public void onFailure(Throwable ex) {
                    if (ex == expected) {
                        latch.countDown();
                    }
                }
            }
    );
    verifyNoMoreInteractions(this.webSocketHandler);
}

@Test
public void getEntityCallback() throws Exception {
    ListenableFuture<ResponseEntity<String>> futureEntity =
            template.getForEntity(baseUrl + "/{method}", String.class, "get");
    futureEntity.addCallback(new ListenableFutureCallback<ResponseEntity<String>>() {
        @Override
        public void onSuccess(ResponseEntity<String> entity) {
            assertEquals("Invalid content", helloWorld, entity.getBody());
            assertFalse("No headers", entity.getHeaders().isEmpty());
            assertEquals("Invalid content-type", textContentType, entity.getHeaders().getContentType());
            assertEquals("Invalid status code", HttpStatus.OK, entity.getStatusCode());
        }
        @Override
        public void onFailure(Throwable ex) {
            fail(ex.getMessage());
        }
    });
    // wait till done
    while (!futureEntity.isDone()) {
    }
}

@Test
public void headForHeadersCallback() throws Exception {
    ListenableFuture<HttpHeaders> headersFuture = template.headForHeaders(baseUrl + "/get");
    headersFuture.addCallback(new ListenableFutureCallback<HttpHeaders>() {
        @Override
        public void onSuccess(HttpHeaders result) {
            assertTrue("No Content-Type header", result.containsKey("Content-Type"));
        }
        @Override
        public void onFailure(Throwable ex) {
            fail(ex.getMessage());
        }
    });
    while (!headersFuture.isDone()) {
    }
}

@Test
public void postForLocationCallback() throws Exception  {
    HttpHeaders entityHeaders = new HttpHeaders();
    entityHeaders.setContentType(new MediaType("text", "plain", Charset.forName("ISO-8859-15")));
    HttpEntity<String> entity = new HttpEntity<String>(helloWorld, entityHeaders);
    final URI expected = new URI(baseUrl + "/post/1");
    ListenableFuture<URI> locationFuture = template.postForLocation(baseUrl + "/{method}", entity, "post");
    locationFuture.addCallback(new ListenableFutureCallback<URI>() {
        @Override
        public void onSuccess(URI result) {
            assertEquals("Invalid location", expected, result);
        }
        @Override
        public void onFailure(Throwable ex) {
            fail(ex.getMessage());
        }
    });
    while (!locationFuture.isDone()) {
    }
}

@Test
public void postForEntityCallback() throws Exception  {
    HttpEntity<String> requestEntity = new HttpEntity<>(helloWorld);
    ListenableFuture<ResponseEntity<String>> responseEntityFuture =
            template.postForEntity(baseUrl + "/{method}", requestEntity, String.class, "post");
    responseEntityFuture.addCallback(new ListenableFutureCallback<ResponseEntity<String>>() {
        @Override
        public void onSuccess(ResponseEntity<String> result) {
            assertEquals("Invalid content", helloWorld, result.getBody());
        }
        @Override
        public void onFailure(Throwable ex) {
            fail(ex.getMessage());
        }
    });
    while (!responseEntityFuture.isDone()) {
    }
}

@Test
public void putCallback() throws Exception  {
    HttpEntity<String> requestEntity = new HttpEntity<>(helloWorld);
    ListenableFuture<?> responseEntityFuture = template.put(baseUrl + "/{method}", requestEntity, "put");
    responseEntityFuture.addCallback(new ListenableFutureCallback<Object>() {
        @Override
        public void onSuccess(Object result) {
            assertNull(result);
        }
        @Override
        public void onFailure(Throwable ex) {
            fail(ex.getMessage());
        }
    });
    while (!responseEntityFuture.isDone()) {
    }
}

@Test
public void deleteCallback() throws Exception  {
    ListenableFuture<?> deletedFuture = template.delete(new URI(baseUrl + "/delete"));
    deletedFuture.addCallback(new ListenableFutureCallback<Object>() {
        @Override
        public void onSuccess(Object result) {
            assertNull(result);
        }
        @Override
        public void onFailure(Throwable ex) {
            fail(ex.getMessage());
        }
    });
    while (!deletedFuture.isDone()) {
    }
}

@Test
public void notFoundCallback() throws Exception {
    ListenableFuture<?> future = template.execute(baseUrl + "/status/notfound", HttpMethod.GET, null, null);
    future.addCallback(new ListenableFutureCallback<Object>() {
        @Override
        public void onSuccess(Object result) {
            fail("onSuccess not expected");
        }
        @Override
        public void onFailure(Throwable t) {
            assertTrue(t instanceof HttpClientErrorException);
            HttpClientErrorException ex = (HttpClientErrorException) t;
            assertEquals(HttpStatus.NOT_FOUND, ex.getStatusCode());
            assertNotNull(ex.getStatusText());
            assertNotNull(ex.getResponseBodyAsString());
        }
    });
    while (!future.isDone()) {
    }
}

@Test
public void serverErrorCallback() throws Exception {
    ListenableFuture<Void> future = template.execute(baseUrl + "/status/server", HttpMethod.GET, null, null);
    future.addCallback(new ListenableFutureCallback<Void>() {
        @Override
        public void onSuccess(Void result) {
            fail("onSuccess not expected");
        }
        @Override
        public void onFailure(Throwable ex) {
            assertTrue(ex instanceof HttpServerErrorException);
            HttpServerErrorException hsex = (HttpServerErrorException) ex;
            assertEquals(HttpStatus.INTERNAL_SERVER_ERROR, hsex.getStatusCode());
            assertNotNull(hsex.getStatusText());
            assertNotNull(hsex.getResponseBodyAsString());
        }
    });
    while (!future.isDone()) {
    }
}

@Test
public void optionsForAllowCallback() throws Exception {
    ListenableFuture<Set<HttpMethod>> allowedFuture = template.optionsForAllow(new URI(baseUrl + "/get"));
    allowedFuture.addCallback(new ListenableFutureCallback<Set<HttpMethod>>() {
        @Override
        public void onSuccess(Set<HttpMethod> result) {
            assertEquals("Invalid response", EnumSet.of(HttpMethod.GET, HttpMethod.OPTIONS,
                    HttpMethod.HEAD, HttpMethod.TRACE), result);
        }
        @Override
        public void onFailure(Throwable ex) {
            fail(ex.getMessage());
        }
    });
    while (!allowedFuture.isDone()) {
    }
}

@Test
@SuppressWarnings({ "unchecked", "rawtypes" })
public void exchangeGetCallback() throws Exception {
    HttpHeaders requestHeaders = new HttpHeaders();
    requestHeaders.set("MyHeader", "MyValue");
    HttpEntity<?> requestEntity = new HttpEntity(requestHeaders);
    ListenableFuture<ResponseEntity<String>> responseFuture =
            template.exchange(baseUrl + "/{method}", HttpMethod.GET, requestEntity, String.class, "get");
    responseFuture.addCallback(new ListenableFutureCallback<ResponseEntity<String>>() {
        @Override
        public void onSuccess(ResponseEntity<String> result) {
            assertEquals("Invalid content", helloWorld, result.getBody());
        }
        @Override
        public void onFailure(Throwable ex) {
            fail(ex.getMessage());
        }
    });
    while (!responseFuture.isDone()) {
    }
}

@Test
public void exchangePostCallback() throws Exception {
    HttpHeaders requestHeaders = new HttpHeaders();
    requestHeaders.set("MyHeader", "MyValue");
    requestHeaders.setContentType(MediaType.TEXT_PLAIN);
    HttpEntity<String> requestEntity = new HttpEntity<String>(helloWorld, requestHeaders);
    ListenableFuture<ResponseEntity<Void>> resultFuture =
            template.exchange(baseUrl + "/{method}", HttpMethod.POST, requestEntity, Void.class, "post");
    final URI expected =new URI(baseUrl + "/post/1");
    resultFuture.addCallback(new ListenableFutureCallback<ResponseEntity<Void>>() {
        @Override
        public void onSuccess(ResponseEntity<Void> result) {
            assertEquals("Invalid location", expected, result.getHeaders().getLocation());
            assertFalse(result.hasBody());
        }
        @Override
        public void onFailure(Throwable ex) {
            fail(ex.getMessage());
        }
    });
    while (!resultFuture.isDone()) {
    }
}

@Before
public void beforeMethod() {
    listenableFutureCallbackMock = mock(ListenableFutureCallback.class);

    successInObj = new Object();
    failureInObj = new Exception("kaboom");
    throwExceptionDuringCall = false;
    currentSpanStackWhenListenableFutureCallbackWasCalled = new ArrayList<>();
    currentMdcInfoWhenListenableFutureCallbackWasCalled = new ArrayList<>();
    doAnswer(invocation -> {
        currentSpanStackWhenListenableFutureCallbackWasCalled.add(Tracer.getInstance().getCurrentSpanStackCopy());
        currentMdcInfoWhenListenableFutureCallbackWasCalled.add(MDC.getCopyOfContextMap());
        if (throwExceptionDuringCall)
            throw new RuntimeException("kaboom");
        return null;
    }).when(listenableFutureCallbackMock).onSuccess(successInObj);
    doAnswer(invocation -> {
        currentSpanStackWhenListenableFutureCallbackWasCalled.add(Tracer.getInstance().getCurrentSpanStackCopy());
        currentMdcInfoWhenListenableFutureCallbackWasCalled.add(MDC.getCopyOfContextMap());
        if (throwExceptionDuringCall)
            throw new RuntimeException("kaboom");
        return null;
    }).when(listenableFutureCallbackMock).onFailure(failureInObj);

    resetTracing();
}

public void sendMessage(String topic, String message) {
    // the KafkaTemplate provides asynchronous send methods returning a
    // Future
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);

    // you can register a callback with the listener to receive the result
    // of the send asynchronously
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

        @Override
        public void onSuccess(SendResult<String, String> result) {
            logger.info("sent message='{}' with offset={}", message, result.getRecordMetadata().offset());
        }

        @Override
        public void onFailure(Throwable ex) {
            logger.error("unable to send message='{}'", message, ex);
        }
    });

    // alternatively, to block the sending thread, to await the result,
    // invoke the future’s get() method
}

/**
 * Will use the WriteListener if the TaskExecutor is an instance of AsyncListenableTaskExecutor. The WriteListener
 * will then be used to listen for failures.
 *
 * @param runnable
 * @param items
 */
protected void executeRunnable(Runnable runnable, final List<? extends DocumentWriteOperation> items) {
    if (writeListener != null && taskExecutor instanceof AsyncListenableTaskExecutor) {
        AsyncListenableTaskExecutor asyncListenableTaskExecutor = (AsyncListenableTaskExecutor)taskExecutor;
        ListenableFuture<?> future = asyncListenableTaskExecutor.submitListenable(runnable);
        future.addCallback(new ListenableFutureCallback<Object>() {
            @Override
            public void onFailure(Throwable ex) {
                writeListener.onWriteFailure(ex, items);
            }
            @Override
            public void onSuccess(Object result) {
            }
        });
    } else {
        taskExecutor.execute(runnable);
    }
}

@Override
public void waitForCalculationToFinish(ListenableFuture<String> convertedFuture) throws InterruptedException {
    final CountDownLatch latch = new CountDownLatch(1);
    convertedFuture.addCallback(new ListenableFutureCallback<String>() {
        @Override
        public void onSuccess(String result) {
            latch.countDown();
        }

        @Override
        public void onFailure(Throwable t) {
            latch.countDown();
        }
    });
    latch.await(1, TimeUnit.SECONDS);
}

@Override
public void addCallbackTo(ListenableFuture<String> convertedFuture) {
    convertedFuture.addCallback(new ListenableFutureCallback<String>() {
        @Override
        public void onSuccess(String result) {
            callback.onSuccess(result);
            callbackCalled();
        }

        @Override
        public void onFailure(Throwable t) {
            callback.onFailure(t);
            callbackCalled();
        }
    });
}

public void getSleepResource(long replyAfterMillis,
        final Callback<String> callback) throws Exception {
    ListenableFuture<ResponseEntity<String>> response = client
            .getForEntity(
                    "http://localhost:8001?replyAfterMillis={replyAfterMillis}",
                    String.class, replyAfterMillis);
    response.addCallback(new ListenableFutureCallback<ResponseEntity<String>>() {
        @Override
        public void onSuccess(ResponseEntity<String> result) {
            callback.callback(result.getBody());
        }

        @Override
        public void onFailure(Throwable t) {
            t.printStackTrace();
        }
    });
}

@RequestMapping("/")
@ResponseBody
DeferredResult<String> home() {
    // Create DeferredResult with timeout 5s
    final DeferredResult<String> result = new DeferredResult<>(5000);

    // Let's call the backend
    ListenableFuture<ResponseEntity<String>> future = restTemplate.getForEntity("http://www.google.com", String.class);
    future.addCallback(new ListenableFutureCallback<ResponseEntity<String>>() {
        @Override
        public void onSuccess(ResponseEntity<String> response) {
            // Will be called in HttpClient thread
            log("Success");
            result.setResult(response.getBody());
        }

        @Override
        public void onFailure(Throwable t) {
            result.setErrorResult(t.getMessage());
        }
    });
    // Return the thread to servlet container, the response will be processed by another thread.
    return result;
}

@Test
public void testTransformationFromSpring() {
    ListenableFuture<String> springListenableFuture = createSpringListenableFuture();

    CompletableCompletionStage<Object> completionStage = factory.createCompletionStage();
    springListenableFuture.addCallback(new ListenableFutureCallback<String>() {
        @Override
        public void onSuccess(String result) {
            completionStage.complete(result);
        }

        @Override
        public void onFailure(Throwable t) {
            completionStage.completeExceptionally(t);
        }
    });

    completionStage.thenAccept(System.out::println);
}

@RequestMapping("/")
@ResponseBody
DeferredResult<String> home() {
    DeferredResult<String> result = new DeferredResult<>(5000);

    ListenableFuture<ResponseEntity<String>> creditRatingFuture = restTemplate.getForEntity("http://www.google.com", String.class);
    creditRatingFuture.addCallback(new ListenableFutureCallback<ResponseEntity<String>>() {
        @Override
        public void onSuccess(ResponseEntity<String>  response) {
            log("Success");
            result.setResult(response.getBody());
        }

        @Override
        public void onFailure(Throwable t) {
            result.setErrorResult(t.getMessage());
        }
    });
    return result;
}

@Override
public final Mono<String> request(String path) {
    return this.host
        .flatMap(host -> Mono.<ResponseEntity<String>>create(emitter -> this.restOperations.getForEntity(String.format("http://%s%s", host, path), String.class)
            .addCallback(new ListenableFutureCallback<ResponseEntity<String>>() {

                @Override
                public void onFailure(Throwable ex) {
                    emitter.error(ex);
                }

                @Override
                public void onSuccess(ResponseEntity<String> result) {
                    emitter.success(result);
                }

            })))
        .doOnError(t -> this.logger.warn("Error while making request: {}", t.getMessage()))
        .onErrorResume(this::printRecentLogs)
        .map(HttpEntity::getBody);
}

public void send(String topic, String data) {
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, data);
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
        @Override
        public void onSuccess(SendResult<String, String> result) {
            logger.info("Success on sending message \"" + data + "\" to topic " + topic);
        }

        @Override
        public void onFailure(Throwable ex) {
            logger.error("Error on sending message \"" + data + "\", stacktrace " + ex.getMessage());
        }
    });
}

public void send(String topic, String data) {
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, data);
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
        @Override
        public void onSuccess(SendResult<String, String> result) {
            logger.info("Success on sending message \"" + data + "\" to topic " + topic);
        }

        @Override
        public void onFailure(Throwable ex) {
            logger.error("Error on sending message \"" + data + "\", stacktrace " + ex.getMessage());
        }
    });
}

/**
 * Metodo funndamental para la informacion en Kafka.
 * @param topic Topico de Kafka.
 * @param data Datos a enviar.
 */
public void send(String topic, String data) {
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, data); //Valido por si solo para el manejo de la informacion.
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
        @Override
        public void onSuccess(SendResult<String, String> result) {
            logger.info("Success on sending message \"" + data + "\" to topic " + topic);
        }

        @Override
        public void onFailure(Throwable ex) {
            logger.error("Error on sending message \"" + data + "\", stacktrace " + ex.getMessage());
        }
    });
}

public void send(String topic, String data) {
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, data);
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
        @Override
        public void onSuccess(SendResult<String, String> result) {
            logger.info(result);
            logger.info("Success on sending message \"" + data + "\" to topic " + topic);
        }

        @Override
        public void onFailure(Throwable ex) {
            logger.error("Error on sending message \"" + data + "\", stacktrace " + ex.getMessage());
        }
    });
}

public static void main(String[] args) {
    ListenableFutureTask listenableFutureTask = new ListenableFutureTask(() -> {
        log.info("listenable future begin to execute");
        TimeUnit.SECONDS.sleep(5);
        return "listenable future task done";
    });

    listenableFutureTask.addCallback(new ListenableFutureCallback() {
        @Override
        public void onFailure(Throwable ex) {
            log.info("listenable future task execute failed, cause: {}", ex.getMessage());
        }

        @Override
        public void onSuccess(Object result) {
            log.info("listenable future task execute successfully, result: {}", result);
        }
    });

    ExecutorService executor = Executors.newSingleThreadExecutor();
    try {
        executor.execute(listenableFutureTask);
        //log.info(listenableFutureTask.get());
    } finally {
        executor.shutdown();
    }
}

@Override
public void run() {
    TcpConnection<byte[]> conn = connection;
    if (conn != null) {
        conn.send(HEARTBEAT).addCallback(
                new ListenableFutureCallback<Void>() {
                    public void onSuccess(Void result) {
                    }
                    public void onFailure(Throwable ex) {
                        handleFailure(ex);
                    }
                });
    }
}

@Before
@SuppressWarnings("unchecked")
public void setup() {
    this.infoReceiver = mock(InfoReceiver.class);
    this.webSocketTransport = new TestTransport("WebSocketTestTransport");
    this.xhrTransport = new XhrTestTransport("XhrTestTransport");

    List<Transport> transports = new ArrayList<>();
    transports.add(this.webSocketTransport);
    transports.add(this.xhrTransport);
    this.sockJsClient = new SockJsClient(transports);
    this.sockJsClient.setInfoReceiver(this.infoReceiver);

    this.connectCallback = mock(ListenableFutureCallback.class);
}

@SuppressWarnings("unchecked")
@Before
public void setup() throws Exception {
    this.connectCallback = mock(ListenableFutureCallback.class);
    this.connectFuture = new SettableListenableFuture<>();
    this.connectFuture.addCallback(this.connectCallback);
    this.webSocketTransport = new TestTransport("WebSocketTestTransport");
    this.xhrTransport = new TestTransport("XhrTestTransport");
}