From 715937a394a89b7d1a647e3619da88bd2a36f9b3 Mon Sep 17 00:00:00 2001 From: andrea Date: Tue, 10 Mar 2015 22:52:08 +0100 Subject: [PATCH] Implemented packetID generation for QoS 1,2 --- .../moquette/server/netty/NettyChannel.java | 2 +- .../eclipse/moquette/spi/IMessagesStore.java | 11 +- .../spi/impl/MemoryStorageService.java | 32 ++++- .../moquette/spi/impl/ProtocolProcessor.java | 123 +++++++++--------- .../spi/persistence/MapDBPersistentStore.java | 42 +++++- .../server/ServerIntegrationPahoTest.java | 1 - 6 files changed, 134 insertions(+), 77 deletions(-) diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java index c8b77379..e89d0361 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java @@ -84,6 +84,6 @@ public class NettyChannel implements ServerChannel { @Override public String toString() { String clientID = (String) getAttribute(Constants.ATTR_CLIENTID); - return "session [clientID "+ clientID +"]"; + return "session [clientID: "+ clientID +"]" + super.toString(); } } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java index 73569aa1..489a77e7 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java @@ -77,13 +77,18 @@ public interface IMessagesStore { */ List listMessagesInSession(String clientID); - void removeMessageInSession(String clientID, int messageID); + void removeMessageInSession(String clientID, int packetID); void dropMessagesInSession(String clientID); - void cleanInFlight(String msgID); + void cleanInFlight(String clientID, int packetID); - void addInFlight(PublishEvent evt, String publishKey); + void addInFlight(PublishEvent evt, String clientID, int packetID); + + /** + * Return the next valid packetIdentifer for the given client session. + * */ + int nextPacketID(String clientID); void close(); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index 4fa0bb77..cdea6f19 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java @@ -39,6 +39,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { //TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map) private Map> m_persistentMessageStore = new HashMap>(); private Map m_inflightStore = new HashMap(); + private Map> m_inflightIDs = new HashMap<>(); private Map m_qos2Store = new HashMap(); private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class); @@ -119,15 +120,40 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } @Override - public void cleanInFlight(String msgID) { - m_inflightStore.remove(msgID); + public void cleanInFlight(String clientID, int packetID) { + String publishKey = String.format("%s%d", clientID, packetID); + m_inflightStore.remove(publishKey); + Set inFlightForClient = m_inflightIDs.get(clientID); + if (inFlightForClient != null) { + inFlightForClient.remove(packetID); + } } @Override - public void addInFlight(PublishEvent evt, String publishKey) { + public void addInFlight(PublishEvent evt, String clientID, int packetID) { + String publishKey = String.format("%s%d", clientID, packetID); m_inflightStore.put(publishKey, evt); } + /** + * Return the next valid packetIdentifer for the given client session. + * */ + @Override + public int nextPacketID(String clientID) { + Set 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, String clientID) { if (!m_persistentSubscriptions.containsKey(clientID)) { 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 2e972f02..19663a7e 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 @@ -19,10 +19,7 @@ import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.dsl.Disruptor; import java.nio.ByteBuffer; -import java.util.Collection; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import org.eclipse.moquette.spi.IMatchingCondition; @@ -141,7 +138,7 @@ class ProtocolProcessor implements EventHandler { @MQTTMessage(message = ConnectMessage.class) void processConnect(ServerChannel session, ConnectMessage msg) { - LOG.debug("processConnect for client {}", msg.getClientID()); + LOG.debug("CONNECT for client <{}>", msg.getClientID()); if (msg.getProcotolVersion() != VERSION_3_1 && msg.getProcotolVersion() != VERSION_3_1_1) { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); @@ -182,6 +179,7 @@ class ProtocolProcessor implements EventHandler { session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession()); //used to track the client in the subscription and publishing phases. session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID()); + LOG.debug("Conent create session <{}>", session); session.setIdleTime(Math.round(keepAlive * 1.5f)); @@ -223,13 +221,11 @@ class ProtocolProcessor implements EventHandler { if (!msg.isCleanSession() && m_sessionsStore.contains(msg.getClientID())) { okResp.setSessionPresent(true); } - LOG.debug("processConnect sent OK ConnAck"); session.write(okResp); - LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession()); - - LOG.info("Create persistent session for clientID {}", msg.getClientID()); + + LOG.info("Create persistent session for clientID <{}>", msg.getClientID()); m_sessionsStore.addNewSubscription(Subscription.createEmptySubscription(msg.getClientID(), true), msg.getClientID()); //null means EmptySubscription - + LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession()); if (!msg.isCleanSession()) { //force the republish of stored QoS1 and QoS2 republishStoredInSession(msg.getClientID()); @@ -240,7 +236,7 @@ class ProtocolProcessor implements EventHandler { * Republish QoS1 and QoS2 messages stored into the session for the clientID. * */ private void republishStoredInSession(String clientID) { - LOG.trace("republishStoredInSession invoked"); + LOG.trace("republishStoredInSession for client <{}>", clientID); List publishedEvents = m_messagesStore.listMessagesInSession(clientID); if (publishedEvents.isEmpty()) { LOG.info("No stored messages for client <{}>", clientID); @@ -285,37 +281,30 @@ class ProtocolProcessor implements EventHandler { } private void processPublish(String clientID, String topic, QOSType qos, ByteBuffer message, boolean retain, Integer messageID) { - LOG.info("Publish received from clientID <{}> on topic <{}> with QoS {}", - clientID, topic, qos); + LOG.info("PUBLISH from clientID <{}> on topic <{}> with QoS {}", clientID, topic, qos); - String publishKey = null; - if (qos == AbstractMessage.QOSType.LEAST_ONE) { - //store the temporary message - publishKey = String.format("%s%d", clientID, messageID); + if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0 + forward2Subscribers(topic, qos, message, retain, messageID); + } else if (qos == AbstractMessage.QOSType.LEAST_ONE) { + String publishKey = String.format("%s%d", clientID, messageID); PublishEvent inFlightEvt = new PublishEvent(topic, qos, message, retain, clientID, messageID); - m_messagesStore.addInFlight(inFlightEvt, publishKey); + //TODO use a message store for TO PUBLISH MESSAGES it has nothing to do with inFlight!! + m_messagesStore.addInFlight(inFlightEvt, clientID, messageID); + forward2Subscribers(topic, qos, message, retain, messageID); + m_messagesStore.cleanInFlight(clientID, messageID); + //NB the PUB_ACK could be sent also after the addInFlight + sendPubAck(new PubAckEvent(messageID, clientID)); + LOG.debug("replying with PubAck to MSG ID {}", messageID); } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { - publishKey = String.format("%s%d", clientID, messageID); + String publishKey = String.format("%s%d", clientID, messageID); //store the message in temp store PublishEvent qos2Persistent = new PublishEvent(topic, qos, message, retain, clientID, messageID); m_messagesStore.persistQoS2Message(publishKey, qos2Persistent); sendPubRec(clientID, messageID); - } - - //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher - if (qos != AbstractMessage.QOSType.EXACTLY_ONCE) { - publish2Subscribers(topic, qos, message, retain, messageID); - } - - if (qos == AbstractMessage.QOSType.LEAST_ONE) { - if (publishKey == null) { - throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + messageID); - } - m_messagesStore.cleanInFlight(publishKey); - sendPubAck(new PubAckEvent(messageID, clientID)); - LOG.debug("replying with PubAck to MSG ID {}", messageID); + //Next the client will send us a pub rel + //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher } if (retain) { @@ -331,19 +320,23 @@ class ProtocolProcessor implements EventHandler { /** * Specialized version to publish will testament message. */ - private void processPublish(WillMessage will, String clientID) { + private void forwardPublishWill(WillMessage will, String clientID) { + //TODO reimplement it! + //TODO it has just to publish the message downstream to the subscribers final String topic = will.getTopic(); final AbstractMessage.QOSType qos = will.getQos(); final ByteBuffer message = will.getPayload(); boolean retain = will.isRetained(); - processPublish(clientID, topic, qos, message, retain, null); + //NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1 + processPublish(clientID, topic, qos, message, retain, 1 /*null*/); } /** * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2 * */ - private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer origMessage, boolean retain, Integer messageID) { - LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic); + private void forward2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer origMessage, + boolean retain, Integer messageID) { + LOG.debug("forward2Subscribers republishing to existing subscribers that matches the topic {}", topic); if (LOG.isDebugEnabled()) { LOG.debug("content <{}>", DebugUtils.payload2Str(origMessage)); LOG.debug("subscription tree {}", subscriptions.dumpTree()); @@ -353,13 +346,13 @@ class ProtocolProcessor implements EventHandler { qos = sub.getRequestedQos(); } - ByteBuffer message = origMessage.duplicate(); - LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", + LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", sub.getClientId(), sub.getTopicFilter(), qos, sub.isActive()); - + ByteBuffer message = origMessage.duplicate(); if (qos == AbstractMessage.QOSType.MOST_ONE && sub.isActive()) { //QoS 0 - sendPublish(sub.getClientId(), topic, qos, message, false); + //forwardPublishQoS0(sub.getClientId(), topic, qos, message, false); + sendPublish(sub.getClientId(), topic, qos, message, false, null); } else { //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it @@ -368,28 +361,23 @@ class ProtocolProcessor implements EventHandler { PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0); m_messagesStore.storePublishForFuture(newPublishEvt); } else { + //TODO also QoS 1 has to be stored in Flight Zone //if QoS 2 then store it in temp memory if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { - String publishKey = String.format("%s%d", sub.getClientId(), messageID); PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0); - m_messagesStore.addInFlight(newPublishEvt, publishKey); + m_messagesStore.addInFlight(newPublishEvt, sub.getClientId(), messageID); } //publish if (sub.isActive()) { - sendPublish(sub.getClientId(), topic, qos, message, false); + int messageId = m_messagesStore.nextPacketID(sub.getClientId()); + sendPublish(sub.getClientId(), topic, qos, message, false, messageId); } } } } } - - private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) { - //TODO pay attention to the message ID can't be 0 and it's the message sent to subscriber - int messageID = 1; - sendPublish(clientId, topic, qos, message, retained, messageID); - } - - private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, int messageID) { + + private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) { LOG.debug("sendPublish invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", clientId, topic, qos, retained, messageID); PublishMessage pubMessage = new PublishMessage(); pubMessage.setRetainFlag(retained); @@ -401,8 +389,13 @@ class ProtocolProcessor implements EventHandler { if (LOG.isDebugEnabled()) { LOG.debug("content <{}>", DebugUtils.payload2Str(message)); } + //set the PacketIdentifier only for QoS > 0 if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) { pubMessage.setMessageID(messageID); + } else { + if (messageID != null) { + throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message with PacketIdentifier: " + messageID); + } } if (m_clientIDs == null) { @@ -413,7 +406,7 @@ class ProtocolProcessor implements EventHandler { throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs)); } LOG.debug("Session for clientId {} is {}", clientId, m_clientIDs.get(clientId).getSession()); -// m_clientIDs.get(clientId).getSession().write(pubMessage); +// m_clientIDs.get(clientId).getSession().write(pubMessage); disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubMessage)); } @@ -465,7 +458,7 @@ class ProtocolProcessor implements EventHandler { final String topic = evt.getTopic(); final AbstractMessage.QOSType qos = evt.getQos(); - publish2Subscribers(topic, qos, evt.getMessage(), evt.isRetain(), evt.getMessageID()); + forward2Subscribers(topic, qos, evt.getMessage(), evt.isRetain(), evt.getMessageID()); m_messagesStore.removeQoS2Message(publishKey); @@ -496,7 +489,8 @@ class ProtocolProcessor implements EventHandler { pubRelMessage.setQos(AbstractMessage.QOSType.LEAST_ONE); // m_clientIDs.get(clientID).getSession().write(pubRelMessage); - disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRelMessage)); + //disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRelMessage)); + session.write(pubRelMessage); } @MQTTMessage(message = PubCompMessage.class) @@ -505,8 +499,7 @@ class ProtocolProcessor implements EventHandler { int messageID = msg.getMessageID(); LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID); //once received the PUBCOMP then remove the message from the temp memory - String publishKey = String.format("%s%d", clientID, messageID); - m_messagesStore.cleanInFlight(publishKey); + m_messagesStore.cleanInFlight(clientID, messageID); } @MQTTMessage(message = DisconnectMessage.class) @@ -526,7 +519,7 @@ class ProtocolProcessor implements EventHandler { //cleanup the will store m_willStore.remove(clientID); - LOG.info("Disconnected client <{}> with clean session {}", clientID, cleanSession); + LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession); } void processConnectionLost(LostConnectionEvent evt) { @@ -548,7 +541,7 @@ class ProtocolProcessor implements EventHandler { //publish the Will message (if any) for the clientID if (m_willStore.containsKey(clientID)) { WillMessage will = m_willStore.get(clientID); - processPublish(will, clientID); + forwardPublishWill(will, clientID); m_willStore.remove(clientID); } } @@ -562,7 +555,7 @@ class ProtocolProcessor implements EventHandler { List topics = msg.topicFilters(); int messageID = msg.getMessageID(); String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); - LOG.debug("processUnsubscribe invoked, removing subscription on topics {}, for clientID <{}>", topics, clientID); + LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); for (String topic : topics) { subscriptions.removeSubscription(topic, clientID); @@ -579,7 +572,7 @@ class ProtocolProcessor implements EventHandler { void processSubscribe(ServerChannel session, SubscribeMessage msg) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION); - LOG.debug("processSubscribe invoked from client {} with msgID {}", clientID, msg.getMessageID()); + LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID()); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; @@ -597,7 +590,7 @@ class ProtocolProcessor implements EventHandler { ackMessage.addType(qos); } - LOG.debug("replying with SubAck to MSG ID {}", msg.getMessageID()); + LOG.debug("SUBACK for packetID {}", msg.getMessageID()); session.write(ackMessage); } @@ -619,7 +612,10 @@ class ProtocolProcessor implements EventHandler { for (IMessagesStore.StoredMessage storedMsg : messages) { //fire the as retained the message LOG.debug("send publish message for topic {}", topic); - sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true); + //forwardPublishQoS0(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true); + Integer packetID = storedMsg.getQos() == QOSType.MOST_ONE ? null : + m_messagesStore.nextPacketID(newSubscription.getClientId()); + sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID); } } @@ -637,6 +633,7 @@ class ProtocolProcessor implements EventHandler { MessagingEvent evt = t.getEvent(); //It's always of type OutputMessagingEvent OutputMessagingEvent outEvent = (OutputMessagingEvent) evt; + LOG.debug("Output event, sending {}", outEvent.getMessage()); outEvent.getChannel().write(outEvent.getMessage()); } 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 c46cfc43..71abcf07 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 @@ -50,6 +50,8 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { private ConcurrentMap> m_persistentMessageStore; //bind clientID+MsgID -> evt message published private ConcurrentMap m_inflightStore; + //map clientID <-> set of currently in flight packet identifiers + Map> m_inFlightIds; //bind clientID+MsgID -> evt message published private ConcurrentMap m_qos2Store; //persistent Map of clientID, set of Subscriptions @@ -66,11 +68,11 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { LOG.error(null, ex); throw new MQTTException("Can't create temp file for subscriptions storage [" + STORAGE_FILE_PATH + "]", ex); } - m_db = DBMaker.newFileDB(tmpFile) - .make(); + m_db = DBMaker.newFileDB(tmpFile).make(); m_retainedStore = m_db.getHashMap("retained"); m_persistentMessageStore = m_db.getHashMap("persistedMessages"); m_inflightStore = m_db.getHashMap("inflight"); + m_inFlightIds = m_db.getHashMap("inflightPacketIDs"); m_persistentSubscriptions = m_db.getHashMap("subscriptions"); m_qos2Store = m_db.getHashMap("qos2Store"); } @@ -115,7 +117,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { List storedEvents; String clientID = evt.getClientID(); if (!m_persistentMessageStore.containsKey(clientID)) { - storedEvents = new ArrayList(); + storedEvents = new ArrayList<>(); } else { storedEvents = m_persistentMessageStore.get(clientID); } @@ -159,17 +161,45 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { m_db.commit(); } - public void cleanInFlight(String msgID) { - m_inflightStore.remove(msgID); + //----------------- In flight methods ----------------- + @Override + public void cleanInFlight(String clientID, int packetID) { + String publishKey = String.format("%s%d", clientID, packetID); + m_inflightStore.remove(publishKey); + Set inFlightForClient = this.m_inFlightIds.get(clientID); + if (inFlightForClient != null) { + inFlightForClient.remove(packetID); + } m_db.commit(); } - public void addInFlight(PublishEvent evt, String publishKey) { + @Override + public void addInFlight(PublishEvent evt, String clientID, int packetID) { + String publishKey = String.format("%s%d", clientID, packetID); StoredPublishEvent storedEvt = convertToStored(evt); m_inflightStore.put(publishKey, storedEvt); m_db.commit(); } + /** + * Return the next valid packetIdentifer for the given client session. + * */ + @Override + public int nextPacketID(String clientID) { + Set inFlightForClient = this.m_inFlightIds.get(clientID); + if (inFlightForClient == null) { + int nextPacketId = 1; + inFlightForClient = new HashSet<>(); + inFlightForClient.add(nextPacketId); + this.m_inFlightIds.put(clientID, inFlightForClient); + return nextPacketId; + } + int maxId = Collections.max(inFlightForClient); + int nextPacketId = (maxId + 1) % 0xFFFF; + inFlightForClient.add(nextPacketId); + return nextPacketId; + } + public void addNewSubscription(Subscription newSubscription, String clientID) { LOG.debug("addNewSubscription invoked with subscription {} for client {}", newSubscription, clientID); if (!m_persistentSubscriptions.containsKey(clientID)) { diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java index 36ed2d1d..11977692 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java @@ -173,7 +173,6 @@ public class ServerIntegrationPahoTest { LOG.info("*** testRetain_maintainMessage_againstClientDestruction ***"); m_client.connect(); m_client.publish("/topic", "Test my payload".getBytes(), 1, true); - m_client.disconnect(); //reconnect and publish