From fdc7f6237e6a7784dce296718afe156dc0904876 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sun, 15 Jun 2014 18:16:24 +0200 Subject: [PATCH] Added memory store for will messages and implemented the correct publishing --- ChangeLog.txt | 2 + .../moquette/messaging/spi/IMessaging.java | 3 ++ .../messaging/spi/impl/ProtocolProcessor.java | 52 ++++++++++++++++-- .../spi/impl/events/PublishEvent.java | 1 + .../netty/MoquetteIdleTimoutHandler.java | 2 + .../server/netty/NettyMQTTHandler.java | 2 +- .../server/ServerIntegrationFuseTest.java | 35 ++++++++++++ ...erverLowlevelMessagesIntegrationTests.java | 53 +++++++++++++++++++ 8 files changed, 145 insertions(+), 5 deletions(-) diff --git a/ChangeLog.txt b/ChangeLog.txt index 495a1ce3..1be85c94 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -3,6 +3,8 @@ Version 0.6: Fix issue #43, missed topic case (finance/+ does has to match finance) Integrated WebSocket support (issue #37) Integrated SSL support (issue #41) + Fixed issue #40, check Logging inside the Felix OSGi + Fixed issue #46, implement correct behaviour for Will stuff Version 0.5: Fix issue 20 (no flush of message on PINGREQ) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 5d93684d..15557d5f 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -21,6 +21,9 @@ import org.dna.mqtt.moquette.server.ServerChannel; /** * Interface to the underling messaging system used to publish, subscribe. * + * It's the abstraction of the messaging stuff attached in after the front protocol + * parsing stuff. + * * @author andrea */ public interface IMessaging { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 8e54fc14..b89948f0 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -36,6 +36,7 @@ import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; import org.dna.mqtt.moquette.proto.messages.PubCompMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.proto.messages.PubAckMessage; @@ -62,12 +63,45 @@ import org.slf4j.LoggerFactory; */ class ProtocolProcessor implements EventHandler { + static final class WillMessage { + private final String topic; + private final ByteBuffer payload; + private final boolean retained; + private final QOSType qos; + + public WillMessage(String topic, ByteBuffer payload, boolean retained, QOSType qos) { + this.topic = topic; + this.payload = payload; + this.retained = retained; + this.qos = qos; + } + + public String getTopic() { + return topic; + } + + public ByteBuffer getPayload() { + return payload; + } + + public boolean isRetained() { + return retained; + } + + public QOSType getQos() { + return qos; + } + + } + private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); private Map m_clientIDs = new HashMap(); private SubscriptionsStore subscriptions; private IStorageService m_storageService; private IAuthenticator m_authenticator; + //maps clientID to Will testament, if specified on CONNECT + private Map m_willStore = new HashMap(); private ExecutorService m_executor; BatchEventProcessor m_eventProcessor; @@ -149,9 +183,9 @@ class ProtocolProcessor implements EventHandler { AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()]; byte[] willPayload = msg.getWillMessage().getBytes(); ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); - PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, - bb, msg.isWillRetain(), msg.getClientID(), session); - processPublish(pubEvt); + //save the will testment in the clientID store + WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos ); + m_willStore.put(msg.getClientID(), will); } //handle user authentication @@ -429,8 +463,10 @@ class ProtocolProcessor implements EventHandler { session.close(true); //de-activate the subscriptions for this ClientID -// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); subscriptions.deactivate(clientID); + //cleanup the will store + m_willStore.remove(clientID); + LOG.info("Disconnected client <{}> with clean session {}", clientID, cleanSession); } @@ -442,6 +478,14 @@ class ProtocolProcessor implements EventHandler { subscriptions.deactivate(clientID); LOG.info("Lost connection with client <{}>", clientID); } + //TODO publish the Will message (if any) for the clientID + if (m_willStore.containsKey(clientID)) { + WillMessage will = m_willStore.get(clientID); + PublishEvent pubEvt = new PublishEvent(will.getTopic(), will.getQos(), + will.getPayload(), will.isRetained(), clientID, null); + processPublish(pubEvt); + m_willStore.remove(clientID); + } } /** diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java index 5d2405d2..7f33dcb8 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java @@ -89,6 +89,7 @@ public class PublishEvent extends MessagingEvent { return m_msgID; } + //TODO remove the session it's never used!! public ServerChannel getSession() { return m_session; } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java index cb2b2dc9..44144b18 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java @@ -26,6 +26,8 @@ class MoquetteIdleTimoutHandler extends ChannelDuplexHandler { if (evt instanceof IdleStateEvent) { IdleState e = ((IdleStateEvent) evt).state(); if (e == IdleState.ALL_IDLE) { + //fire a channelInactive to trigger publish of Will + ctx.fireChannelInactive(); ctx.close(); } /*else if (e.getState() == IdleState.WRITER_IDLE) { ctx.writeAndFlush(new PingMessage()); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java index 50e7bd19..7edf2124 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java @@ -76,7 +76,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { } @Override - public void channelInactive(ChannelHandlerContext ctx)throws Exception { + public void channelInactive(ChannelHandlerContext ctx) throws Exception { NettyChannel channel = m_channelMapper.get(ctx); String clientID = (String) channel.getAttribute(Constants.ATTR_CLIENTID); m_messaging.lostConnection(clientID); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java index a053d9ae..a4bbefe4 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java @@ -93,6 +93,41 @@ public class ServerIntegrationFuseTest { assertTrue(true); } + @Test + public void checkWillTestmaentIsPublishedOnConnectionKill_noRetain() throws Exception { + LOG.info("checkWillTestmaentIsPublishedOnConnectionKill"); + + String willTestamentTopic = "/will/test"; + String willTestamentMsg = "Bye bye"; + + MQTT mqtt = new MQTT(); + mqtt.setHost("localhost", 1883); + mqtt.setClientId("WillTestamentPublisher"); + mqtt.setWillRetain(false); + mqtt.setWillMessage(willTestamentMsg); + mqtt.setWillTopic(willTestamentTopic); + m_publisher = mqtt.blockingConnection(); + m_publisher.connect(); + + m_mqtt.setHost("localhost", 1883); + m_mqtt.setCleanSession(false); + m_mqtt.setClientId("Subscriber"); + m_subscriber = m_mqtt.blockingConnection(); + m_subscriber.connect(); +// Topic[] topics = new Topic[]{new Topic(willTestamentTopic, QoS.AT_LEAST_ONCE)}; + Topic[] topics = new Topic[]{new Topic(willTestamentTopic, QoS.AT_MOST_ONCE)}; + m_subscriber.subscribe(topics); + + //Exercise, kill the publisher connection + m_publisher.kill(); + + //Verify, that the testament is fired + Message msg = m_subscriber.receive(500, TimeUnit.MILLISECONDS); + assertNotNull("We should get notified with 'Will' message", msg); + msg.ack(); + assertEquals(willTestamentMsg, new String(msg.getPayload())); + } + @Test public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws Exception { LOG.info("*** checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1 ***"); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java index 23cd673c..9d6556d9 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -17,20 +17,32 @@ package org.dna.mqtt.moquette.server; import java.io.File; import java.io.IOException; +import java.util.concurrent.TimeUnit; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.testclient.Client; +import org.fusesource.mqtt.client.BlockingConnection; +import org.fusesource.mqtt.client.MQTT; +import org.fusesource.mqtt.client.Message; +import org.fusesource.mqtt.client.QoS; +import org.fusesource.mqtt.client.Topic; import org.junit.After; import org.junit.Before; import org.junit.Test; import static org.junit.Assert.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * * @author andrea */ public class ServerLowlevelMessagesIntegrationTests { + private static final Logger LOG = LoggerFactory.getLogger(ServerLowlevelMessagesIntegrationTests.class); + Server m_server; Client m_client; + MQTT m_subscriberDef; protected void startServer() throws IOException { m_server = new Server(); @@ -41,6 +53,9 @@ public class ServerLowlevelMessagesIntegrationTests { public void setUp() throws Exception { startServer(); m_client = new Client("localhost"); + m_subscriberDef = new MQTT(); + m_subscriberDef.setHost("localhost", 1883); + m_subscriberDef.setClientId("Subscriber"); } @After @@ -68,4 +83,42 @@ public class ServerLowlevelMessagesIntegrationTests { assertTrue(m_client.isConnectionLost()); } + + @Test + public void checkWillMessageIsWiredOnClientKeepAliveExpiry() throws Exception { + LOG.info("*** checkWillMessageIsWiredOnClientKeepAliveExpiry ***"); + String willTestamentTopic = "/will/test"; + String willTestamentMsg = "Bye bye"; + + BlockingConnection willSubscriber = m_subscriberDef.blockingConnection(); + willSubscriber.connect(); + Topic[] topics = new Topic[]{new Topic(willTestamentTopic, QoS.AT_MOST_ONCE)}; + willSubscriber.subscribe(topics); + + int keepAlive = 2; //secs + ConnectMessage connectMessage = new ConnectMessage(); + connectMessage.setProcotolVersion((byte)3); + connectMessage.setClientID("FAKECLNT"); + connectMessage.setKeepAlive(keepAlive); + connectMessage.setWillFlag(true); + connectMessage.setWillMessage(willTestamentMsg); + connectMessage.setWillTopic(willTestamentTopic); + connectMessage.setWillQos((byte) QOSType.MOST_ONE.ordinal()); + + //Execute + m_client.sendMessage(connectMessage); + long connectTime = System.currentTimeMillis(); + + //but after the 2 KEEP ALIVE timeut expires it gets fired, + //NB it's 1,5 * KEEP_ALIVE so 3 secs and some millis to propagate the message + Message msg = willSubscriber.receive(3100, TimeUnit.MILLISECONDS); + long willMessageReceiveTime = System.currentTimeMillis(); + assertNotNull("the will message should be fired after keep alive!", msg); + msg.ack(); + //the will message hasn't to be received before the elapsing of Keep Alive timeout + assertTrue(willMessageReceiveTime - connectTime > 3000); + + assertEquals(willTestamentMsg, new String(msg.getPayload())); + willSubscriber.disconnect(); + } }