### Question:
BeaconBlocksByRootMessageHandler extends PeerRequiredLocalMessageHandler<BeaconBlocksByRootRequestMessage, SignedBeaconBlock> { @Override public void onIncomingMessage( final Eth2Peer peer, final BeaconBlocksByRootRequestMessage message, final ResponseCallback<SignedBeaconBlock> callback) { LOG.trace( "Peer {} requested BeaconBlocks with roots: {}", peer.getId(), message.getBlockRoots()); if (storageClient.getStore() != null) { SafeFuture<Void> future = SafeFuture.COMPLETE; if (!peer.wantToMakeRequest() || !peer.wantToReceiveObjects(callback, message.getBlockRoots().size())) { peer.disconnectCleanly(DisconnectReason.RATE_LIMITING).reportExceptions(); return; } for (Bytes32 blockRoot : message.getBlockRoots()) { future = future.thenCompose( __ -> storageClient .getStore() .retrieveSignedBlock(blockRoot) .thenCompose( block -> block.map(callback::respond).orElse(SafeFuture.COMPLETE))); } future.finish(callback::completeSuccessfully, callback::completeWithUnexpectedError); } else { callback.completeSuccessfully(); } } BeaconBlocksByRootMessageHandler(final RecentChainData storageClient); @Override void onIncomingMessage(
      final Eth2Peer peer,
      final BeaconBlocksByRootRequestMessage message,
      final ResponseCallback<SignedBeaconBlock> callback);  }### Answer:
@Test public void onIncomingMessage_respondsWithAllBlocks() { final List<SignedBeaconBlock> blocks = mockChain(5); final BeaconBlocksByRootRequestMessage message = createRequest(blocks); handler.onIncomingMessage(peer, message, callback); for (SignedBeaconBlock block : blocks) { verify(store).retrieveSignedBlock(block.getRoot()); verify(callback).respond(block); } }
@Test public void onIncomingMessage_interruptedByClosedStream() { final List<SignedBeaconBlock> blocks = mockChain(5); doThrow(new StreamClosedException()).when(callback).respond(any()); final BeaconBlocksByRootRequestMessage message = createRequest(blocks); handler.onIncomingMessage(peer, message, callback); verify(store, times(1)).retrieveSignedBlock(any()); verify(callback, times(1)).respond(any()); }