@Test public void testSubscribeWithBadFormattedTopic() { MqttSubscribeMessage msg = MqttMessageBuilders.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, BAD_FORMATTED_TOPIC) .messageId(10).build(); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); m_processor.processSubscribe(m_channel, msg); assertEquals(0, subscriptions.size()); Object recvSubAckMessage = m_channel.readOutbound(); assertTrue(recvSubAckMessage instanceof MqttSubAckMessage); verifyFailureQos((MqttSubAckMessage) recvSubAckMessage); }
public void processSubscribe(Channel channel, MqttSubscribeMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); LOG.info("Processing SUBSCRIBE message. CId={}, messageId={}", clientID, messageID); RunningSubscription executionKey = new RunningSubscription(clientID, messageID); SubscriptionState currentStatus = subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED); if (currentStatus != null) { LOG.warn("Client sent another SUBSCRIBE message while this one was being processed CId={}, messageId={}", clientID, messageID); return; } String username = NettyUtils.userName(channel); List<MqttTopicSubscription> ackTopics = doVerify(clientID, username, msg); MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID); if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) { LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified CId={}, " + "messageId={}", clientID, messageID); return; } LOG.info("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics); List<Subscription> newSubscriptions = doStoreSubscription(ackTopics, clientID); for (Subscription subscription : newSubscriptions) { subscriptions.add(subscription.asClientTopicCouple()); } LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID); channel.writeAndFlush(ackMessage); for (Subscription subscription : newSubscriptions) { publishRetainedMessagesInSession(subscription, username); } boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED); if (!success) { LOG.warn("Unable to perform the final subscription state update CId={}, messageId={}", clientID, messageID); } }
ProtocolProcessor { public void processSubscribe(Channel channel, MqttSubscribeMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); LOG.info("Processing SUBSCRIBE message. CId={}, messageId={}", clientID, messageID); RunningSubscription executionKey = new RunningSubscription(clientID, messageID); SubscriptionState currentStatus = subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED); if (currentStatus != null) { LOG.warn("Client sent another SUBSCRIBE message while this one was being processed CId={}, messageId={}", clientID, messageID); return; } String username = NettyUtils.userName(channel); List<MqttTopicSubscription> ackTopics = doVerify(clientID, username, msg); MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID); if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) { LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified CId={}, " + "messageId={}", clientID, messageID); return; } LOG.info("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics); List<Subscription> newSubscriptions = doStoreSubscription(ackTopics, clientID); for (Subscription subscription : newSubscriptions) { subscriptions.add(subscription.asClientTopicCouple()); } LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID); channel.writeAndFlush(ackMessage); for (Subscription subscription : newSubscriptions) { publishRetainedMessagesInSession(subscription, username); } boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED); if (!success) { LOG.warn("Unable to perform the final subscription state update CId={}, messageId={}", clientID, messageID); } } }
ProtocolProcessor { public void processSubscribe(Channel channel, MqttSubscribeMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); LOG.info("Processing SUBSCRIBE message. CId={}, messageId={}", clientID, messageID); RunningSubscription executionKey = new RunningSubscription(clientID, messageID); SubscriptionState currentStatus = subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED); if (currentStatus != null) { LOG.warn("Client sent another SUBSCRIBE message while this one was being processed CId={}, messageId={}", clientID, messageID); return; } String username = NettyUtils.userName(channel); List<MqttTopicSubscription> ackTopics = doVerify(clientID, username, msg); MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID); if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) { LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified CId={}, " + "messageId={}", clientID, messageID); return; } LOG.info("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics); List<Subscription> newSubscriptions = doStoreSubscription(ackTopics, clientID); for (Subscription subscription : newSubscriptions) { subscriptions.add(subscription.asClientTopicCouple()); } LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID); channel.writeAndFlush(ackMessage); for (Subscription subscription : newSubscriptions) { publishRetainedMessagesInSession(subscription, username); } boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED); if (!success) { LOG.warn("Unable to perform the final subscription state update CId={}, messageId={}", clientID, messageID); } } ProtocolProcessor(); }
ProtocolProcessor { public void processSubscribe(Channel channel, MqttSubscribeMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); LOG.info("Processing SUBSCRIBE message. CId={}, messageId={}", clientID, messageID); RunningSubscription executionKey = new RunningSubscription(clientID, messageID); SubscriptionState currentStatus = subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED); if (currentStatus != null) { LOG.warn("Client sent another SUBSCRIBE message while this one was being processed CId={}, messageId={}", clientID, messageID); return; } String username = NettyUtils.userName(channel); List<MqttTopicSubscription> ackTopics = doVerify(clientID, username, msg); MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID); if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) { LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified CId={}, " + "messageId={}", clientID, messageID); return; } LOG.info("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics); List<Subscription> newSubscriptions = doStoreSubscription(ackTopics, clientID); for (Subscription subscription : newSubscriptions) { subscriptions.add(subscription.asClientTopicCouple()); } LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID); channel.writeAndFlush(ackMessage); for (Subscription subscription : newSubscriptions) { publishRetainedMessagesInSession(subscription, username); } boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED); if (!success) { LOG.warn("Unable to perform the final subscription state update CId={}, messageId={}", clientID, messageID); } } 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 processSubscribe(Channel channel, MqttSubscribeMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); LOG.info("Processing SUBSCRIBE message. CId={}, messageId={}", clientID, messageID); RunningSubscription executionKey = new RunningSubscription(clientID, messageID); SubscriptionState currentStatus = subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED); if (currentStatus != null) { LOG.warn("Client sent another SUBSCRIBE message while this one was being processed CId={}, messageId={}", clientID, messageID); return; } String username = NettyUtils.userName(channel); List<MqttTopicSubscription> ackTopics = doVerify(clientID, username, msg); MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID); if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) { LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified CId={}, " + "messageId={}", clientID, messageID); return; } LOG.info("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics); List<Subscription> newSubscriptions = doStoreSubscription(ackTopics, clientID); for (Subscription subscription : newSubscriptions) { subscriptions.add(subscription.asClientTopicCouple()); } LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID); channel.writeAndFlush(ackMessage); for (Subscription subscription : newSubscriptions) { publishRetainedMessagesInSession(subscription, username); } boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED); if (!success) { LOG.warn("Unable to perform the final subscription state update CId={}, messageId={}", clientID, messageID); } } 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();  }