@Test void sendRequest_positiveCase() { SafeFuture<RpcStream> future = rpcHandler.sendRequest(connection, Bytes.fromHexString("0x11223344"), rpcRequestHandler); assertThat(future).isNotDone(); streamPromise.getStream().complete(stream); assertThat(future).isNotDone(); streamPromise.getController().complete(controller); assertThat(future).isNotDone(); writeFuture.complete(null); verify(stream, never()).close(); verify(controller, never()).close(); assertThat(future).isCompletedWithValue(rpcStream); assertThat(closeFuture.getNumberOfDependents()).isEqualTo(0); closeFuture.complete(null); verify(stream, never()).close(); verify(controller, never()).close(); assertThat(future).isCompletedWithValue(rpcStream); assertThat(closeFuture.getNumberOfDependents()).isEqualTo(0); }
@SuppressWarnings("unchecked") public SafeFuture<RpcStream> sendRequest( Connection connection, Bytes initialPayload, RpcRequestHandler handler) { Interruptor closeInterruptor = SafeFuture.createInterruptor(connection.closeFuture(), PeerDisconnectedException::new); Interruptor timeoutInterruptor = SafeFuture.createInterruptor( asyncRunner.getDelayedFuture(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), () -> new StreamTimeoutException( "Timed out waiting to initialize stream for method " + rpcMethod.getId())); return SafeFuture.notInterrupted(closeInterruptor) .thenApply( __ -> connection.muxerSession().createStream(Multistream.create(this).toStreamHandler())) .thenWaitFor(StreamPromise::getStream) .orInterrupt(closeInterruptor, timeoutInterruptor) .thenCompose( streamPromise -> SafeFuture.of(streamPromise.getController()) .orInterrupt(closeInterruptor, timeoutInterruptor) .thenPeek(ctr -> ctr.setRequestHandler(handler)) .thenApply(Controller::getRpcStream) .thenWaitFor(rpcStream -> rpcStream.writeBytes(initialPayload)) .orInterrupt(closeInterruptor, timeoutInterruptor) .whenException(err -> closeStreamAbruptly(streamPromise.getStream().join()))); }
RpcHandler implements ProtocolBinding<Controller> { @SuppressWarnings("unchecked") public SafeFuture<RpcStream> sendRequest( Connection connection, Bytes initialPayload, RpcRequestHandler handler) { Interruptor closeInterruptor = SafeFuture.createInterruptor(connection.closeFuture(), PeerDisconnectedException::new); Interruptor timeoutInterruptor = SafeFuture.createInterruptor( asyncRunner.getDelayedFuture(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), () -> new StreamTimeoutException( "Timed out waiting to initialize stream for method " + rpcMethod.getId())); return SafeFuture.notInterrupted(closeInterruptor) .thenApply( __ -> connection.muxerSession().createStream(Multistream.create(this).toStreamHandler())) .thenWaitFor(StreamPromise::getStream) .orInterrupt(closeInterruptor, timeoutInterruptor) .thenCompose( streamPromise -> SafeFuture.of(streamPromise.getController()) .orInterrupt(closeInterruptor, timeoutInterruptor) .thenPeek(ctr -> ctr.setRequestHandler(handler)) .thenApply(Controller::getRpcStream) .thenWaitFor(rpcStream -> rpcStream.writeBytes(initialPayload)) .orInterrupt(closeInterruptor, timeoutInterruptor) .whenException(err -> closeStreamAbruptly(streamPromise.getStream().join()))); } }
RpcHandler implements ProtocolBinding<Controller> { @SuppressWarnings("unchecked") public SafeFuture<RpcStream> sendRequest( Connection connection, Bytes initialPayload, RpcRequestHandler handler) { Interruptor closeInterruptor = SafeFuture.createInterruptor(connection.closeFuture(), PeerDisconnectedException::new); Interruptor timeoutInterruptor = SafeFuture.createInterruptor( asyncRunner.getDelayedFuture(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), () -> new StreamTimeoutException( "Timed out waiting to initialize stream for method " + rpcMethod.getId())); return SafeFuture.notInterrupted(closeInterruptor) .thenApply( __ -> connection.muxerSession().createStream(Multistream.create(this).toStreamHandler())) .thenWaitFor(StreamPromise::getStream) .orInterrupt(closeInterruptor, timeoutInterruptor) .thenCompose( streamPromise -> SafeFuture.of(streamPromise.getController()) .orInterrupt(closeInterruptor, timeoutInterruptor) .thenPeek(ctr -> ctr.setRequestHandler(handler)) .thenApply(Controller::getRpcStream) .thenWaitFor(rpcStream -> rpcStream.writeBytes(initialPayload)) .orInterrupt(closeInterruptor, timeoutInterruptor) .whenException(err -> closeStreamAbruptly(streamPromise.getStream().join()))); } RpcHandler(final AsyncRunner asyncRunner, RpcMethod rpcMethod); }
RpcHandler implements ProtocolBinding<Controller> { @SuppressWarnings("unchecked") public SafeFuture<RpcStream> sendRequest( Connection connection, Bytes initialPayload, RpcRequestHandler handler) { Interruptor closeInterruptor = SafeFuture.createInterruptor(connection.closeFuture(), PeerDisconnectedException::new); Interruptor timeoutInterruptor = SafeFuture.createInterruptor( asyncRunner.getDelayedFuture(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), () -> new StreamTimeoutException( "Timed out waiting to initialize stream for method " + rpcMethod.getId())); return SafeFuture.notInterrupted(closeInterruptor) .thenApply( __ -> connection.muxerSession().createStream(Multistream.create(this).toStreamHandler())) .thenWaitFor(StreamPromise::getStream) .orInterrupt(closeInterruptor, timeoutInterruptor) .thenCompose( streamPromise -> SafeFuture.of(streamPromise.getController()) .orInterrupt(closeInterruptor, timeoutInterruptor) .thenPeek(ctr -> ctr.setRequestHandler(handler)) .thenApply(Controller::getRpcStream) .thenWaitFor(rpcStream -> rpcStream.writeBytes(initialPayload)) .orInterrupt(closeInterruptor, timeoutInterruptor) .whenException(err -> closeStreamAbruptly(streamPromise.getStream().join()))); } RpcHandler(final AsyncRunner asyncRunner, RpcMethod rpcMethod); @SuppressWarnings("unchecked") SafeFuture<RpcStream> sendRequest(
      Connection connection, Bytes initialPayload, RpcRequestHandler handler); @NotNull @Override ProtocolDescriptor getProtocolDescriptor(); @NotNull @Override SafeFuture<Controller> initChannel(P2PChannel channel, String s); }
RpcHandler implements ProtocolBinding<Controller> { @SuppressWarnings("unchecked") public SafeFuture<RpcStream> sendRequest( Connection connection, Bytes initialPayload, RpcRequestHandler handler) { Interruptor closeInterruptor = SafeFuture.createInterruptor(connection.closeFuture(), PeerDisconnectedException::new); Interruptor timeoutInterruptor = SafeFuture.createInterruptor( asyncRunner.getDelayedFuture(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), () -> new StreamTimeoutException( "Timed out waiting to initialize stream for method " + rpcMethod.getId())); return SafeFuture.notInterrupted(closeInterruptor) .thenApply( __ -> connection.muxerSession().createStream(Multistream.create(this).toStreamHandler())) .thenWaitFor(StreamPromise::getStream) .orInterrupt(closeInterruptor, timeoutInterruptor) .thenCompose( streamPromise -> SafeFuture.of(streamPromise.getController()) .orInterrupt(closeInterruptor, timeoutInterruptor) .thenPeek(ctr -> ctr.setRequestHandler(handler)) .thenApply(Controller::getRpcStream) .thenWaitFor(rpcStream -> rpcStream.writeBytes(initialPayload)) .orInterrupt(closeInterruptor, timeoutInterruptor) .whenException(err -> closeStreamAbruptly(streamPromise.getStream().join()))); } RpcHandler(final AsyncRunner asyncRunner, RpcMethod rpcMethod); @SuppressWarnings("unchecked") SafeFuture<RpcStream> sendRequest(
      Connection connection, Bytes initialPayload, RpcRequestHandler handler); @NotNull @Override ProtocolDescriptor getProtocolDescriptor(); @NotNull @Override SafeFuture<Controller> initChannel(P2PChannel channel, String s);  }