Separated ISessionsStore and IMessagesStore Memory implementations
This commit is contained in:
andsel
@@ -0,0 +1,123 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2015 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
package org.eclipse.moquette.spi.impl;
|
||||
|
||||
import org.eclipse.moquette.spi.IMatchingCondition;
|
||||
import org.eclipse.moquette.spi.IMessagesStore;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
import static org.eclipse.moquette.spi.impl.Utils.defaultGet;
|
||||
|
||||
/**
|
||||
* @author andrea
|
||||
*/
|
||||
public class MemoryMessagesStore implements IMessagesStore {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(MemoryMessagesStore.class);
|
||||
|
||||
private Map<String, String> m_retainedStore = new HashMap<>();
|
||||
private Map<String, StoredMessage> m_persistentMessageStore = new HashMap<>();
|
||||
private Map<String, Map<Integer, String>> m_messageToGuids;
|
||||
private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
|
||||
|
||||
MemoryMessagesStore(Map<String, Map<Integer, String>> messageToGuids) {
|
||||
m_messageToGuids = messageToGuids;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void initStore() {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void storeRetained(String topic, String guid) {
|
||||
m_retainedStore.put(topic, guid);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
|
||||
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
|
||||
|
||||
List<StoredMessage> results = new ArrayList<>();
|
||||
|
||||
for (Map.Entry<String, String> entry : m_retainedStore.entrySet()) {
|
||||
final String guid = entry.getValue();
|
||||
StoredMessage storedMsg = m_persistentMessageStore.get(guid);
|
||||
if (condition.match(entry.getKey())) {
|
||||
results.add(storedMsg);
|
||||
}
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String storePublishForFuture(StoredMessage evt) {
|
||||
LOG.debug("storePublishForFuture store evt {}", evt);
|
||||
String guid = UUID.randomUUID().toString();
|
||||
evt.setGuid(guid);
|
||||
m_persistentMessageStore.put(guid, evt);
|
||||
HashMap<Integer, String> guids = (HashMap<Integer, String>) defaultGet(m_messageToGuids,
|
||||
evt.getClientID(), new HashMap<Integer, String>());
|
||||
guids.put(evt.getMessageID(), guid);
|
||||
return guid;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<StoredMessage> listMessagesInSession(Collection<String> guids) {
|
||||
List<StoredMessage> ret = new ArrayList<>();
|
||||
for (String guid : guids) {
|
||||
ret.add(m_persistentMessageStore.get(guid));
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void dropMessagesInSession(String clientID) {
|
||||
m_persistentMessageStore.remove(clientID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public StoredMessage getMessageByGuid(String guid) {
|
||||
return m_persistentMessageStore.get(guid);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the next valid packetIdentifier for the given client session.
|
||||
* */
|
||||
@Override
|
||||
public int nextPacketID(String clientID) {
|
||||
Set<Integer> inFlightForClient = m_inflightIDs.get(clientID);
|
||||
if (inFlightForClient == null) {
|
||||
int nextPacketId = 1;
|
||||
inFlightForClient = new HashSet<>();
|
||||
inFlightForClient.add(nextPacketId);
|
||||
m_inflightIDs.put(clientID, inFlightForClient);
|
||||
return nextPacketId;
|
||||
}
|
||||
int maxId = Collections.max(inFlightForClient);
|
||||
int nextPacketId = (maxId + 1) % 0xFFFF;
|
||||
inFlightForClient.add(nextPacketId);
|
||||
return nextPacketId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cleanRetained(String topic) {
|
||||
m_retainedStore.remove(topic);
|
||||
}
|
||||
}
|
||||
@@ -15,227 +15,29 @@
|
||||
*/
|
||||
package org.eclipse.moquette.spi.impl;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
import org.eclipse.moquette.spi.ClientSession;
|
||||
import org.eclipse.moquette.spi.IMatchingCondition;
|
||||
import org.eclipse.moquette.spi.IMessagesStore;
|
||||
import org.eclipse.moquette.spi.persistence.MemorySessionStore;
|
||||
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
|
||||
|
||||
import org.eclipse.moquette.spi.ISessionsStore;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.eclipse.moquette.spi.persistence.MemorySessionStore;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.eclipse.moquette.spi.impl.Utils.defaultGet;
|
||||
|
||||
/**
|
||||
*/
|
||||
public class MemoryStorageService implements IMessagesStore, ISessionsStore {
|
||||
public class MemoryStorageService {
|
||||
|
||||
private Map<String, String> m_retainedStore = new HashMap<>();
|
||||
//TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map)
|
||||
private Map<String, StoredMessage> m_persistentMessageStore = new HashMap<>();
|
||||
//maps clientID->[MessageId -> guid]
|
||||
private Map<String, Map<Integer, String>> m_inflightStore = new HashMap<>();
|
||||
//maps clientID->[guid*]
|
||||
private Map<String, Set<String>> m_enqueuedStore = new HashMap<>();
|
||||
//maps clientID->[messageID*]
|
||||
private Map<String, Set<Integer>> m_secondPhaseStore = new HashMap<>();
|
||||
private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
|
||||
private Map<String, Map<Integer, String>> m_messageToGuids = new HashMap<>();
|
||||
private MemorySessionStore m_sessionsStore;
|
||||
private MemoryMessagesStore m_messagesStore;
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class);
|
||||
|
||||
public void initStore() {
|
||||
m_sessionsStore = new MemorySessionStore(this);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cleanRetained(String topic) {
|
||||
m_retainedStore.remove(topic);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void storeRetained(String topic, String guid) {
|
||||
m_retainedStore.put(topic, guid);
|
||||
Map<String, Map<Integer, String>> messageToGuids = new HashMap<>();
|
||||
m_messagesStore = new MemoryMessagesStore(messageToGuids);
|
||||
m_sessionsStore = new MemorySessionStore(m_messagesStore, messageToGuids);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
|
||||
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
|
||||
|
||||
List<StoredMessage> results = new ArrayList<>();
|
||||
|
||||
for (Map.Entry<String, String> entry : m_retainedStore.entrySet()) {
|
||||
final String guid = entry.getValue();
|
||||
StoredMessage storedMsg = m_persistentMessageStore.get(guid);
|
||||
if (condition.match(entry.getKey())) {
|
||||
results.add(storedMsg);
|
||||
}
|
||||
}
|
||||
|
||||
return results;
|
||||
public IMessagesStore messagesStore() {
|
||||
return m_messagesStore;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String storePublishForFuture(StoredMessage evt) {
|
||||
LOG.debug("storePublishForFuture store evt {}", evt);
|
||||
String guid = UUID.randomUUID().toString();
|
||||
evt.setGuid(guid);
|
||||
m_persistentMessageStore.put(guid, evt);
|
||||
HashMap<Integer, String> guids = (HashMap<Integer, String>) defaultGet(m_messageToGuids,
|
||||
evt.getClientID(), new HashMap<Integer, String>());
|
||||
guids.put(evt.getMessageID(), guid);
|
||||
return guid;
|
||||
public ISessionsStore sessionsStore() {
|
||||
return m_sessionsStore;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<StoredMessage> listMessagesInSession(Collection<String> guids) {
|
||||
List<StoredMessage> ret = new ArrayList<>();
|
||||
for (String guid : guids) {
|
||||
ret.add(m_persistentMessageStore.get(guid));
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void dropMessagesInSession(String clientID) {
|
||||
m_persistentMessageStore.remove(clientID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public StoredMessage getMessageByGuid(String guid) {
|
||||
return m_persistentMessageStore.get(guid);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String mapToGuid(String clientID, int messageID) {
|
||||
HashMap<Integer, String> guids = (HashMap<Integer, String>) defaultGet(m_messageToGuids,
|
||||
clientID, new HashMap<Integer, String>());
|
||||
return guids.get(messageID);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the next valid packetIdentifier for the given client session.
|
||||
* */
|
||||
@Override
|
||||
public int nextPacketID(String clientID) {
|
||||
Set<Integer> inFlightForClient = m_inflightIDs.get(clientID);
|
||||
if (inFlightForClient == null) {
|
||||
int nextPacketId = 1;
|
||||
inFlightForClient = new HashSet<>();
|
||||
inFlightForClient.add(nextPacketId);
|
||||
m_inflightIDs.put(clientID, inFlightForClient);
|
||||
return nextPacketId;
|
||||
}
|
||||
int maxId = Collections.max(inFlightForClient);
|
||||
int nextPacketId = (maxId + 1) % 0xFFFF;
|
||||
inFlightForClient.add(nextPacketId);
|
||||
return nextPacketId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void addNewSubscription(Subscription newSubscription) {
|
||||
m_sessionsStore.addNewSubscription(newSubscription);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void removeSubscription(String topic, String clientID) {
|
||||
m_sessionsStore.removeSubscription(topic, clientID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void wipeSubscriptions(String sessionID) {
|
||||
m_sessionsStore.wipeSubscriptions(sessionID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void updateSubscriptions(String clientID, Set<Subscription> subscriptions) {
|
||||
m_sessionsStore.updateSubscriptions(clientID, subscriptions);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<Subscription> listAllSubscriptions() {
|
||||
return m_sessionsStore.listAllSubscriptions();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean contains(String clientID) {
|
||||
return m_sessionsStore.contains(clientID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ClientSession createNewSession(String clientID, boolean cleanSession) {
|
||||
return m_sessionsStore.createNewSession(clientID, cleanSession);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ClientSession sessionForClient(String clientID) {
|
||||
return m_sessionsStore.sessionForClient(clientID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void activate(String clientID) {
|
||||
m_sessionsStore.activate(clientID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deactivate(String clientID) {
|
||||
m_sessionsStore.deactivate(clientID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void inFlightAck(String clientID, int messageID) {
|
||||
Map<Integer, String> m = this.m_inflightStore.get(clientID);
|
||||
if (m == null) {
|
||||
LOG.error("Can't find the inFlight record for client <{}>", clientID);
|
||||
return;
|
||||
}
|
||||
m.remove(messageID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void inFlight(String clientID, int messageID, String guid) {
|
||||
Map<Integer, String> m = this.m_inflightStore.get(clientID);
|
||||
if (m == null) {
|
||||
m = new HashMap<>();
|
||||
}
|
||||
m.put(messageID, guid);
|
||||
this.m_inflightStore.put(clientID, m);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void bindToDeliver(String guid, String clientID) {
|
||||
Set<String> guids = defaultGet(m_enqueuedStore, clientID, new HashSet<String>());
|
||||
guids.add(guid);
|
||||
m_enqueuedStore.put(clientID, guids);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Collection<String> enqueued(String clientID) {
|
||||
return defaultGet(m_enqueuedStore, clientID, new HashSet<String>());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void removeEnqueued(String clientID, String guid) {
|
||||
Set<String> guids = defaultGet(m_enqueuedStore, clientID, new HashSet<String>());
|
||||
guids.remove(guid);
|
||||
m_enqueuedStore.put(clientID, guids);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void secondPhaseAcknowledged(String clientID, int messageID) {
|
||||
Set<Integer> messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
|
||||
messageIDs.remove(messageID);
|
||||
m_secondPhaseStore.put(clientID, messageIDs);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void secondPhaseAckWaiting(String clientID, int messageID) {
|
||||
Set<Integer> messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
|
||||
messageIDs.add(messageID);
|
||||
m_secondPhaseStore.put(clientID, messageIDs);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,29 +15,27 @@
|
||||
*/
|
||||
package org.eclipse.moquette.spi.impl;
|
||||
|
||||
import org.eclipse.moquette.interception.InterceptHandler;
|
||||
import org.eclipse.moquette.proto.messages.*;
|
||||
import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType;
|
||||
import org.eclipse.moquette.server.netty.NettyChannel;
|
||||
import org.eclipse.moquette.spi.ClientSession;
|
||||
import org.eclipse.moquette.spi.IMatchingCondition;
|
||||
import org.eclipse.moquette.spi.IMessagesStore;
|
||||
import org.eclipse.moquette.spi.IMessagesStore.StoredMessage;
|
||||
import org.eclipse.moquette.spi.ISessionsStore;
|
||||
import org.eclipse.moquette.spi.impl.security.PermitAllAuthorizator;
|
||||
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
|
||||
import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.eclipse.moquette.interception.InterceptHandler;
|
||||
import org.eclipse.moquette.proto.messages.*;
|
||||
import org.eclipse.moquette.server.netty.NettyChannel;
|
||||
import org.eclipse.moquette.spi.ClientSession;
|
||||
import org.eclipse.moquette.spi.IMatchingCondition;
|
||||
import org.eclipse.moquette.spi.IMessagesStore;
|
||||
import org.eclipse.moquette.spi.IMessagesStore.*;
|
||||
import org.eclipse.moquette.spi.ISessionsStore;
|
||||
import org.eclipse.moquette.spi.persistence.MemorySessionStore;
|
||||
import org.eclipse.moquette.spi.impl.security.PermitAllAuthorizator;
|
||||
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
|
||||
import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
|
||||
|
||||
import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
/**
|
||||
@@ -63,7 +61,7 @@ public class ProtocolProcessorTest {
|
||||
ConnectMessage connMsg;
|
||||
ProtocolProcessor m_processor;
|
||||
|
||||
IMessagesStore m_storageService;
|
||||
IMessagesStore m_messagesStore;
|
||||
ISessionsStore m_sessionStore;
|
||||
SubscriptionsStore subscriptions;
|
||||
MockAuthenticator m_mockAuthenticator;
|
||||
@@ -79,18 +77,18 @@ public class ProtocolProcessorTest {
|
||||
Thread.sleep(300);
|
||||
MemoryStorageService memStorage = new MemoryStorageService();
|
||||
memStorage.initStore();
|
||||
m_storageService = memStorage;
|
||||
m_sessionStore = memStorage;
|
||||
//m_storageService.initStore();
|
||||
m_messagesStore = memStorage.messagesStore();
|
||||
m_sessionStore = memStorage.sessionsStore();
|
||||
//m_messagesStore.initStore();
|
||||
|
||||
Map<String, byte[]> users = new HashMap<>();
|
||||
users.put(TEST_USER, TEST_PWD);
|
||||
m_mockAuthenticator = new MockAuthenticator(users);
|
||||
|
||||
subscriptions = new SubscriptionsStore();
|
||||
subscriptions.init(memStorage);
|
||||
subscriptions.init(memStorage.sessionsStore());
|
||||
m_processor = new ProtocolProcessor();
|
||||
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true,
|
||||
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true,
|
||||
new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
}
|
||||
|
||||
@@ -115,8 +113,8 @@ public class ProtocolProcessorTest {
|
||||
//simulate a connect that register a clientID to an IoSession
|
||||
MemoryStorageService storageService = new MemoryStorageService();
|
||||
storageService.initStore();
|
||||
subs.init(storageService);
|
||||
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
subs.init(storageService.sessionsStore());
|
||||
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
ConnectMessage connectMessage = new ConnectMessage();
|
||||
connectMessage.setProtocolVersion((byte) 3);
|
||||
connectMessage.setClientID(FAKE_CLIENT_ID);
|
||||
@@ -162,8 +160,8 @@ public class ProtocolProcessorTest {
|
||||
//simulate a connect that register a clientID to an IoSession
|
||||
MemoryStorageService storageService = new MemoryStorageService();
|
||||
storageService.initStore();
|
||||
subs.init(storageService);
|
||||
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
subs.init(storageService.sessionsStore());
|
||||
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
|
||||
MockReceiverChannel firstReceiverSession = new MockReceiverChannel();
|
||||
ConnectMessage connectMessage = new ConnectMessage();
|
||||
@@ -320,10 +318,10 @@ public class ProtocolProcessorTest {
|
||||
};
|
||||
MemoryStorageService storageService = new MemoryStorageService();
|
||||
storageService.initStore();
|
||||
subs.init(storageService);
|
||||
subs.init(storageService.sessionsStore());
|
||||
|
||||
//simulate a connect that register a clientID to an IoSession
|
||||
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
ConnectMessage connectMessage = new ConnectMessage();
|
||||
connectMessage.setClientID(FAKE_PUBLISHER_ID);
|
||||
connectMessage.setProtocolVersion((byte) 3);
|
||||
@@ -363,9 +361,9 @@ public class ProtocolProcessorTest {
|
||||
retainedMessage.setRetained(true);
|
||||
retainedMessage.setMessageID(120);
|
||||
retainedMessage.setClientID(FAKE_PUBLISHER_ID);
|
||||
m_storageService.storePublishForFuture(retainedMessage);
|
||||
m_messagesStore.storePublishForFuture(retainedMessage);
|
||||
|
||||
m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
ConnectMessage connectMessage = new ConnectMessage();
|
||||
connectMessage.setClientID(FAKE_PUBLISHER_ID);
|
||||
connectMessage.setProtocolVersion((byte) 3);
|
||||
@@ -374,7 +372,7 @@ public class ProtocolProcessorTest {
|
||||
|
||||
//Verify no messages are still stored
|
||||
Collection<String> guids = m_sessionStore.enqueued(FAKE_PUBLISHER_ID);
|
||||
assertTrue(m_storageService.listMessagesInSession(guids).isEmpty());
|
||||
assertTrue(m_messagesStore.listMessagesInSession(guids).isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -387,7 +385,7 @@ public class ProtocolProcessorTest {
|
||||
List<Subscription> inactiveSubscriptions = Arrays.asList(inactiveSub);
|
||||
when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions);
|
||||
m_processor = new ProtocolProcessor();
|
||||
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
m_processor.init(mockedSubscriptions, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
|
||||
//Exercise
|
||||
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
|
||||
@@ -413,7 +411,7 @@ public class ProtocolProcessorTest {
|
||||
List<Subscription> inactiveSubscriptions = Arrays.asList(inactiveSub);
|
||||
when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions);
|
||||
m_processor = new ProtocolProcessor();
|
||||
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(),
|
||||
m_processor.init(mockedSubscriptions, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(),
|
||||
NO_OBSERVERS_INTERCEPTOR);
|
||||
|
||||
//Exercise
|
||||
@@ -460,7 +458,7 @@ public class ProtocolProcessorTest {
|
||||
m_processor.processPublish(m_session, cleanPubMsg);
|
||||
|
||||
//Verify
|
||||
Collection<IMessagesStore.StoredMessage> messages = m_storageService.searchMatching(new IMatchingCondition() {
|
||||
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
|
||||
public boolean match(String key) {
|
||||
return SubscriptionsStore.matchTopics(key, FAKE_TOPIC);
|
||||
}
|
||||
@@ -476,8 +474,10 @@ public class ProtocolProcessorTest {
|
||||
forwardPublish.setRetained(true);
|
||||
forwardPublish.setMessageID(1);
|
||||
|
||||
IMessagesStore memoryMessageStore = new MemoryStorageService();
|
||||
ISessionsStore sessionsStore = new MemorySessionStore(memoryMessageStore);
|
||||
MemoryStorageService memStore = new MemoryStorageService();
|
||||
memStore.initStore();
|
||||
IMessagesStore memoryMessageStore = memStore.messagesStore();
|
||||
ISessionsStore sessionsStore = memStore.sessionsStore();
|
||||
sessionsStore.createNewSession("Sub A", false).activate();
|
||||
sessionsStore.createNewSession("Sub B", false).activate();
|
||||
|
||||
|
||||
+9
-9
@@ -44,7 +44,7 @@ public class ProtocolProcessor_CONNECT_Test {
|
||||
ConnectMessage connMsg;
|
||||
ProtocolProcessor m_processor;
|
||||
|
||||
IMessagesStore m_storageService;
|
||||
IMessagesStore m_messagesStore;
|
||||
ISessionsStore m_sessionStore;
|
||||
SubscriptionsStore subscriptions;
|
||||
MockAuthenticator m_mockAuthenticator;
|
||||
@@ -60,18 +60,18 @@ public class ProtocolProcessor_CONNECT_Test {
|
||||
Thread.sleep(300);
|
||||
MemoryStorageService memStorage = new MemoryStorageService();
|
||||
memStorage.initStore();
|
||||
m_storageService = memStorage;
|
||||
m_sessionStore = memStorage;
|
||||
//m_storageService.initStore();
|
||||
m_messagesStore = memStorage.messagesStore();
|
||||
m_sessionStore = memStorage.sessionsStore();
|
||||
//m_messagesStore.initStore();
|
||||
|
||||
Map<String, byte[]> users = new HashMap<>();
|
||||
users.put(TEST_USER, TEST_PWD);
|
||||
m_mockAuthenticator = new MockAuthenticator(users);
|
||||
|
||||
subscriptions = new SubscriptionsStore();
|
||||
subscriptions.init(memStorage);
|
||||
subscriptions.init(m_sessionStore);
|
||||
m_processor = new ProtocolProcessor();
|
||||
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true,
|
||||
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true,
|
||||
new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
|
||||
}
|
||||
|
||||
@@ -163,7 +163,7 @@ public class ProtocolProcessor_CONNECT_Test {
|
||||
@Test
|
||||
public void prohibitAnonymousClient() {
|
||||
connMsg.setClientID("123");
|
||||
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
|
||||
//Exercise
|
||||
m_processor.processConnect(m_session, connMsg);
|
||||
@@ -177,7 +177,7 @@ public class ProtocolProcessor_CONNECT_Test {
|
||||
connMsg.setClientID("123");
|
||||
connMsg.setUserFlag(true);
|
||||
connMsg.setUsername(TEST_USER + "_fake");
|
||||
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
|
||||
//Exercise
|
||||
m_processor.processConnect(m_session, connMsg);
|
||||
@@ -189,7 +189,7 @@ public class ProtocolProcessor_CONNECT_Test {
|
||||
@Test
|
||||
public void acceptAnonymousClient() {
|
||||
connMsg.setClientID("123");
|
||||
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
|
||||
|
||||
//Exercise
|
||||
m_processor.processConnect(m_session, connMsg);
|
||||
|
||||
+7
-5
@@ -20,6 +20,7 @@ import java.text.ParseException;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import org.eclipse.moquette.spi.ISessionsStore;
|
||||
import org.eclipse.moquette.spi.impl.MemoryStorageService;
|
||||
import org.eclipse.moquette.proto.messages.AbstractMessage;
|
||||
|
||||
@@ -43,7 +44,7 @@ public class SubscriptionsStoreTest {
|
||||
store = new SubscriptionsStore();
|
||||
MemoryStorageService storageService = new MemoryStorageService();
|
||||
storageService.initStore();
|
||||
store.init(storageService);
|
||||
store.init(storageService.sessionsStore());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -235,7 +236,7 @@ public class SubscriptionsStoreTest {
|
||||
store = new SubscriptionsStore();
|
||||
MemoryStorageService memStore = new MemoryStorageService();
|
||||
memStore.initStore();
|
||||
store.init(memStore);
|
||||
store.init(memStore.sessionsStore());
|
||||
Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false);
|
||||
store.add(sub);
|
||||
assertFalse(store.matches(topic).isEmpty());
|
||||
@@ -245,7 +246,7 @@ public class SubscriptionsStoreTest {
|
||||
store = new SubscriptionsStore();
|
||||
MemoryStorageService memStore = new MemoryStorageService();
|
||||
memStore.initStore();
|
||||
store.init(memStore);
|
||||
store.init(memStore.sessionsStore());
|
||||
Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false);
|
||||
store.add(sub);
|
||||
assertTrue(store.matches(topic).isEmpty());
|
||||
@@ -346,11 +347,12 @@ public class SubscriptionsStoreTest {
|
||||
SubscriptionsStore aStore = new SubscriptionsStore();
|
||||
MemoryStorageService memStore = new MemoryStorageService();
|
||||
memStore.initStore();
|
||||
aStore.init(memStore);
|
||||
ISessionsStore sessionsStore = memStore.sessionsStore();
|
||||
aStore.init(sessionsStore);
|
||||
//subscribe a not active clientID1 to /topic
|
||||
Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false);
|
||||
aStore.add(slashSub);
|
||||
memStore.createNewSession("FAKE_CLI_ID_1", true).deactivate();
|
||||
sessionsStore.createNewSession("FAKE_CLI_ID_1", true).deactivate();
|
||||
|
||||
//subscribe an active clientID2 to /topic
|
||||
Subscription slashSub2 = new Subscription("FAKE_CLI_ID_2", "/topic", AbstractMessage.QOSType.MOST_ONE, false);
|
||||
|
||||
@@ -46,9 +46,11 @@ public class MemorySessionStore implements ISessionsStore {
|
||||
//maps clientID->[messageID*]
|
||||
private Map<String, Set<Integer>> m_secondPhaseStore = new HashMap<>();
|
||||
|
||||
private Map<String, Map<Integer, String>> m_messageToGuids;
|
||||
private final IMessagesStore m_messagesStore;
|
||||
|
||||
public MemorySessionStore(IMessagesStore messagesStore) {
|
||||
public MemorySessionStore(IMessagesStore messagesStore, Map<String, Map<Integer, String>> messageToGuids) {
|
||||
m_messageToGuids = messageToGuids;
|
||||
this.m_messagesStore = messagesStore;
|
||||
}
|
||||
|
||||
@@ -217,6 +219,8 @@ public class MemorySessionStore implements ISessionsStore {
|
||||
|
||||
@Override
|
||||
public String mapToGuid(String clientID, int messageID) {
|
||||
return ((MemoryStorageService) m_messagesStore).mapToGuid(clientID, messageID);
|
||||
HashMap<Integer, String> guids = (HashMap<Integer, String>) defaultGet(m_messageToGuids,
|
||||
clientID, new HashMap<Integer, String>());
|
||||
return guids.get(messageID);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user