import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public static <K, V> KafkaTemplate<K, V> mockTemplate()
        throws InterruptedException, ExecutionException, TimeoutException {
    KafkaTemplate<K, V> template = mock(KafkaTemplate.class);
    SettableListenableFuture future = mock(SettableListenableFuture.class);
    when(future.get(30000, TimeUnit.MILLISECONDS)).thenReturn(null);
    when(template.send(any(), any(), any())).thenReturn(future);
    return template;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@Test
public void sendWithExecutionException() throws Exception {

	this.session.afterConnected(this.connection);
	assertTrue(this.session.isConnected());

	IllegalStateException exception = new IllegalStateException("simulated exception");
	SettableListenableFuture<Void> future = new SettableListenableFuture<>();
	future.setException(exception);

	when(this.connection.send(any())).thenReturn(future);
	this.expected.expect(MessageDeliveryException.class);
	this.expected.expectCause(Matchers.sameInstance(exception));

	this.session.send("/topic/foo", "sample payload".getBytes(UTF_8));

	verifyNoMoreInteractions(this.connection);
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@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;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@Override
public final ListenableFuture<WebSocketSession> doHandshake(
		WebSocketHandler handler, WebSocketHttpHeaders headers, URI url) {

	Assert.notNull(handler, "WebSocketHandler is required");
	Assert.notNull(url, "URL is required");

	String scheme = url.getScheme();
	if (!supportedProtocols.contains(scheme)) {
		throw new IllegalArgumentException("Invalid scheme: '" + scheme + "'");
	}

	SettableListenableFuture<WebSocketSession> connectFuture = new SettableListenableFuture<WebSocketSession>();
	try {
		SockJsUrlInfo sockJsUrlInfo = new SockJsUrlInfo(url);
		ServerInfo serverInfo = getServerInfo(sockJsUrlInfo, getHttpRequestHeaders(headers));
		createRequest(sockJsUrlInfo, headers, serverInfo).connect(handler, connectFuture);
	}
	catch (Throwable exception) {
		if (logger.isErrorEnabled()) {
			logger.error("Initial SockJS \"Info\" request to server failed, url=" + url, exception);
		}
		connectFuture.setException(exception);
	}
	return connectFuture;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@Override
@SuppressWarnings("deprecation")
public ListenableFuture<WebSocketSession> connect(TransportRequest request, WebSocketHandler handler) {
	SettableListenableFuture<WebSocketSession> connectFuture = new SettableListenableFuture<WebSocketSession>();
	XhrClientSockJsSession session = new XhrClientSockJsSession(request, handler, this, connectFuture);
	request.addTimeoutTask(session.getTimeoutTask());

	URI receiveUrl = request.getTransportUrl();
	if (logger.isDebugEnabled()) {
		logger.debug("Starting XHR " +
				(isXhrStreamingDisabled() ? "Polling" : "Streaming") + "session url=" + receiveUrl);
	}

	HttpHeaders handshakeHeaders = new HttpHeaders();
	handshakeHeaders.putAll(getRequestHeaders());
	handshakeHeaders.putAll(request.getHandshakeHeaders());

	connectInternal(request, handler, receiveUrl, handshakeHeaders, session, connectFuture);
	return connectFuture;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@Override
public ListenableFuture<Void> send(Message<byte[]> message) {
	updateLastWriteTime();
	SettableListenableFuture<Void> future = new SettableListenableFuture<Void>();
	try {
		this.session.sendMessage(this.codec.encode(message, this.session.getClass()));
		future.set(null);
	}
	catch (Throwable ex) {
		future.setException(ex);
	}
	finally {
		updateLastWriteTime();
	}
	return future;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@Override
protected ListenableFuture<ClientHttpResponse> executeInternal(final HttpHeaders headers) throws IOException {
	final SettableListenableFuture<ClientHttpResponse> responseFuture =
			new SettableListenableFuture<ClientHttpResponse>();

	ChannelFutureListener connectionListener = new ChannelFutureListener() {
		@Override
		public void operationComplete(ChannelFuture future) throws Exception {
			if (future.isSuccess()) {
				Channel channel = future.channel();
				channel.pipeline().addLast(new RequestExecuteHandler(responseFuture));
				FullHttpRequest nettyRequest = createFullHttpRequest(headers);
				channel.writeAndFlush(nettyRequest);
			}
			else {
				responseFuture.setException(future.cause());
			}
		}
	};

	this.bootstrap.connect(this.uri.getHost(), getPort(this.uri)).addListener(connectionListener);

	return responseFuture;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
ListenableFuture<ResponseEntity<?>> validateAndStoreSpans(String encoding, Codec codec,
    byte[] body) {
  SettableListenableFuture<ResponseEntity<?>> result = new SettableListenableFuture<>();
  metrics.incrementMessages();
  if (encoding != null && encoding.contains("gzip")) {
    try {
      body = gunzip(body);
    } catch (IOException e) {
      metrics.incrementMessagesDropped();
      result.set(ResponseEntity.badRequest().body("Cannot gunzip spans: " + e.getMessage() + "\n"));
    }
  }
  collector.acceptSpans(body, codec, new Callback<Void>() {
    @Override public void onSuccess(@Nullable Void value) {
      result.set(SUCCESS);
    }

    @Override public void onError(Throwable t) {
      String message = t.getMessage() == null ? t.getClass().getSimpleName() : t.getMessage();
      result.set(t.getMessage() == null || message.startsWith("Cannot store")
          ? ResponseEntity.status(500).body(message + "\n")
          : ResponseEntity.status(400).body(message + "\n"));
    }
  });
  return result;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
protected ListenableFuture<ClientHttpResponse> executeInternal(final HttpHeaders headers) {
    final SettableListenableFuture<ClientHttpResponse> responseFuture = new SettableListenableFuture<>();

    ChannelFutureListener connectionListener = future -> {
        if (future.isSuccess()) {
            Channel channel = future.channel();
            channel.pipeline().addLast(new NettyResponseHandler(responseFuture));
            FullHttpRequest nettyRequest = createFullHttpRequest(headers);
            channel.writeAndFlush(nettyRequest);
        }
        else {
            responseFuture.setException(future.cause());
        }
    };

    this.bootstrap.connect(this.uri.getHost(), getPort(this.uri)).addListener(connectionListener);

    return responseFuture;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
public ListenableFuture<List<Build>> getBuilds(String projectId, String baseBuildNumber) {
  SettableListenableFuture<List<Build>> builds = new SettableListenableFuture<>();
  ListenableFuture<List<BuildStep>> projectBuildSteps = getProjectBuildSteps(projectId);
  projectBuildSteps.addCallback(
      buildSteps -> {
        List<ListenableFuture<StepBuilds>> allBuilds = new ArrayList<>();
        buildSteps.forEach(buildStep -> {
          allBuilds.add(fetchBuildsForStep(buildStep));
        });
        ListenableFuture<List<StepBuilds>> buildsOnFuture = asyncUtil.flatMapCommands(allBuilds);
        buildsOnFuture.addCallback(
            buildsList -> {
              Collections.sort(buildsList);
              List<Build> relatedBuilds = new ArrayList<>();
              buildsList.forEach(currentBuilds -> {
                addRelatedBuildStep(currentBuilds.getBuild(), relatedBuilds, baseBuildNumber);
              });
              builds.set(relatedBuilds);
            },
            builds::setException);
      },
      builds::setException);
  return builds;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@Override
public ListenableFuture<String> publish(final String topic, PubsubMessage pubsubMessage) {
	ApiFuture<String> publishFuture =
			this.publisherFactory.createPublisher(topic).publish(pubsubMessage);

	final SettableListenableFuture<String> settableFuture = new SettableListenableFuture<>();
	ApiFutures.addCallback(publishFuture, new ApiFutureCallback<String>() {

		@Override
		public void onFailure(Throwable throwable) {
			LOGGER.warn("Publishing to " + topic + " topic failed.", throwable);
			settableFuture.setException(throwable);
		}

		@Override
		public void onSuccess(String result) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.debug(
						"Publishing to " + topic + " was successful. Message ID: " + result);
			}
			settableFuture.set(result);
		}

	});

	return settableFuture;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@Before
public void setUp() {
	this.message = new GenericMessage<>("testPayload",
			ImmutableMap.of("key1", "value1", "key2", "value2"));
	SettableListenableFuture<String> future = new SettableListenableFuture<>();
	future.set("benfica");
	when(this.pubSubTemplate.publish(eq("testTopic"),
			eq(ByteString.copyFrom("testPayload", Charset.defaultCharset())),
			isA(Map.class)))
			.thenReturn(future);
	this.adapter = new PubSubMessageHandler(this.pubSubTemplate, "testTopic");

}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@Before
public void setUp() throws Exception {

	MockitoAnnotations.initMocks(this);

	this.sessionHandler = mock(StompSessionHandler.class);
	this.connectHeaders = new StompHeaders();
	this.session = new DefaultStompSession(this.sessionHandler, this.connectHeaders);
	this.session.setMessageConverter(new StringMessageConverter());

	SettableListenableFuture<Void> future = new SettableListenableFuture<>();
	future.set(null);
	when(this.connection.send(this.messageCaptor.capture())).thenReturn(future);
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
protected AbstractClientSockJsSession(TransportRequest request, WebSocketHandler handler,
		SettableListenableFuture<WebSocketSession> connectFuture) {

	Assert.notNull(request, "'request' is required");
	Assert.notNull(handler, "'handler' is required");
	Assert.notNull(connectFuture, "'connectFuture' is required");
	this.request = request;
	this.webSocketHandler = handler;
	this.connectFuture = connectFuture;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@Override
protected void connectInternal(final TransportRequest transportRequest, final WebSocketHandler handler,
		final URI receiveUrl, final HttpHeaders handshakeHeaders, final XhrClientSockJsSession session,
		final SettableListenableFuture<WebSocketSession> connectFuture) {

	getTaskExecutor().execute(new Runnable() {
		@Override
		public void run() {
			HttpHeaders httpHeaders = transportRequest.getHttpRequestHeaders();
			XhrRequestCallback requestCallback = new XhrRequestCallback(handshakeHeaders);
			XhrRequestCallback requestCallbackAfterHandshake = new XhrRequestCallback(httpHeaders);
			XhrReceiveExtractor responseExtractor = new XhrReceiveExtractor(session);
			while (true) {
				if (session.isDisconnected()) {
					session.afterTransportClosed(null);
					break;
				}
				try {
					if (logger.isTraceEnabled()) {
						logger.trace("Starting XHR receive request, url=" + receiveUrl);
					}
					getRestTemplate().execute(receiveUrl, HttpMethod.POST, requestCallback, responseExtractor);
					requestCallback = requestCallbackAfterHandshake;
				}
				catch (Throwable ex) {
					if (!connectFuture.isDone()) {
						connectFuture.setException(ex);
					}
					else {
						session.handleTransportError(ex);
						session.afterTransportClosed(new CloseStatus(1006, ex.getMessage()));
					}
					break;
				}
			}
		}
	});
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
@Override
protected void connectInternal(TransportRequest request, WebSocketHandler handler, URI receiveUrl,
		HttpHeaders handshakeHeaders, XhrClientSockJsSession session,
		SettableListenableFuture<WebSocketSession> connectFuture) {

	executeReceiveRequest(request, receiveUrl, handshakeHeaders, session, connectFuture);
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
private void executeReceiveRequest(final TransportRequest transportRequest,
		final URI url, final HttpHeaders headers, final XhrClientSockJsSession session,
		final SettableListenableFuture<WebSocketSession> connectFuture) {

	if (logger.isTraceEnabled()) {
		logger.trace("Starting XHR receive request for " + url);
	}

	ClientCallback<ClientConnection> clientCallback = new ClientCallback<ClientConnection>() {
		@Override
		public void completed(ClientConnection connection) {
			ClientRequest request = new ClientRequest().setMethod(Methods.POST).setPath(url.getPath());
			HttpString headerName = HttpString.tryFromString(HttpHeaders.HOST);
			request.getRequestHeaders().add(headerName, url.getHost());
			addHttpHeaders(request, headers);
			HttpHeaders httpHeaders = transportRequest.getHttpRequestHeaders();
			connection.sendRequest(request, createReceiveCallback(transportRequest,
					url, httpHeaders, session, connectFuture));
		}

		@Override
		public void failed(IOException ex) {
			throw new SockJsTransportFailureException("Failed to execute request to " + url, ex);
		}
	};

	this.undertowBufferSupport.httpClientConnect(this.httpClient, clientCallback, url, worker, this.optionMap);
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
public SockJsResponseListener(TransportRequest request, ClientConnection connection, URI url,
		HttpHeaders headers, XhrClientSockJsSession sockJsSession,
		SettableListenableFuture<WebSocketSession> connectFuture) {

	this.request = request;
	this.connection = connection;
	this.url = url;
	this.headers = headers;
	this.session = sockJsSession;
	this.connectFuture = connectFuture;
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
public XhrClientSockJsSession(TransportRequest request, WebSocketHandler handler,
		XhrTransport transport, SettableListenableFuture<WebSocketSession> connectFuture) {

	super(request, handler, connectFuture);
	Assert.notNull(transport, "'restTemplate' is required");
	this.transport = transport;
	this.headers = request.getHttpRequestHeaders();
	this.sendHeaders = new HttpHeaders();
	if (this.headers != null) {
		this.sendHeaders.putAll(this.headers);
	}
	this.sendHeaders.setContentType(MediaType.APPLICATION_JSON);
	this.sendUrl = request.getSockJsUrlInfo().getTransportUrl(TransportType.XHR_SEND);
}
 

import org.springframework.util.concurrent.SettableListenableFuture; //导入依赖的package包/类
public void connect(WebSocketHandler handler, SettableListenableFuture<WebSocketSession> future) {
	if (logger.isTraceEnabled()) {
		logger.trace("Starting " + this);
	}
	ConnectCallback connectCallback = new ConnectCallback(handler, future);
	scheduleConnectTimeoutTask(connectCallback);
	this.transport.connect(this, handler).addCallback(connectCallback);
}