void processRegularSessionMsg(ChannelHandlerContext ctx, MqttMessage msg) {
        switch (msg.fixedHeader().messageType()) {
            case PUBLISH:
                processPublish(ctx, (MqttPublishMessage) msg);
                break;
            case SUBSCRIBE:
                processSubscribe(ctx, (MqttSubscribeMessage) msg);
                break;
            case UNSUBSCRIBE:
                processUnsubscribe(ctx, (MqttUnsubscribeMessage) msg);
                break;
            case PINGREQ:
                if (checkConnected(ctx, msg)) {
                    ctx.writeAndFlush(new MqttMessage(new MqttFixedHeader(PINGRESP, false, AT_MOST_ONCE, false, 0)));
                    transportService.recordActivity(deviceSessionCtx.getSessionInfo());
                    if (gatewaySessionHandler != null) {
                        gatewaySessionHandler.onGatewayPing();
                    }
                }
                break;
            case DISCONNECT:
                closeCtx(ctx, MqttReasonCodes.Disconnect.NORMAL_DISCONNECT);
                break;
            case PUBACK:
                int msgId = ((MqttPubAckMessage) msg).variableHeader().messageId();
                TransportProtos.ToDeviceRpcRequestMsg rpcRequest = rpcAwaitingAck.remove(msgId);
                if (rpcRequest != null) {
                    transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, true, TransportServiceCallback.EMPTY);
                }
                break;
            default:
                break;
        }
    }
@Test
    public void givenMqttMessage_whenDeviceProfileMqttTransport_thenTopicAddedToMetadata() {
        MqttPublishMessage message = getDeviceMqttPublishMessage();
        when(context.getJsonMqttAdaptor()).thenReturn(new JsonMqttAdaptor());
        handler.deviceSessionCtx.setConnected(true);
        DeviceProfile deviceProfile = new DeviceProfile();
        DeviceProfileData deviceProfileData = new DeviceProfileData();
        MqttDeviceProfileTransportConfiguration mqttDeviceProfileTransportConfiguration = new MqttDeviceProfileTransportConfiguration();
        mqttDeviceProfileTransportConfiguration.setTransportPayloadTypeConfiguration(new JsonTransportPayloadConfiguration());
        deviceProfileData.setTransportConfiguration(mqttDeviceProfileTransportConfiguration);
        deviceProfile.setProfileData(deviceProfileData);
        deviceProfile.setTransportType(DeviceTransportType.MQTT);
        handler.deviceSessionCtx.setDeviceProfile(deviceProfile);

        handler.processRegularSessionMsg(ctx, message);

        TbMsgMetaData expectedMd = new TbMsgMetaData();
        expectedMd.putValue(DataConstants.MQTT_TOPIC, message.variableHeader().topicName());

        verify(transportService, times(1)).process(any(), (TransportProtos.PostTelemetryMsg) any(), eq(expectedMd), any());
    }