public MqttMessage processConnect(MqttConnectMessage msg, SessionContext sessionData) {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_LEAST_ONCE, false, 0);

    String clientId = sessionData.getClientId();
    if (StringUtils.isBlank(clientId)) {
        log.error("clientId is empty, reject");
        return MqttMessageFactory.newMessage(fixedHeader,
                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false), null);
    }

    // verify userName and password
    String username = msg.payload().userName();
    String password = msg.payload().passwordInBytes() == null ? null : new String(msg.payload().passwordInBytes(), StandardCharsets.UTF_8);
    if (!this.authService.verifyUserName(username, password)) {
        log.error("verify account failed, reject");
        return MqttMessageFactory.newMessage(fixedHeader,
                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false), null);
    }

    if (this.sessionStore.existSession(clientId)) {
        log.info("exist client id, force to delete the older");
        this.sessionStore.removeSession(clientId);
    }

    // store new session
    this.sessionStore.addSession(clientId, sessionData);

    log.info("MQTT connected, clientId: {}", clientId);
    return MqttMessageFactory.newMessage(fixedHeader,
            new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, false), null);
}
 

/**
 * B - S, B - P
 * @param channel
 * @param msg
 */
public void processConnectBack(Channel channel, MqttConnAckMessage msg) {
	MqttConnAckVariableHeader mqttConnAckVariableHeader = msg.variableHeader();
	String sErrorMsg = "";
	switch (mqttConnAckVariableHeader.connectReturnCode()) {
	case CONNECTION_ACCEPTED:
		clientProcess.loginFinish(true, null);
		return;
	case CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD:
		sErrorMsg = "用户名密码错误";
		break;
	case CONNECTION_REFUSED_IDENTIFIER_REJECTED:
		sErrorMsg = "clientId不允许链接";
		break;
	case CONNECTION_REFUSED_SERVER_UNAVAILABLE:
		sErrorMsg = "服务不可用";
		break;
	case CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION:
		sErrorMsg = "mqtt 版本不可用";
		break;
	case CONNECTION_REFUSED_NOT_AUTHORIZED:
		sErrorMsg = "未授权登录";
		break;
	default:
		break;
	}

	clientProcess.loginFinish(false, new LoginException(sErrorMsg));
	channel.close();
}
 

private MqttConnAckMessage sendAckToClient(Channel client, MqttConnectMessage connectMessage, MqttConnectReturnCode ackCode, boolean sessionPresent) {

        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, connectMessage.fixedHeader().qosLevel(), false, 0);
        MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                mqttFixedHeader,
                new MqttConnAckVariableHeader(ackCode, sessionPresent),
                null);
        client.writeAndFlush(connAckMessage);
        return connAckMessage;
    }
 

public static MqttConnAckMessage connack(MqttConnectReturnCode returnCode, boolean sessionPresent) {
	MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false,
			2);
	MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent);

	return new MqttConnAckMessage(fixedHeader, variableHeader);
}
 

public static MqttConnAckMessage connAckMessage(MqttConnectReturnCode code, boolean sessionPresent) {
	return (MqttConnAckMessage) MqttMessageFactory.newMessage(
			new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
			new MqttConnAckVariableHeader(code, sessionPresent), null);
}
 

public static MqttConnAckMessage getConnectAckMessage(MqttConnectReturnCode returnCode,boolean sessionPresent){
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
    MqttConnAckVariableHeader variableHeade = new MqttConnAckVariableHeader(returnCode,sessionPresent);
    return new MqttConnAckMessage(fixedHeader,variableHeade);
}
 

private MqttConnAckMessage createMqttConnAckMsg(MqttConnectReturnCode returnCode) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(CONNACK, false, AT_MOST_ONCE, false, 0);
  MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, true);
  return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}
 

private static MqttConnAckMessage createConnAckMessage(MqttConnectReturnCode code) {
    MqttFixedHeader mqttFixedHeader =
            new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 2);
    MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(code, true);
    return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}
 

void sendConnack(MqttConnectReturnCode returnCode) {
   MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
   MqttConnAckVariableHeader varHeader = new MqttConnAckVariableHeader(returnCode, true);
   MqttConnAckMessage message = new MqttConnAckMessage(fixedHeader, varHeader);
   sendToClient(message);
}