class DefaultWireConnection implements WireConnection {

  public void messageReceived(RLPxMessage message) {
    if (message.messageId() == 0) {
      peerHelloMessage = HelloMessage.read(message.content());
      logger.debug("Received peer Hello message {}", peerHelloMessage);
      initSupportedRange(peerHelloMessage.capabilities());
      if (peerHelloMessage.nodeId() == null || peerHelloMessage.nodeId().isEmpty()) {
        disconnect(DisconnectReason.NULL_NODE_IDENTITY_RECEIVED);
        ready.cancel();
        return;
      }

      if (!peerHelloMessage.nodeId().equals(peerNodeId)) {
        disconnect(DisconnectReason.UNEXPECTED_IDENTITY);
        ready.cancel();
        return;
      }

      if (peerHelloMessage.nodeId().equals(nodeId)) {
        disconnect(DisconnectReason.CONNECTED_TO_SELF);
        ready.cancel();
        return;
      }

      if (peerHelloMessage.p2pVersion() > p2pVersion) {
        disconnect(DisconnectReason.INCOMPATIBLE_DEVP2P_VERSION);
        ready.cancel();
        return;
      }

      if (myHelloMessage == null) {
        sendHello();
      }

      afterHandshakeListener.accept(peerHelloMessage);
      if (subprotocolRangeMap.asMapOfRanges().isEmpty()) {
        ready.complete(this);
      }

      AsyncCompletion allSubProtocols = AsyncCompletion
          .allOf(
              subprotocolRangeMap
                  .asMapOfRanges()
                  .values()
                  .stream()
                  .map(subprotocols::get)
                  .map(handler -> handler.handleNewPeerConnection(this)));
      allSubProtocols.thenRun(() -> ready.complete(this));
      return;
    } else if (message.messageId() == 1) {
      DisconnectMessage disconnect = DisconnectMessage.read(message.content());
      logger.info("Received disconnect {}", disconnect);
      disconnectHandler.run();
      disconnectReceived = true;
      disconnectReason = DisconnectReason.valueOf(disconnect.reason());
      if (!ready.isDone()) {
        ready.cancel();
      }
      return;
    }

    if (peerHelloMessage == null || myHelloMessage == null) {
      logger.debug("Message sent before hello exchanged {}", message.messageId());
      disconnect(DisconnectReason.PROTOCOL_BREACH);
    }

    if (message.messageId() == 2) {
      sendPong();
    } else if (message.messageId() == 3) {
      if (awaitingPong != null) {
        awaitingPong.complete();
      }
    } else {
      Map.Entry<Range<Integer>, SubProtocol> subProtocolEntry = subprotocolRangeMap.getEntry(message.messageId());
      if (subProtocolEntry == null) {
        logger.debug("Unknown message received {}", message.messageId());
        disconnect(DisconnectReason.PROTOCOL_BREACH);
        if (!ready.isDone()) {
          ready.cancel();
        }
      } else {
        int offset = subProtocolEntry.getKey().lowerEndpoint();
        logger.trace("Received message of type {}", message.messageId() - offset);
        SubProtocolHandler handler = subprotocols.get(subProtocolEntry.getValue());
        try {
          handler.handle(this, message.messageId() - offset, message.content());
        } catch (Throwable t) {
          logger.error("Handler " + handler.toString() + " threw an exception", t);
        }
      }
    }
  }

  public  DefaultWireConnection(
      Bytes nodeId,
      Bytes peerNodeId,
      Consumer<RLPxMessage> writer,
      Consumer<HelloMessage> afterHandshakeListener,
      Runnable disconnectHandler,
      LinkedHashMap<SubProtocol, SubProtocolHandler> subprotocols,
      int p2pVersion,
      String clientId,
      int advertisedPort,
      CompletableAsyncResult<WireConnection> ready,
      String peerHost,
      int peerPort);

}