ProtocolProcessor { public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository) { init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false, authorizator, interceptor, sessionsRepository); } ProtocolProcessor(); void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                     IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator,
                     BrokerInterceptor interceptor, SessionsRepository sessionsRepository); void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                     IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
                     IAuthorizator authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository); 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(); void shutdown();  }
@Test public void publishNoPublishToInactiveSession() { ISubscriptionsDirectory mockedSubscriptions = mock(ISubscriptionsDirectory.class); Subscription inactiveSub = new Subscription("Subscriber", new Topic("/topic"), MqttQoS.AT_LEAST_ONCE); List<Subscription> inactiveSubscriptions = Collections.singletonList(inactiveSub); when(mockedSubscriptions.matches(eq(new Topic("/topic")))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); m_processor.init(mockedSubscriptions, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); m_processor.sessionsRepository.createNewSession("Subscriber", false); connectAsClient("Publisher"); publishToAs("Publisher", "/topic", AT_MOST_ONCE, true); verifyNoPublishIsReceived(); }
@Test public void testPublishToItself() throws InterruptedException { final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE); ISubscriptionsDirectory subs = new CTrieSubscriptionDirectory() { @Override public List<Subscription> matches(Topic topic) { if (topic.toString().equals(FAKE_TOPIC)) { return Collections.singletonList(subscription); } else { throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic); } } }; MemoryStorageService storageService = new MemoryStorageService(null, null); SessionsRepository sessionsRepository = new SessionsRepository(storageService.sessionsStore(), null); subs.init(sessionsRepository); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); connect_v3_1(); publishToAs("FakeCLI", FAKE_TOPIC, AT_MOST_ONCE, false); verifyPublishIsReceived(); }
@Test public void testSubscribedToNotAuthorizedTopic() { final String fakeUserName = "UnAuthUser"; NettyUtils.userName(m_channel, fakeUserName); IAuthorizator mockAuthorizator = mock(IAuthorizator.class); when(mockAuthorizator.canRead(eq(new Topic(FAKE_TOPIC)), eq(fakeUserName), eq(FAKE_CLIENT_ID))) .thenReturn(false); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, mockAuthorizator, NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); connect(); MqttSubAckMessage subAckMsg = subscribeWithoutVerify(FAKE_TOPIC, AT_MOST_ONCE); verifyFailureQos(subAckMsg); }
@Test public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception { final Subscription subscription = new Subscription(FAKE_PUBLISHER_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE); ISubscriptionsDirectory subs = new CTrieSubscriptionDirectory() { @Override public List<Subscription> matches(Topic topic) { if (topic.toString().equals(FAKE_TOPIC)) { return Collections.singletonList(subscription); } else { throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic); } } }; MemoryStorageService storageService = new MemoryStorageService(null, null); SessionsRepository sessionsRepository = new SessionsRepository(storageService.sessionsStore(), null); subs.init(sessionsRepository); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); connect_v3_1_asClient(FAKE_PUBLISHER_ID); publishToAs(FAKE_PUBLISHER_ID, FAKE_TOPIC, AT_MOST_ONCE, true); NettyUtils.cleanSession(m_channel, false); subscribeAndNotReadResponse("#", AT_MOST_ONCE); verifyPublishIsReceived(); }
@Test public void testRepublishAndConsumePersistedMessages_onReconnect() { ISubscriptionsDirectory subs = mock(ISubscriptionsDirectory.class); List<Subscription> emptySubs = Collections.emptyList(); when(subs.matches(any(Topic.class))).thenReturn(emptySubs); StoredMessage retainedMessage = new StoredMessage("Hello".getBytes(), MqttQoS.EXACTLY_ONCE, "/topic"); retainedMessage.setRetained(true); retainedMessage.setClientID(FAKE_PUBLISHER_ID); m_messagesStore.storeRetained(new Topic("/topic"), retainedMessage); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); connect_v3_1_asClient(FAKE_PUBLISHER_ID); assertTrue(this.sessionsRepository.sessionForClient(FAKE_PUBLISHER_ID).isEmptyQueue()); }