@Test public void testUnsubscribeWithBadFormattedTopic() { MqttUnsubscribeMessage msg = MqttMessageBuilders.unsubscribe().addTopicFilter(BAD_FORMATTED_TOPIC).messageId(1) .build(); m_processor.processUnsubscribe(m_channel, msg); assertFalse("If client unsubscribe with bad topic than channel must be closed", m_channel.isOpen()); }
public void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg) { List<String> topics = msg.payload().topics(); String clientID = NettyUtils.clientID(channel); LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}", clientID, topics); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); for (String t : topics) { Topic topic = new Topic(t); boolean validTopic = topic.isValid(); if (!validTopic) { channel.close(); LOG.error("Topic filter is not valid. CId={}, topics={}, badTopicFilter={}", clientID, topics, topic); return; } if(LOG.isDebugEnabled()){ LOG.debug("Removing subscription. CId={}, topic={}", clientID, topic); } subscriptions.removeSubscription(topic, clientID); clientSession.unsubscribeFrom(topic); String username = NettyUtils.userName(channel); m_interceptor.notifyTopicUnsubscribed(topic.toString(), clientID, username); } int messageID = msg.variableHeader().messageId(); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_LEAST_ONCE, false, 0); MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID)); LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}", clientID, topics, messageID); channel.writeAndFlush(ackMessage); }
ProtocolProcessor { public void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg) { List<String> topics = msg.payload().topics(); String clientID = NettyUtils.clientID(channel); LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}", clientID, topics); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); for (String t : topics) { Topic topic = new Topic(t); boolean validTopic = topic.isValid(); if (!validTopic) { channel.close(); LOG.error("Topic filter is not valid. CId={}, topics={}, badTopicFilter={}", clientID, topics, topic); return; } if(LOG.isDebugEnabled()){ LOG.debug("Removing subscription. CId={}, topic={}", clientID, topic); } subscriptions.removeSubscription(topic, clientID); clientSession.unsubscribeFrom(topic); String username = NettyUtils.userName(channel); m_interceptor.notifyTopicUnsubscribed(topic.toString(), clientID, username); } int messageID = msg.variableHeader().messageId(); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_LEAST_ONCE, false, 0); MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID)); LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}", clientID, topics, messageID); channel.writeAndFlush(ackMessage); } }
ProtocolProcessor { public void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg) { List<String> topics = msg.payload().topics(); String clientID = NettyUtils.clientID(channel); LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}", clientID, topics); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); for (String t : topics) { Topic topic = new Topic(t); boolean validTopic = topic.isValid(); if (!validTopic) { channel.close(); LOG.error("Topic filter is not valid. CId={}, topics={}, badTopicFilter={}", clientID, topics, topic); return; } if(LOG.isDebugEnabled()){ LOG.debug("Removing subscription. CId={}, topic={}", clientID, topic); } subscriptions.removeSubscription(topic, clientID); clientSession.unsubscribeFrom(topic); String username = NettyUtils.userName(channel); m_interceptor.notifyTopicUnsubscribed(topic.toString(), clientID, username); } int messageID = msg.variableHeader().messageId(); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_LEAST_ONCE, false, 0); MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID)); LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}", clientID, topics, messageID); channel.writeAndFlush(ackMessage); } ProtocolProcessor(); }
ProtocolProcessor { public void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg) { List<String> topics = msg.payload().topics(); String clientID = NettyUtils.clientID(channel); LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}", clientID, topics); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); for (String t : topics) { Topic topic = new Topic(t); boolean validTopic = topic.isValid(); if (!validTopic) { channel.close(); LOG.error("Topic filter is not valid. CId={}, topics={}, badTopicFilter={}", clientID, topics, topic); return; } if(LOG.isDebugEnabled()){ LOG.debug("Removing subscription. CId={}, topic={}", clientID, topic); } subscriptions.removeSubscription(topic, clientID); clientSession.unsubscribeFrom(topic); String username = NettyUtils.userName(channel); m_interceptor.notifyTopicUnsubscribed(topic.toString(), clientID, username); } int messageID = msg.variableHeader().messageId(); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_LEAST_ONCE, false, 0); MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID)); LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}", clientID, topics, messageID); channel.writeAndFlush(ackMessage); } ProtocolProcessor(); void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                     IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator,
                     BrokerInterceptor interceptor); void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                     IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
                     IAuthorizator authorizator, BrokerInterceptor interceptor); void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                     IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
                     IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort); void processConnect(Channel channel, MqttConnectMessage msg); void processPubAck(Channel channel, MqttPubAckMessage msg); static IMessagesStore.StoredMessage asStoredMessage(MqttPublishMessage msg); void processPublish(Channel channel, MqttPublishMessage msg); void internalPublish(MqttPublishMessage msg, final String clientId); void processPubRel(Channel channel, MqttMessage msg); void processPubRec(Channel channel, MqttMessage msg); void processPubComp(Channel channel, MqttMessage msg); void processDisconnect(Channel channel); void processConnectionLost(String clientID, Channel channel); void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg); void processSubscribe(Channel channel, MqttSubscribeMessage msg); void notifyChannelWritable(Channel channel); void addInterceptHandler(InterceptHandler interceptHandler); void removeInterceptHandler(InterceptHandler interceptHandler); IMessagesStore getMessagesStore(); ISessionsStore getSessionsStore(); }
ProtocolProcessor { public void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg) { List<String> topics = msg.payload().topics(); String clientID = NettyUtils.clientID(channel); LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}", clientID, topics); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); for (String t : topics) { Topic topic = new Topic(t); boolean validTopic = topic.isValid(); if (!validTopic) { channel.close(); LOG.error("Topic filter is not valid. CId={}, topics={}, badTopicFilter={}", clientID, topics, topic); return; } if(LOG.isDebugEnabled()){ LOG.debug("Removing subscription. CId={}, topic={}", clientID, topic); } subscriptions.removeSubscription(topic, clientID); clientSession.unsubscribeFrom(topic); String username = NettyUtils.userName(channel); m_interceptor.notifyTopicUnsubscribed(topic.toString(), clientID, username); } int messageID = msg.variableHeader().messageId(); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_LEAST_ONCE, false, 0); MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID)); LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}", clientID, topics, messageID); channel.writeAndFlush(ackMessage); } ProtocolProcessor(); void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                     IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator,
                     BrokerInterceptor interceptor); void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                     IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
                     IAuthorizator authorizator, BrokerInterceptor interceptor); void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                     IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
                     IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort); void processConnect(Channel channel, MqttConnectMessage msg); void processPubAck(Channel channel, MqttPubAckMessage msg); static IMessagesStore.StoredMessage asStoredMessage(MqttPublishMessage msg); void processPublish(Channel channel, MqttPublishMessage msg); void internalPublish(MqttPublishMessage msg, final String clientId); void processPubRel(Channel channel, MqttMessage msg); void processPubRec(Channel channel, MqttMessage msg); void processPubComp(Channel channel, MqttMessage msg); void processDisconnect(Channel channel); void processConnectionLost(String clientID, Channel channel); void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg); void processSubscribe(Channel channel, MqttSubscribeMessage msg); void notifyChannelWritable(Channel channel); void addInterceptHandler(InterceptHandler interceptHandler); void removeInterceptHandler(InterceptHandler interceptHandler); IMessagesStore getMessagesStore(); ISessionsStore getSessionsStore();  }