### Question:
Eth2OutgoingRequestHandler implements RpcRequestHandler { @Override public void complete(NodeId nodeId, RpcStream rpcStream) { try { responseDecoder.complete(); completeRequest(rpcStream); } catch (RpcException e) { cancelRequest(rpcStream, e); } } Eth2OutgoingRequestHandler(
      final AsyncRunner asyncRunner,
      final AsyncRunner timeoutRunner,
      final Eth2RpcMethod<TRequest, TResponse> method,
      final int maximumResponseChunks); void handleInitialPayloadSent(final RpcStream stream); @Override void active(NodeId nodeId, RpcStream rpcStream); @Override void processData(final NodeId nodeId, final RpcStream rpcStream, final ByteBuf data); @Override void complete(NodeId nodeId, RpcStream rpcStream); ResponseStreamImpl<TResponse> getResponseStream(); @Override String toString();  }### Answer:
@Test public void processAllExpectedChunks() throws Exception { sendInitialPayload(); verify(rpcStream).closeWriteStream(); for (int i = 0; i < maxChunks; i++) { deliverChunk(i); assertThat(finishedProcessingFuture).isNotDone(); } complete(); asyncRequestRunner.waitForExactly(maxChunks - 1); assertThat(finishedProcessingFuture).isNotDone(); asyncRequestRunner.waitForExactly(1); timeoutRunner.executeUntilDone(); Waiter.waitFor(() -> assertThat(finishedProcessingFuture).isDone()); assertThat(finishedProcessingFuture).isCompletedWithValue(null); verify(rpcStream).close(); assertThat(blocks.size()).isEqualTo(3); }
@Test public void receiveAllChunksThenEncounterErrorWhileProcessing() throws Exception { sendInitialPayload(); verify(rpcStream).closeWriteStream(); for (int i = 0; i < maxChunks; i++) { deliverChunk(i); assertThat(finishedProcessingFuture).isNotDone(); } complete(); asyncRequestRunner.waitForExactly(maxChunks - 1); assertThat(finishedProcessingFuture).isNotDone(); final RuntimeException error = new RuntimeException("whoops"); responseListener.set( b -> { throw error; }); asyncRequestRunner.waitForExactly(1); timeoutRunner.executeUntilDone(); Waiter.waitFor(() -> assertThat(finishedProcessingFuture).isDone()); assertThat(finishedProcessingFuture).isCompletedExceptionally(); verify(rpcStream, atLeastOnce()).close(); assertThat(blocks.size()).isEqualTo(maxChunks - 1); assertThatThrownBy(finishedProcessingFuture::get).hasRootCause(error); }
@Test public void processAllValidChunksWhenErrorIsEncountered() throws Exception { sendInitialPayload(); verify(rpcStream).closeWriteStream(); deliverChunk(0); assertThat(finishedProcessingFuture).isNotDone(); deliverError(); complete(); asyncRequestRunner.waitForExactly(1); Waiter.waitFor(() -> assertThat(finishedProcessingFuture).isDone()); verify(rpcStream).close(); assertThat(blocks.size()).isEqualTo(1); assertThat(finishedProcessingFuture).isCompletedExceptionally(); }
@Test public void sendTooManyChunks() throws Exception { sendInitialPayload(); verify(rpcStream).closeWriteStream(); for (int i = 0; i < maxChunks; i++) { if (i == maxChunks - 1) { final Bytes lastChunk = chunks.get(i); final Bytes lastChunkWithExtraChunk = Bytes.concatenate(lastChunk, lastChunk); deliverBytes(lastChunkWithExtraChunk); } else { deliverChunk(i); } } complete(); asyncRequestRunner.waitForExactly(maxChunks - 1); timeoutRunner.executeUntilDone(); Waiter.waitFor(() -> assertThat(finishedProcessingFuture).isDone()); verify(rpcStream).close(); assertThat(blocks.size()).isEqualTo(2); assertThat(finishedProcessingFuture).isCompletedExceptionally(); assertThatThrownBy(finishedProcessingFuture::get) .hasRootCause(new ExtraDataAppendedException()); }