diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java index 771f64e3..98d60187 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java @@ -15,22 +15,36 @@ */ package org.eclipse.moquette.spi; +import org.eclipse.moquette.proto.messages.AbstractMessage; import org.eclipse.moquette.spi.impl.events.PublishEvent; +import org.eclipse.moquette.spi.impl.subscriptions.Subscription; +import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.util.HashSet; import java.util.List; +import java.util.Set; /** * @author andrea */ public class ClientSession { + private final static Logger LOG = LoggerFactory.getLogger(ClientSession.class); + public final String clientID; private final IMessagesStore messagesStore; - public ClientSession(String clientID, IMessagesStore messagesStore) { + private final ISessionsStore m_sessionsStore; + + private Set subscriptions = new HashSet<>(); + + public ClientSession(String clientID, IMessagesStore messagesStore, ISessionsStore sessionsStore) { this.clientID = clientID; this.messagesStore = messagesStore; + this.m_sessionsStore = sessionsStore; } //List of client's subscriptions //list of messages not acknowledged by client @@ -56,4 +70,18 @@ public class ClientSession { "clientID='" + clientID + '\'' + '}'; } + + public boolean subscribe(String topicFilter, Subscription newSubscription) { + LOG.info("<{}> subscribed to topicFilter <{}> with QoS {}", + newSubscription.getClientId(), topicFilter, + AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos())); + boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter()); + if (!validTopic) { + //send SUBACK with 0x80 for this topic filter + return false; + } + subscriptions.add(newSubscription); + m_sessionsStore.addNewSubscription(newSubscription); + return true; + } } \ No newline at end of file diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 10374516..fca6e847 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -465,9 +465,8 @@ public class ProtocolProcessor { if (m_clientIDs.get(clientId) == null) { throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs)); } -// LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); m_clientIDs.get(clientId).getSession().write(pubAckMessage); - }catch(Throwable t) { + } catch(Throwable t) { LOG.error(null, t); } } @@ -531,7 +530,6 @@ public class ProtocolProcessor { //cleanup topic subscriptions cleanSession(clientID); } -// m_notifier.disconnect(evt.getSession()); m_clientIDs.remove(clientID); session.close(true); @@ -596,35 +594,36 @@ public class ProtocolProcessor { boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION); LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID()); + ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); //ack the client SubAckMessage ackMessage = new SubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); + List newSubscriptions = new ArrayList<>(); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.getQos()); Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos, cleanSession); - boolean valid = subscribeSingleTopic(newSubscription, req.getTopicFilter()); + //boolean valid = subscribeSingleTopic(newSubscription, req.getTopicFilter()); + boolean valid = clientSession.subscribe(req.getTopicFilter(), newSubscription); ackMessage.addType(valid ? qos : AbstractMessage.QOSType.FAILURE); + if (valid) { + newSubscriptions.add(newSubscription); + } } + //save session, persist subscriptions from session LOG.debug("SUBACK for packetID {}", msg.getMessageID()); session.write(ackMessage); + + //fire the publish + for(Subscription subscription : newSubscriptions) { + subscribeSingleTopic(subscription, subscription.getTopicFilter()); + } } private boolean subscribeSingleTopic(Subscription newSubscription, final String topic) { - LOG.info("<{}> subscribed to topic <{}> with QoS {}", - newSubscription.getClientId(), topic, - AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos())); - boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter()); - if (!validTopic) { - //send SUBACK with 0x80 for this topic filter - return false; - } subscriptions.add(newSubscription); - //notify the Observables - m_interceptor.notifyTopicSubscribed(newSubscription); - //scans retained messages to be published to the new subscription Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() { public boolean match(String key) { @@ -640,6 +639,9 @@ public class ProtocolProcessor { m_messagesStore.nextPacketID(newSubscription.getClientId()); sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID); } + + //notify the Observables + m_interceptor.notifyTopicSubscribed(newSubscription); return true; } } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java index 2648b271..3e80b622 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -20,6 +20,7 @@ import java.util.*; import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.atomic.AtomicReference; +import org.eclipse.moquette.spi.ClientSession; import org.eclipse.moquette.spi.ISessionsStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -163,8 +164,14 @@ public class SubscriptionsStore { return new NodeCouple(newRoot, current); } +// public void save(ClientSession clientSession) { +// for(Subscription subscription : clientSession.getSubscriptions()) { +// add(subscription); +// } +// } + public void add(Subscription newSubscription) { - m_sessionsStore.addNewSubscription(newSubscription); +// m_sessionsStore.addNewSubscription(newSubscription); addDirect(newSubscription); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java index 07e63f87..c0d9b113 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java @@ -309,7 +309,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); m_persistentSubscriptions.put(clientID, new HashSet()); - return new ClientSession(clientID, this); + return new ClientSession(clientID, this, this); } @Override @@ -318,7 +318,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { return null; } - return new ClientSession(clientID, this); + return new ClientSession(clientID, this, this); } @Override diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java index 581ff262..f9df08e8 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java @@ -98,7 +98,7 @@ public class MemorySessionStore implements ISessionsStore { } LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); m_persistentSubscriptions.put(clientID, new HashSet()); - return new ClientSession(clientID, m_messagesStore); + return new ClientSession(clientID, m_messagesStore, this); } @Override @@ -107,12 +107,12 @@ public class MemorySessionStore implements ISessionsStore { return null; } - return new ClientSession(clientID, m_messagesStore); + return new ClientSession(clientID, m_messagesStore, this); } @Override public List listAllSubscriptions() { - List allSubscriptions = new ArrayList(); + List allSubscriptions = new ArrayList<>(); for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { allSubscriptions.addAll(entry.getValue()); } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 36d73182..c08f80d4 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -207,6 +207,7 @@ public class ProtocolProcessorTest { msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); + m_sessionStore.createNewSession(FAKE_CLIENT_ID); m_processor.processSubscribe(m_session, msg/*, FAKE_CLIENT_ID, false*/); //Verify @@ -221,6 +222,7 @@ public class ProtocolProcessorTest { msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); + m_sessionStore.createNewSession(FAKE_CLIENT_ID); assertEquals(0, subscriptions.size()); m_processor.processSubscribe(m_session, msg); @@ -241,6 +243,7 @@ public class ProtocolProcessorTest { msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), BAD_FORMATTED_TOPIC)); m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); + m_sessionStore.createNewSession(FAKE_CLIENT_ID); assertEquals(0, subscriptions.size()); //Exercise diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index d374aec4..e7854754 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -238,7 +238,9 @@ public class SubscriptionsStoreTest { private void assertMatch(String subscription, String topic) { store = new SubscriptionsStore(); - store.init(new MemoryStorageService()); + MemoryStorageService memStore = new MemoryStorageService(); + memStore.initStore(); + store.init(memStore); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); store.add(sub); assertFalse(store.matches(topic).isEmpty()); @@ -246,7 +248,9 @@ public class SubscriptionsStoreTest { private void assertNotMatch(String subscription, String topic) { store = new SubscriptionsStore(); - store.init(new MemoryStorageService()); + MemoryStorageService memStore = new MemoryStorageService(); + memStore.initStore(); + store.init(memStore); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); store.add(sub); assertTrue(store.matches(topic).isEmpty()); @@ -345,7 +349,9 @@ public class SubscriptionsStoreTest { @Test public void removeSubscription_withDifferentClients_subscribedSameTopic() { SubscriptionsStore aStore = new SubscriptionsStore(); - aStore.init(new MemoryStorageService()); + MemoryStorageService memStore = new MemoryStorageService(); + memStore.initStore(); + aStore.init(memStore); //subscribe a not active clientID1 to /topic Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false); aStore.add(slashSub);