@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(); }
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 { 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 { 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(); }
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(); }
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();  }