### Question:
Eth2IncomingRequestHandler implements RpcRequestHandler { @Override public void complete(NodeId nodeId, RpcStream rpcStream) { try { Optional<Eth2Peer> peer = peerLookup.getConnectedPeer(nodeId); requestDecoder .complete() .ifPresent( request -> handleRequest(peer, request, new RpcResponseCallback<>(rpcStream, rpcEncoder))); ; } catch (RpcException e) { new RpcResponseCallback<>(rpcStream, rpcEncoder).completeWithErrorResponse(e); LOG.debug("RPC Request stream closed prematurely", e); } } Eth2IncomingRequestHandler(
      final AsyncRunner asyncRunner,
      final Eth2RpcMethod<TRequest, TResponse> method,
      final PeerLookup peerLookup,
      final LocalMessageHandler<TRequest, TResponse> localMessageHandler); @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); @Override String toString();  }### Answer:
@Test public void testEmptyRequestMessage() { Eth2IncomingRequestHandler<EmptyMessage, MetadataMessage> requestHandler = beaconChainMethods.getMetadata().createIncomingRequestHandler(); requestHandler.complete(nodeId, rpcStream); asyncRunner.executeQueuedActions(); verify(rpcStream).writeBytes(argThat(bytes -> bytes.get(0) == 0)); verify(rpcStream).close(); }