@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);
    // Verify no messages are still stored
    assertTrue(this.sessionsRepository.sessionForClient(FAKE_PUBLISHER_ID).isEmptyQueue());
}