MqttEncoder Class — netty Architecture

Source Code

codec-mqtt/src/main/java/io/netty/handler/codec/mqtt/MqttEncoder.java lines 69–765

@ChannelHandler.Sharable
public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {

    public static final MqttEncoder INSTANCE = new MqttEncoder();

    private MqttEncoder() {
        super(MqttMessage.class);
    }

    @Override
    protected void encode(ChannelHandlerContext ctx, MqttMessage msg, List<Object> out) throws Exception {
        out.add(doEncode(ctx, msg));
    }

    /**
     * This is the main encoding method.
     * It's only visible for testing.
     *
     * @param message MQTT message to encode
     * @return ByteBuf with encoded bytes
     */
    static ByteBuf doEncode(ChannelHandlerContext ctx,
                     MqttMessage message) {

        switch (message.fixedHeader().messageType()) {
            case CONNECT:
                return encodeConnectMessage(ctx, (MqttConnectMessage) message);

            case CONNACK:
                return encodeConnAckMessage(ctx, (MqttConnAckMessage) message);

            case PUBLISH:
                return encodePublishMessage(ctx, (MqttPublishMessage) message);

            case SUBSCRIBE:
                return encodeSubscribeMessage(ctx, (MqttSubscribeMessage) message);

            case UNSUBSCRIBE:
                return encodeUnsubscribeMessage(ctx,  (MqttUnsubscribeMessage) message);

            case SUBACK:
                return encodeSubAckMessage(ctx, (MqttSubAckMessage) message);

            case UNSUBACK:
                if (message instanceof MqttUnsubAckMessage) {
                    return encodeUnsubAckMessage(ctx, (MqttUnsubAckMessage) message);
                }
                return encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(ctx.alloc(), message);

            case PUBACK:
            case PUBREC:
            case PUBREL:
            case PUBCOMP:
                return encodePubReplyMessage(ctx, message);

            case DISCONNECT:
            case AUTH:
                return encodeReasonCodePlusPropertiesMessage(ctx, message);

            case PINGREQ:
            case PINGRESP:
                return encodeMessageWithOnlySingleByteFixedHeader(ctx.alloc(), message);

            default:
                throw new IllegalArgumentException(
                        "Unknown message type: " + message.fixedHeader().messageType().value());
        }
    }

    private static ByteBuf encodeConnectMessage(
            ChannelHandlerContext ctx,
            MqttConnectMessage message) {
        int payloadBufferSize = 0;

        MqttFixedHeader mqttFixedHeader = message.fixedHeader();
        MqttConnectVariableHeader variableHeader = message.variableHeader();
        MqttConnectPayload payload = message.payload();
        MqttVersion mqttVersion = MqttVersion.fromProtocolNameAndLevel(variableHeader.name(),
                (byte) variableHeader.version());
        setMqttVersion(ctx, mqttVersion);