### Question:
EthPeer { public RequestManager.ResponseStream getHeadersByHash( final Hash hash, final int maxHeaders, final int skip, final boolean reverse) throws PeerNotConnected { final GetBlockHeadersMessage message = GetBlockHeadersMessage.create(hash, maxHeaders, skip, reverse); return sendRequest(headersRequestManager, message); } EthPeer(
      final PeerConnection connection,
      final String protocolName,
      final Consumer<EthPeer> onStatusesExchanged,
      final List<PeerValidator> peerValidators,
      final Clock clock); void markValidated(final PeerValidator validator); boolean isFullyValidated(); boolean isDisconnected(); long addChainEstimatedHeightListener(final ChainState.EstimatedHeightListener listener); void removeChainEstimatedHeightListener(final long listenerId); void recordRequestTimeout(final int requestCode); void recordUselessResponse(final String requestType); void disconnect(final DisconnectReason reason); RequestManager.ResponseStream send(final MessageData messageData); RequestManager.ResponseStream getHeadersByHash(
      final Hash hash, final int maxHeaders, final int skip, final boolean reverse); RequestManager.ResponseStream getHeadersByNumber(
      final long blockNumber, final int maxHeaders, final int skip, final boolean reverse); RequestManager.ResponseStream getBodies(final List<Hash> blockHashes); RequestManager.ResponseStream getReceipts(final List<Hash> blockHashes); RequestManager.ResponseStream getNodeData(final Iterable<Hash> nodeHashes); RequestManager.ResponseStream getPooledTransactions(final List<Hash> hashes); Map<Integer, AtomicInteger> timeoutCounts(); void registerKnownBlock(final Hash hash); void registerStatusSent(); void registerStatusReceived(final Hash hash, final Difficulty td); boolean readyForRequests(); boolean statusHasBeenReceived(); boolean statusHasBeenSentToPeer(); boolean hasSeenBlock(final Hash hash); ChainState chainState(); ChainHeadEstimate chainStateSnapshot(); void registerHeight(final Hash blockHash, final long height); int outstandingRequests(); long getLastRequestTimestamp(); boolean hasAvailableRequestCapacity(); Set<Capability> getAgreedCapabilities(); Bytes nodeId(); @Override String toString();  }### Answer:
@Test public void getHeadersStream() throws PeerNotConnected { final ResponseStreamSupplier getStream = (peer) -> peer.getHeadersByHash(gen.hash(), 5, 0, false); final MessageData targetMessage = BlockHeadersMessage.create(asList(gen.header(), gen.header())); final MessageData otherMessage = BlockBodiesMessage.create(asList(gen.body(), gen.body())); messageStream(getStream, targetMessage, otherMessage); }