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