import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> out) throws Exception {
    in.resetReaderIndex();
    //Common decoding part
    ConnAckMessage message = new ConnAckMessage();
    if (!decodeCommonHeader(message, 0x00, in)) {
        in.resetReaderIndex();
        return;
    }
    //skip reserved byte
    in.skipBytes(1);
    
    //read  return code
    message.setReturnCode(in.readByte());
    out.add(message);
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
private static TransportException toTransportException(
    Throwable cause, AttributeMap channelAttrs) {
  String service = channelAttrs.attr(ChannelAttributes.SERVICE).get();
  String procedure = channelAttrs.attr(ChannelAttributes.PROCEDURE).get();
  if (cause instanceof DeadlineExceededException) {
    // A DeadlineExceededException on inbound translates to a TimeoutException on the caller
    String caller = channelAttrs.attr(ChannelAttributes.CALLER).get();
    Instant start = channelAttrs.attr(ChannelAttributes.REQUEST_START).get();
    Instant end = ((DeadlineExceededException) cause).getTimeExceeded();
    long timeSpent = start != null ? Duration.between(start, end).toMillis() : -1;
    return new TimeoutException(service, procedure, caller, timeSpent);
  }
  if (isRemoteException(cause)) {
    // Remote exceptions do not bubble up to the caller
    return UnexpectedException.wrap(cause, service, procedure);
  }
  if (cause instanceof TransportException) {
    return (TransportException) cause;
  }
  log.warn(
      "Caught unexpected error in procedure \"{}\" of service \"{}\"", procedure, service, cause);
  return UnexpectedException.wrap(cause, service, procedure);
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> out) throws Exception {
    in.resetReaderIndex();
    //Common decoding part
    ConnAckMessage message = new ConnAckMessage();
    if (!decodeCommonHeader(message, 0x00, in)) {
        in.resetReaderIndex();
        return;
    }
    //skip reserved byte
    in.skipBytes(1);

    //read  return code
    message.setReturnCode(in.readByte());
    out.add(message);
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
public SubscribeMessage decode(AttributeMap ctx, ByteBuf in) throws Exception
{
    //Common decoding part
    SubscribeMessage message = new SubscribeMessage();
    in.resetReaderIndex();
    if (!decodeCommonHeader(message, 0x02, in)) {
        in.resetReaderIndex();
        return null;
    }

    //check qos level
    if (message.getQos() != AbstractMessage.QOSType.LEAST_ONE) {
        throw new CorruptedFrameException("Received SUBSCRIBE message with QoS other than LEAST_ONE, was: " + message.getQos());
    }

    int start = in.readerIndex();
    //read  messageIDs
    message.setMessageID(in.readUnsignedShort());
    int read = in.readerIndex() - start;
    while (read < message.getRemainingLength()) {
        decodeSubscription(in, message);
        read = in.readerIndex() - start;
    }

    if (message.subscriptions().isEmpty()) {
        throw new CorruptedFrameException("subscribe MUST have got at least 1 couple topic/QoS");
    }

    return message;
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> out) throws Exception {
    //Common decoding part
    in.resetReaderIndex();
    DisconnectMessage message = new DisconnectMessage();
    if (!decodeCommonHeader(message, 0x00, in)) {
        in.resetReaderIndex();
        return;
    }
    out.add(message);
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> out) throws UnsupportedEncodingException {
    in.resetReaderIndex();
    //Common decoding part
    MessageIDMessage message = new PubRelMessage();
    if (!decodeCommonHeader(message, 0x02, in)) {
        in.resetReaderIndex();
        return;
    }
    
    //read  messageIDs
    message.setMessageID(in.readUnsignedShort());
    out.add(message);
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> out) throws Exception {
    //Common decoding part
    in.resetReaderIndex();
    PingRespMessage message = new PingRespMessage();
    if (!decodeCommonHeader(message, 0x00, in)) {
        in.resetReaderIndex();
        return;
    }
    out.add(message);
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> out) throws Exception {
    //Common decoding part
    in.resetReaderIndex();
    PingReqMessage message = new PingReqMessage();
    if (!decodeCommonHeader(message, 0x00, in)) {
        in.resetReaderIndex();
        return;
    }
    out.add(message);
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> out) throws Exception {
    //Common decoding part
    in.resetReaderIndex();
    UnsubscribeMessage message = new UnsubscribeMessage();
    if (!decodeCommonHeader(message, 0x02, in)) {
        in.resetReaderIndex();
        return;
    }
    
    //check qos level
    if (message.getQos() != AbstractMessage.QOSType.LEAST_ONE) {
        throw new CorruptedFrameException("Found an Unsubscribe message with qos other than LEAST_ONE, was: " + message.getQos());
    }
        
    int start = in.readerIndex();
    //read  messageIDs
    message.setMessageID(in.readUnsignedShort());
    int read = in.readerIndex() - start;
    while (read < message.getRemainingLength()) {
        String topicFilter = Utils.decodeString(in);
        //check topic is at least one char [MQTT-4.7.3-1]
        if (topicFilter.length() == 0) {
            throw new CorruptedFrameException("Received an UNSUBSCRIBE with empty topic filter");
        }
        message.addTopicFilter(topicFilter);
        read = in.readerIndex() - start;
    }
    if (message.topicFilters().isEmpty()) {
        throw new CorruptedFrameException("unsubscribe MUST have got at least 1 topic");
    }
    out.add(message);
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> out) throws Exception {
    //Common decoding part
    SubscribeMessage message = new SubscribeMessage();
    in.resetReaderIndex();
    if (!decodeCommonHeader(message, 0x02, in)) {
        in.resetReaderIndex();
        return;
    }
    
    //check qos level
    if (message.getQos() != QOSType.LEAST_ONE) {
        throw new CorruptedFrameException("Received SUBSCRIBE message with QoS other than LEAST_ONE, was: " + message.getQos());
    }
        
    int start = in.readerIndex();
    //read  messageIDs
    message.setMessageID(in.readUnsignedShort());
    int read = in.readerIndex() - start;
    while (read < message.getRemainingLength()) {
        decodeSubscription(in, message);
        read = in.readerIndex() - start;
    }
    
    if (message.subscriptions().isEmpty()) {
        throw new CorruptedFrameException("subscribe MUST have got at least 1 couple topic/QoS");
    } 
    
    out.add(message);
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
static boolean isMQTT3_1_1(AttributeMap attrsMap) {
    Attribute<Integer> versionAttr = attrsMap.attr(MQTTDecoder.PROTOCOL_VERSION);
    Integer protocolVersion = versionAttr.get();
    if (protocolVersion == null) {
        return true;
    } 
    return protocolVersion == VERSION_3_1_1;
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> out) throws Exception {
    in.resetReaderIndex();
    //Common decoding part
    MessageIDMessage message = createMessage();
    if (!decodeCommonHeader(message, 0x00, in)) {
        in.resetReaderIndex();
        return;
    }
    
    //read  messageIDs
    message.setMessageID(in.readUnsignedShort());
    out.add(message);
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> out) throws Exception {
    //Common decoding part
    in.resetReaderIndex();
    SubAckMessage message = new SubAckMessage();
    if (!decodeCommonHeader(message, 0x00, in)) {
        in.resetReaderIndex();
        return;
    }
    int remainingLength = message.getRemainingLength();
    
    //MessageID
    message.setMessageID(in.readUnsignedShort());
    remainingLength -= 2;
    
    //Qos array
    if (in.readableBytes() < remainingLength ) {
        in.resetReaderIndex();
        return;
    }
    for (int i = 0; i < remainingLength; i++) {
        byte qos = in.readByte();
        message.addType(AbstractMessage.QOSType.valueOf(qos));
    }
    
    out.add(message);
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
public HttpRequest buildStartMessage(TransportRequest request, AttributeMap channelAttrs) {
  DefaultHttpRequest httpRequest =
      new DefaultHttpRequest(HttpTransport.HTTP_VERSION, HttpMethod.POST, url.getPath());
  HttpHeaders httpHeaders = httpRequest.headers();
  setCommonHeaders(httpHeaders, request, channelAttrs);
  httpHeaders.set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
  return httpRequest;
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
public HttpRequest buildFullMessage(
    TransportRequest request, byte[] body, AttributeMap channelAttrs) {
  DefaultHttpRequest httpRequest =
      new DefaultFullHttpRequest(
          HttpTransport.HTTP_VERSION,
          HttpMethod.POST,
          url.getPath(),
          Unpooled.wrappedBuffer(body));
  setCommonHeaders(httpRequest.headers(), request, channelAttrs);
  HttpUtil.setContentLength(httpRequest, body.length);
  return httpRequest;
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
private void setCommonHeaders(
    HttpHeaders httpHeaders, TransportRequest request, AttributeMap channelAttrs) {
  HeaderMapper.toHttpHeaders(request.getHeaders(), httpHeaders);
  httpHeaders.set(HeaderMapper.SERVICE, request.getService());
  httpHeaders.set(HeaderMapper.PROCEDURE, request.getProcedure());
  httpHeaders.set(HeaderMapper.CALLER, request.getCaller());
  httpHeaders.set(HeaderMapper.ENCODING, request.getEncoding());
  // Required headers for HTTP
  httpHeaders.set(HttpHeaderNames.HOST, hostString);
  if (request.getDeadline() != null) {
    Instant now = Instant.now();
    long timeRemaining = ChronoUnit.MILLIS.between(now, request.getDeadline());
    if (timeRemaining <= 0) {
      throw new DeadlineExceededException();
    }
    httpHeaders.set(HeaderMapper.TIMEOUT, String.valueOf(timeRemaining));
  }
  if (request.getShardKey() != null) {
    httpHeaders.set(HeaderMapper.SHARD_KEY, request.getShardKey());
  }
  if (request.getRoutingKey() != null) {
    httpHeaders.set(HeaderMapper.ROUTING_KEY, request.getRoutingKey());
  }
  if (request.getRoutingDelegate() != null) {
    httpHeaders.set(HeaderMapper.ROUTING_DELEGATE, request.getRoutingDelegate());
  }
  if (request.getSpan() != null) {
    channelAttrs.attr(ChannelAttributes.SPAN).set(request.getSpan());
    tracer.inject(
        request.getSpan().context(), Builtin.HTTP_HEADERS, new TextMapInjectAdapter(httpHeaders));
  }
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
public TransportRequest build(
    HttpRequest request, TransportBody body, AttributeMap channelAttrs) {
  HttpHeaders httpHeaders = request.headers();

  // Keep track of request information
  String service = httpHeaders.get(HeaderMapper.SERVICE);
  channelAttrs.attr(ChannelAttributes.SERVICE).set(service);
  String procedure = httpHeaders.get(HeaderMapper.PROCEDURE);
  channelAttrs.attr(ChannelAttributes.PROCEDURE).set(procedure);
  String caller = httpHeaders.get(HeaderMapper.CALLER);
  channelAttrs.attr(ChannelAttributes.CALLER).set(caller);

  // Keep track of when the request started
  Instant start = Instant.now();
  channelAttrs.attr(ChannelAttributes.REQUEST_START).set(start);

  Instant deadline =
      createDeadline(start, service, procedure, httpHeaders.get(HeaderMapper.TIMEOUT));
  Span span = createSpan(start, service, procedure, caller, httpHeaders);
  channelAttrs.attr(ChannelAttributes.SPAN).set(span);

  return DefaultTransportRequest.builder()
      .service(service)
      .procedure(procedure)
      .deadline(deadline)
      .caller(caller)
      .encoding(httpHeaders.get(HeaderMapper.ENCODING))
      .shardKey(httpHeaders.get(HeaderMapper.SHARD_KEY))
      .routingKey(httpHeaders.get(HeaderMapper.ROUTING_KEY))
      .routingDelegate(httpHeaders.get(HeaderMapper.ROUTING_DELEGATE))
      .headers(HeaderMapper.fromHttpHeaders(httpHeaders))
      .span(span)
      .body(body)
      .build();
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Test(expected = DeadlineExceededException.class)
public void testEncodeRequestWithPastDeadline() throws Exception {
  TransportRequest request =
      DefaultTransportRequest.builder()
          .caller("caller")
          .service("service")
          .procedure("procedure")
          .encoding("http")
          .body(TransportBody.fromByteArray(new byte[0]))
          .deadline(Instant.now().minusSeconds(60))
          .build();
  encoderConfig.buildFullMessage(request, new byte[0], mock(AttributeMap.class));
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
public void onChannelActive(AttributeMap attMap) {
  C newConnection = safeRequestProcessor.openConnection();
  Connection oldConnection = attMap.attr(connection).setIfAbsent(
      newConnection
  );
  if (oldConnection != null) {
    throw new IllegalArgumentException("A connection with id "
        + oldConnection.getConnectionId() + " was stored before "
        + "channel became active!");
  }
}
 

import io.netty.util.AttributeMap; //导入依赖的package包/类
@Override
public void onChannelInactive(AttributeMap attMap) {
  C connection = attMap.attr(this.connection).getAndRemove();
  if (connection != null) {
    connection.close();
  }
}