import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
private void processAuthTokenConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) {
  String userName = msg.payload().userName();
  String clientIdentifier = msg.payload().clientIdentifier();
  if (StringUtils.isEmpty(userName)) {
    // ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD));
    // ctx.close();
    ctx.writeAndFlush(createMqttConnAckMsg(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD));
    connected = false;
  } else {
    boolean login = deviceSessionCtx.login(new DeviceTokenCredentials(userName));
    if (!login) {
      ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_REFUSED_NOT_AUTHORIZED));
      connected = false;
    } else {
      MemoryMetaPool.registerClienId(clientIdentifier, ctx.channel());

      ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_ACCEPTED));
      connected = true;
      checkGatewaySession();
    }
    // }
  }

}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
public static MqttConnectMessage connect(ConnectOptions options) {
	MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false,
			10);
	MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader(options.version().protocolName(),
			options.version().protocolLevel(), options.userName() != null, options.password() != null,
			options.will() == null ? false : options.will().isRetain(),
			options.will() == null ? 0 : options.will().qos().value(), options.will() != null,
			options.cleanSession(), options.keepAliveTimeSeconds());

	MqttConnectPayload payload = new MqttConnectPayload(Strings.nullToEmpty(options.clientId()),
			options.will() == null ? "" : options.will().topicName(),
			options.will() == null ? "" : new String(options.will().message(), CharsetUtil.UTF_8),
			Strings.nullToEmpty(options.userName()), Strings.nullToEmpty(options.password()));

	return new MqttConnectMessage(fixedHeader, variableHeader, payload);
}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
private MqttConnAckMessage executeNormalChannelRead0(String clientId, boolean cleanSession, ChannelId channelId)
		throws Exception {
	MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false,
			10);
	MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader("MQTT", 4, true, true, true, 0, true,
			cleanSession, 60);
	MqttConnectPayload payload = new MqttConnectPayload(clientId, "willtopic", "willmessage", "username",
			"password");

	MqttConnectMessage msg = new MqttConnectMessage(fixedHeader, variableHeader, payload);

	ChannelId cid = channelId == null ? TestUtil.newChannelId(clientId, false) : channelId;

	EmbeddedChannel channel = new EmbeddedChannel(cid, new ConnectReceiver());

	channel.writeInbound(msg);

	return channel.readOutbound();
}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
private void processMqttMsg(ChannelHandlerContext ctx, MqttMessage msg) {
  // deviceSessionCtx.setChannel(ctx);
  // assetSessionCtx.setChannel(ctx);

  switch (msg.fixedHeader().messageType()) {
  case CONNECT:
    processConnect(ctx, (MqttConnectMessage) msg);
    break;
  case PUBLISH:
    processPublish(ctx, (MqttPublishMessage) msg);
    // System.out.println("write...");
    // ctx.write("just for test");
    break;
  case SUBSCRIBE:
    processSubscribe(ctx, (MqttSubscribeMessage) msg);
    break;
  case UNSUBSCRIBE:
    processUnsubscribe(ctx, (MqttUnsubscribeMessage) msg);
    break;
  case PINGREQ:
    if (checkConnected(ctx)) {
      ctx.writeAndFlush(new MqttMessage(new MqttFixedHeader(PINGRESP, false, AT_MOST_ONCE, false, 0)));
    }
    break;
  case DISCONNECT:
    if (checkConnected(ctx)) {
      processDisconnect(ctx);
    }
    break;
  }
}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
private Message newWill(String clientId, MqttConnectMessage conn) {
	if (!conn.variableHeader().isWillFlag()) { return null; } // [MQTT-3.1.2-12]

	return new Message(-1, conn.payload().willTopic(), clientId,
			conn.payload().willMessage().getBytes(CharsetUtil.UTF_8),
			MqttQoS.valueOf(conn.variableHeader().willQos()), conn.variableHeader().isWillRetain());
}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
private boolean filterPlugins(ChannelHandlerContext ctx, MqttConnectMessage msg) {
	String clientId = msg.payload().clientIdentifier();
	String userName = msg.variableHeader().hasUserName() ? msg.payload().userName() : null;
	String password = msg.variableHeader().hasPassword() ? msg.payload().password() : null;

	if (!Plugins.INSTANCE.get(ServiceChecker.class).isServiceAvailable()) {
		sendNoneAcceptMessage(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
		return false;
	}

	if (!Plugins.INSTANCE.get(Authenticator.class).isValid(clientId)) {
		sendNoneAcceptMessage(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED); // [MQTT-3.1.3-9]
		return false;
	}

	if (!Plugins.INSTANCE.get(Authenticator.class).isValid(clientId, userName, password)) {
		sendNoneAcceptMessage(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
		return false;
	}

	if (!Plugins.INSTANCE.get(Authorizer.class).isAuthorized(clientId, userName)) {
		sendNoneAcceptMessage(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED);
		return false;
	}

	return true;
}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
@Override
public boolean intercept(final MqttMessage mqttMessage, RemotingConnection connection) {
   System.out.println("MQTT control packet was intercepted " + mqttMessage.fixedHeader().messageType());

   // If you need to handle an specific packet type:
   if (mqttMessage instanceof MqttPublishMessage) {
      MqttPublishMessage message = (MqttPublishMessage) mqttMessage;


      String originalMessage = message.payload().toString(Charset.forName("UTF-8"));
      System.out.println("Original message: " + originalMessage);

      // The new message content must not be bigger that the original content.
      String modifiedMessage = "Modified message ";

      message.payload().setBytes(0, modifiedMessage.getBytes());
   } else {
      if (mqttMessage instanceof MqttConnectMessage) {
         MqttConnectMessage connectMessage = (MqttConnectMessage) mqttMessage;
         System.out.println("MQTT CONNECT control packet was intercepted " + connectMessage);
      }
   }


   // We return true which means "call next interceptor" (if there is one) or target.
   // If we returned false, it means "abort call" - no more interceptors would be called and neither would
   // the target
   return true;
}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
/**
 * Called during connection.
 *
 * @param connect
 */
void handleConnect(MqttConnectMessage connect, ChannelHandlerContext ctx) throws Exception {
   this.ctx = ctx;
   connectionEntry.ttl = connect.variableHeader().keepAliveTimeSeconds() * 1500L;

   String clientId = connect.payload().clientIdentifier();
   session.getConnectionManager().connect(clientId, connect.payload().userName(), connect.payload().passwordInBytes(), connect.variableHeader().isWillFlag(), connect.payload().willMessageInBytes(), connect.payload().willTopic(), connect.variableHeader().isWillRetain(), connect.variableHeader().willQos(), connect.variableHeader().isCleanSession());
}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
@Override
public void notifyClientConnected(final MqttConnectMessage msg) {
    for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) {
        LOG.debug("Sending MQTT CONNECT message to interceptor. CId={}, interceptorId={}",
                msg.payload().clientIdentifier(), handler.getID());
        executor.execute(() -> handler.onConnect(new InterceptConnectMessage(msg)));
    }
}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
private Session newSession(MqttConnectMessage msg, boolean cleanSession, String clientId, String clientIp,
		int clientPort) {
	return new Session(clientId, clientIp, clientPort, msg.variableHeader().keepAliveTimeSeconds(), cleanSession,
			newWill(clientId, msg));
}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
   try {
      if (stopped) {
         disconnect(true);
         return;
      }

      MqttMessage message = (MqttMessage) msg;

      // Disconnect if Netty codec failed to decode the stream.
      if (message.decoderResult().isFailure()) {
         log.debug("Bad Message Disconnecting Client.");
         disconnect(true);
         return;
      }

      connection.dataReceived();

      MQTTUtil.logMessage(session.getState(), message, true);

      this.protocolManager.invokeIncoming(message, this.connection);

      switch (message.fixedHeader().messageType()) {
         case CONNECT:
            handleConnect((MqttConnectMessage) message, ctx);
            break;
         case PUBLISH:
            handlePublish((MqttPublishMessage) message);
            break;
         case PUBACK:
            handlePuback((MqttPubAckMessage) message);
            break;
         case PUBREC:
            handlePubrec(message);
            break;
         case PUBREL:
            handlePubrel(message);
            break;
         case PUBCOMP:
            handlePubcomp(message);
            break;
         case SUBSCRIBE:
            handleSubscribe((MqttSubscribeMessage) message);
            break;
         case UNSUBSCRIBE:
            handleUnsubscribe((MqttUnsubscribeMessage) message);
            break;
         case PINGREQ:
            handlePingreq();
            break;
         case DISCONNECT:
            disconnect(false);
            break;
         case UNSUBACK:
         case SUBACK:
         case PINGRESP:
         case CONNACK: // The server does not instantiate connections therefore any CONNACK received over a connection is an invalid control message.
         default:
            disconnect(true);
      }
   } catch (Exception e) {
      log.debug("Error processing Control Packet, Disconnecting Client", e);
      disconnect(true);
   } finally {
      ReferenceCountUtil.release(msg);
   }
}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
public InterceptConnectMessage(MqttConnectMessage msg) {
    super(msg);
    this.msg = msg;
}
 

import io.netty.handler.codec.mqtt.MqttConnectMessage; //导入依赖的package包/类
void notifyClientConnected(MqttConnectMessage msg);