class DeFramer extends ByteToMessageDecoder {

  @Override
  protected void decode(final ChannelHandlerContext ctx, final ByteBuf in, final List<Object> out) {
    MessageData message;
    while ((message = framer.deframe(in)) != null) {

      if (hellosExchanged) {
        out.add(message);
      } else if (message.getCode() == WireMessageCodes.HELLO) {
        hellosExchanged = true;
        // Decode first hello and use the payload to modify pipeline
        final PeerInfo peerInfo;
        try {
          peerInfo = HelloMessage.readFrom(message).getPeerInfo();
        } catch (final RLPException e) {
          LOG.debug("Received invalid HELLO message", e);
          connectFuture.completeExceptionally(e);
          ctx.close();
          return;
        }
        LOG.debug("Received HELLO message: {}", peerInfo);
        if (peerInfo.getVersion() >= 5) {
          LOG.debug("Enable compression for p2pVersion: {}", peerInfo.getVersion());
          framer.enableCompression();
        }

        final CapabilityMultiplexer capabilityMultiplexer =
            new CapabilityMultiplexer(
                subProtocols,
                localNode.getPeerInfo().getCapabilities(),
                peerInfo.getCapabilities());
        final Peer peer = expectedPeer.orElse(createPeer(peerInfo, ctx));
        final PeerConnection connection =
            new NettyPeerConnection(
                ctx,
                peer,
                peerInfo,
                capabilityMultiplexer,
                connectionEventDispatcher,
                outboundMessagesCounter);

        // Check peer is who we expected
        if (expectedPeer.isPresent()
            && !Objects.equals(expectedPeer.get().getId(), peerInfo.getNodeId())) {
          String unexpectedMsg =
              String.format(
                  "Expected id %s, but got %s", expectedPeer.get().getId(), peerInfo.getNodeId());
          connectFuture.completeExceptionally(new UnexpectedPeerConnectionException(unexpectedMsg));
          LOG.debug("{}. Disconnecting.", unexpectedMsg);
          connection.disconnect(DisconnectMessage.DisconnectReason.UNEXPECTED_ID);
        }

        // Check that we have shared caps
        if (capabilityMultiplexer.getAgreedCapabilities().size() == 0) {
          LOG.debug("Disconnecting because no capabilities are shared: {}", peerInfo);
          connectFuture.completeExceptionally(
              new IncompatiblePeerException("No shared capabilities"));
          connection.disconnect(DisconnectMessage.DisconnectReason.USELESS_PEER);
        }

        // Setup next stage
        final AtomicBoolean waitingForPong = new AtomicBoolean(false);
        ctx.channel()
            .pipeline()
            .addLast(
                new IdleStateHandler(15, 0, 0),
                new WireKeepAlive(connection, waitingForPong),
                new ApiHandler(
                    capabilityMultiplexer, connection, connectionEventDispatcher, waitingForPong),
                new MessageFramer(capabilityMultiplexer, framer));
        connectFuture.complete(connection);
      } else if (message.getCode() == WireMessageCodes.DISCONNECT) {
        DisconnectMessage disconnectMessage = DisconnectMessage.readFrom(message);
        LOG.debug(
            "Peer disconnected before sending HELLO.  Reason: " + disconnectMessage.getReason());
        ctx.close();
        connectFuture.completeExceptionally(
            new PeerDisconnectedException(disconnectMessage.getReason()));
      } else {
        // Unexpected message - disconnect
        LOG.debug(
            "Message received before HELLO's exchanged, disconnecting.  Code: {}, Data: {}",
            message.getCode(),
            message.getData().toString());
        ctx.writeAndFlush(
                new OutboundMessage(
                    null,
                    DisconnectMessage.create(
                        DisconnectMessage.DisconnectReason.BREACH_OF_PROTOCOL)))
            .addListener((f) -> ctx.close());
        connectFuture.completeExceptionally(
            new BreachOfProtocolException("Message received before HELLO's exchanged"));
      }
    }
  }

    DeFramer(
      final Framer framer,
      final List<SubProtocol> subProtocols,
      final LocalNode localNode,
      final Optional<Peer> expectedPeer,
      final PeerConnectionEventDispatcher connectionEventDispatcher,
      final CompletableFuture<PeerConnection> connectFuture,
      final MetricsSystem metricsSystem);

}