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

@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);
}

@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;
}

@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;
}

@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;
}

@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;
}

@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;
}

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;
}

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;
}

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;
}

@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;
}

@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");

}

@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;
}

@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);
}

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;
}

@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;
                }
            }
        }
    });
}

@Override
protected void connectInternal(TransportRequest request, WebSocketHandler handler, URI receiveUrl,
        HttpHeaders handshakeHeaders, XhrClientSockJsSession session,
        SettableListenableFuture<WebSocketSession> connectFuture) {

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

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);
}

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;
}

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);
}

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);
}

@Override
protected void connectInternal(TransportRequest transportRequest, WebSocketHandler handler,
        URI url, HttpHeaders handshakeHeaders, XhrClientSockJsSession session,
        SettableListenableFuture<WebSocketSession> connectFuture) {

    HttpHeaders httpHeaders = transportRequest.getHttpRequestHeaders();
    SockJsResponseListener listener = new SockJsResponseListener(url, httpHeaders, session, connectFuture);
    executeReceiveRequest(url, handshakeHeaders, listener);
}

public SockJsResponseListener(URI url, HttpHeaders headers, XhrClientSockJsSession sockJsSession,
        SettableListenableFuture<WebSocketSession> connectFuture) {

    this.transportUrl = url;
    this.receiveHeaders = headers;
    this.connectFuture = connectFuture;
    this.sockJsSession = sockJsSession;
}

@Override
protected void connectInternal(TransportRequest request, WebSocketHandler handler, URI receiveUrl,
        HttpHeaders handshakeHeaders, XhrClientSockJsSession session,
        SettableListenableFuture<WebSocketSession> connectFuture) {

    this.actualHandshakeHeaders = handshakeHeaders;
    this.actualSession = session;
}

@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");
}

@Before
public void setup() throws Exception {
    SockJsUrlInfo urlInfo = new SockJsUrlInfo(new URI("http://example.com"));
    Transport transport = mock(Transport.class);
    TransportRequest request = new DefaultTransportRequest(urlInfo, null, null, transport, TransportType.XHR, CODEC);
    this.handler = mock(WebSocketHandler.class);
    this.connectFuture = new SettableListenableFuture<>();
    this.session = new TestClientSockJsSession(request, this.handler, this.connectFuture);
}

@Before
public void setUp() throws Exception {
    MockitoAnnotations.initMocks(this);

    WebSocketClient webSocketClient = mock(WebSocketClient.class);
    this.stompClient = new TestWebSocketStompClient(webSocketClient);
    this.stompClient.setTaskScheduler(this.taskScheduler);
    this.stompClient.setStompSession(this.stompSession);

    this.webSocketHandlerCaptor = ArgumentCaptor.forClass(WebSocketHandler.class);
    this.handshakeFuture = new SettableListenableFuture<>();
    when(webSocketClient.doHandshake(this.webSocketHandlerCaptor.capture(), any(), any(URI.class)))
            .thenReturn(this.handshakeFuture);
}

@Before
public void beforeMethod() throws IOException {
    resetTracing();

    spanRecorder = new SpanRecorder();
    Tracer.getInstance().addSpanLifecycleListener(spanRecorder);

    method = HttpMethod.PATCH;
    uri = URI.create("http://localhost:4242/" + UUID.randomUUID().toString());
    httpRequest = new HttpRequest() {
        @Override
        public HttpHeaders getHeaders() { return new HttpHeaders(); }

        @Override
        public HttpMethod getMethod() {
            return method;
        }

        @Override
        public URI getURI() {
            return uri;
        }
    };

    body = UUID.randomUUID().toString().getBytes();
    executionMock = mock(AsyncClientHttpRequestExecution.class);
    doAnswer(invocation -> {
        tracingStateAtTimeOfExecution = TracingState.getCurrentThreadTracingState();
        executionResponseFuture = new SettableListenableFuture<>();
        return executionResponseFuture;
    }).when(executionMock).executeAsync(any(HttpRequest.class), any(byte[].class));
}

public ListenableFuture<List<BuildStep>> getProjectBuildSteps(String projectId) {
  SettableListenableFuture<List<BuildStep>> returnFuture = new SettableListenableFuture<>();
  ListenableFuture<Project> project = getProject(projectId);
  project.addCallback(
      result -> returnFuture.set(getBuildSteps(result)),
      returnFuture::setException);

  return returnFuture;
}

public ConnectCallback(WebSocketHandler handler, SettableListenableFuture<WebSocketSession> future) {
    this.handler = handler;
    this.future = future;
}

public WebSocketClientSockJsSession(TransportRequest request, WebSocketHandler handler,
        SettableListenableFuture<WebSocketSession> connectFuture) {

    super(request, handler, connectFuture);
}

protected TestClientSockJsSession(TransportRequest request, WebSocketHandler handler,
        SettableListenableFuture<WebSocketSession> connectFuture) {
    super(request, handler, connectFuture);
}

public RequestExecuteHandler(SettableListenableFuture<ClientHttpResponse> responseFuture) {
    this.responseFuture = responseFuture;
}

@Override
public ListenableFuture<ClientHttpResponse> executeAsync() throws IOException {
    SettableListenableFuture<ClientHttpResponse> future = new SettableListenableFuture<ClientHttpResponse>();
    future.set(execute());
    return future;
}

NettyResponseHandler(SettableListenableFuture<ClientHttpResponse> responseFuture) {
    this.responseFuture = responseFuture;
}

public ListenableFuture<List<Build>> getProjectBuilds(String projectId) {
  final SettableListenableFuture<List<Build>> projectBuilds = new SettableListenableFuture<>();

  ListenableFuture<Project> project = getProject(projectId);
  project.addCallback(result -> populateBuilds(getFirstBuildStep(result), projectBuilds),
      projectBuilds::setException);
  return projectBuilds;
}

private void populateBuilds(BuildStep buildStep, SettableListenableFuture<List<Build>> returnFuture) {
  fetchBuildsForStep(buildStep).addCallback(
      result -> returnFuture.set(result.getBuild()),
      returnFuture::setException);
}

private ValueSourceBackedListenableFuture(ValueSource<T> valueSource) {
    super(new SettableListenableFuture<>());
    this.valueSource = valueSource;
    valueSource.addCallbacks(value -> getWrappedFuture().set(value), ex -> getWrappedFuture().setException(ex));
}

@Override
protected SettableListenableFuture<T> getWrappedFuture() {
    return (SettableListenableFuture<T>) super.getWrappedFuture();
}

@Override
public ListenableFuture<String> createFinishedFuture() {
    SettableListenableFuture<String> future = new SettableListenableFuture<>();
    future.set(AbstractConverterTest.VALUE);
    return future;
}

protected abstract void connectInternal(TransportRequest request, WebSocketHandler handler,
URI receiveUrl, HttpHeaders handshakeHeaders, XhrClientSockJsSession session,
SettableListenableFuture<WebSocketSession> connectFuture);