From 16a5556fa31d2f3fbec8db9da6abd31d0f65db0f Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 8 Dec 2015 18:12:00 +0100 Subject: [PATCH] Added internal publish once the broker is embedded in hosting applications --- ChangeLog.txt | 1 + .../main/java/io/moquette/server/Server.java | 23 ++ .../moquette/spi/impl/ProtocolProcessor.java | 83 +++++- .../ServerIntegrationEmbeddedPublishTest.java | 251 ++++++++++++++++++ .../testembedded/EmbeddedLauncher.java | 18 +- 5 files changed, 374 insertions(+), 2 deletions(-) create mode 100644 broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java diff --git a/ChangeLog.txt b/ChangeLog.txt index 47e7f32f..414ca6e8 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -15,6 +15,7 @@ Version 0.8: Refactored the storage part to be more rational and use ClientSession concept Added config property 'autosave_interval' (issue #97) Removed leaking ref to parent's node (issue #119 and #126) + Added internal publish once the broker is embedded (issue #113) Version 0.7: Fix issue #38, added aligned to MQTT 3.1.1 diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index f75e047c..dbddee0e 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -17,6 +17,7 @@ package io.moquette.server; import io.moquette.BrokerConstants; import io.moquette.interception.InterceptHandler; +import io.moquette.proto.messages.PublishMessage; import io.moquette.server.config.MemoryConfig; import io.moquette.spi.impl.SimpleMessaging; import io.moquette.server.config.FilesystemConfig; @@ -26,6 +27,7 @@ import io.moquette.spi.impl.ProtocolProcessor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.swing.*; import java.io.File; import java.io.IOException; import java.util.Collections; @@ -42,6 +44,10 @@ public class Server { private ServerAcceptor m_acceptor; + private volatile boolean m_initialized; + + private ProtocolProcessor m_processor; + public static void main(String[] args) throws IOException { final Server server = new Server(); server.startServer(); @@ -111,12 +117,29 @@ public class Server { m_acceptor = new NettyAcceptor(); m_acceptor.initialize(processor, config); + m_processor = processor; + m_initialized = true; + } + + /** + * Use the broker to publish a message. It's intended for embedding applications. + * It can be used only after the server is correctly started with startServer. + * + * @param msg the message to forward. + * @throws IllegalStateException if the server is not yet started + * */ + public void internalPublish(PublishMessage msg) { + if (!m_initialized) { + throw new IllegalStateException("Can't publish on a server is not yet started"); + } + m_processor.internalPublish(msg); } public void stopServer() { LOG.info("Server stopping..."); m_acceptor.close(); SimpleMessaging.getInstance().shutdown(); + m_initialized = false; LOG.info("Server stopped"); } } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index a74c391c..f301ea94 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -303,7 +303,7 @@ public class ProtocolProcessor { route2Subscribers(toStoreMsg); sendPubAck(clientID, messageID); LOG.debug("replying with PubAck to MSG ID {}", messageID); - } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 + } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 guid = m_messagesStore.storePublishForFuture(toStoreMsg); sendPubRec(clientID, messageID); //Next the client will send us a pub rel @@ -328,6 +328,87 @@ public class ProtocolProcessor { } m_interceptor.notifyTopicPublished(msg, clientID); } + + /** + * Intended usage is only for embedded versions of the broker, where the hosting application want to use the + * broker to send a publish message. + * Inspired by {@link #processPublish} but with some changes to avoid security check, and the handshake phases + * for Qos1 and Qos2. + * It also doesn't notifyTopicPublished because using internally the owner should already know where + * it's publishing. + * */ +// public void internalPublish(PublishMessage msg) { +// final AbstractMessage.QOSType qos = msg.getQos(); +// final String topic = msg.getTopicName(); +// LOG.info("embedded PUBLISH on topic <{}> with QoS {}", topic, qos); +// +// String guid = null; +// IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); +// toStoreMsg.setClientID("BROKER_SELF"); +// toStoreMsg.setMessageID(1); +// if (qos == AbstractMessage.QOSType.MOST_ONE || qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS0, QoS1 +// route2Subscribers(toStoreMsg); +// } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 +// guid = m_messagesStore.storePublishForFuture(toStoreMsg); +// +// route2Subscribers(toStoreMsg); +// +// if (toStoreMsg.isRetained()) { +// if (!toStoreMsg.getMessage().hasRemaining()) { +// m_messagesStore.cleanRetained(topic); +// } else { +// m_messagesStore.storeRetained(topic, guid); +// } +// } +// return; +// } +// +// if (msg.isRetainFlag()) { +// if (qos == AbstractMessage.QOSType.MOST_ONE) { +// //QoS == 0 && retain => clean old retained +// m_messagesStore.cleanRetained(topic); +// } else { +// if (!msg.getPayload().hasRemaining()) { +// m_messagesStore.cleanRetained(topic); +// } else { +// if (guid == null) { +// //before wasn't stored +// guid = m_messagesStore.storePublishForFuture(toStoreMsg); +// } +// m_messagesStore.storeRetained(topic, guid); +// } +// } +// } +// } + + public void internalPublish(PublishMessage msg) { + final AbstractMessage.QOSType qos = msg.getQos(); + final String topic = msg.getTopicName(); + LOG.info("embedded PUBLISH on topic <{}> with QoS {}", topic, qos); + + String guid = null; + IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); + toStoreMsg.setClientID("BROKER_SELF"); + toStoreMsg.setMessageID(1); + if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 + guid = m_messagesStore.storePublishForFuture(toStoreMsg); + } + route2Subscribers(toStoreMsg); + + if (!msg.isRetainFlag()) { + return; + } + if (qos == AbstractMessage.QOSType.MOST_ONE || !msg.getPayload().hasRemaining()) { + //QoS == 0 && retain => clean old retained + m_messagesStore.cleanRetained(topic); + return; + } + if (guid == null) { + //before wasn't stored + guid = m_messagesStore.storePublishForFuture(toStoreMsg); + } + m_messagesStore.storeRetained(topic, guid); + } /** * Specialized version to publish will testament message. diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java new file mode 100644 index 00000000..26b78de6 --- /dev/null +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java @@ -0,0 +1,251 @@ +/* + * 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 io.moquette.server; + +import io.moquette.proto.messages.AbstractMessage.QOSType; +import io.moquette.proto.messages.PublishMessage; +import io.moquette.server.config.IConfig; +import io.moquette.server.config.MemoryConfig; +import org.fusesource.mqtt.client.*; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Properties; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.*; + +/** + * Created by andrea on 08/12/15. + */ +public class ServerIntegrationEmbeddedPublishTest { + + private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationEmbeddedPublishTest.class); + + Server m_server; + MQTT m_mqtt; + BlockingConnection m_subscriber; + IConfig m_config; + + protected void startServer() throws IOException { + m_server = new Server(); + final Properties configProps = IntegrationUtils.prepareTestPropeties(); + m_config = new MemoryConfig(configProps); + m_server.startServer(m_config); + } + + @Before + public void setUp() throws Exception { + startServer(); + + m_mqtt = new MQTT(); + m_mqtt.setHost("localhost", 1883); + m_mqtt.setClientId("Subscriber"); + } + + @After + public void tearDown() throws Exception { + if (m_subscriber != null) { + m_subscriber.disconnect(); + } + + m_server.stopServer(); + IntegrationUtils.cleanPersistenceFile(m_config); + } + + private void subscribeToWithQos(String topic, QoS qos) throws Exception { + m_subscriber = m_mqtt.blockingConnection(); + m_subscriber.connect(); + Topic[] topics = new Topic[]{new Topic(topic, qos)}; + m_subscriber.subscribe(topics); + } + + private void internalPublishToWithQosAndRetained(String topic, QOSType qos, boolean retained) { + PublishMessage message = new PublishMessage(); + message.setTopicName(topic); + message.setRetainFlag(retained); + message.setQos(qos); + message.setPayload(ByteBuffer.wrap("Hello world MQTT!!".getBytes())); + m_server.internalPublish(message); + } + + private void verifyNoMessageIsReceived() throws Exception { + Message msg = m_subscriber.receive(1, TimeUnit.SECONDS); + assertNull(msg); + } + + private void verifyMessageIsReceivedSuccessfully() throws Exception { + Message msg = m_subscriber.receive(5, TimeUnit.SECONDS); + msg.ack(); + assertNotNull(msg); + assertEquals("Hello world MQTT!!", new String(msg.getPayload())); + } + + @Test + public void testClientSubscribeAfterNotRetainedQoS0IsSent() throws Exception { + LOG.info("*** testClientSubscribeAfterNotRetainedQoS0IsSent ***"); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.MOST_ONE, false); + subscribeToWithQos("/topic", QoS.AT_MOST_ONCE); + + //Verify + verifyNoMessageIsReceived(); + } + + @Test + public void testClientSubscribeBeforeNotRetainedQoS0IsSent() throws Exception { + LOG.info("*** testClientSubscribeBeforeNotRetainedQoS0 ***"); + + subscribeToWithQos("/topic", QoS.AT_MOST_ONCE); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.MOST_ONE, false); + + //Verify + verifyMessageIsReceivedSuccessfully(); + } + + @Test + public void testClientSubscribeBeforeRetainedQoS0IsSent() throws Exception { + LOG.info("*** testClientSubscribeBeforeRetainedQoS0IsSent ***"); + + subscribeToWithQos("/topic", QoS.AT_MOST_ONCE); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.MOST_ONE, false); + + //Verify + verifyMessageIsReceivedSuccessfully(); + } + + @Test + public void testClientSubscribeAfterRetainedQoS0IsSent() throws Exception { + LOG.info("*** testClientSubscribeAfterRetainedQoS0IsSent ***"); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.MOST_ONE, false); + subscribeToWithQos("/topic", QoS.AT_MOST_ONCE); + + //Verify + verifyNoMessageIsReceived(); + } + + @Test + public void testClientSubscribeBeforeNotRetainedQoS1IsSent() throws Exception { + LOG.info("*** testClientSubscribeBeforeNotRetainedQoS1IsSent ***"); + + subscribeToWithQos("/topic", QoS.AT_LEAST_ONCE); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.LEAST_ONE, false); + + //Verify + verifyMessageIsReceivedSuccessfully(); + } + + @Test + public void testClientSubscribeAfterNotRetainedQoS1IsSent() throws Exception { + LOG.info("*** testClientSubscribeAfterNotRetainedQoS1IsSent ***"); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.LEAST_ONE, false); + subscribeToWithQos("/topic", QoS.AT_LEAST_ONCE); + + //Verify + verifyNoMessageIsReceived(); + } + + @Test + public void testClientSubscribeBeforeRetainedQoS1IsSent() throws Exception { + LOG.info("*** testClientSubscribeBeforeRetainedQoS1IsSent ***"); + + subscribeToWithQos("/topic", QoS.AT_LEAST_ONCE); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.LEAST_ONE, false); + + //Verify + verifyMessageIsReceivedSuccessfully(); + } + + @Test + public void testClientSubscribeAfterRetainedQoS1IsSent() throws Exception { + LOG.info("*** testClientSubscribeAfterRetainedQoS0IsSent ***"); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.LEAST_ONE, false); + subscribeToWithQos("/topic", QoS.AT_LEAST_ONCE); + + //Verify + verifyNoMessageIsReceived(); + } + + @Test + public void testClientSubscribeBeforeNotRetainedQoS2IsSent() throws Exception { + LOG.info("*** testClientSubscribeBeforeNotRetainedQoS2IsSent ***"); + + subscribeToWithQos("/topic", QoS.EXACTLY_ONCE); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.EXACTLY_ONCE, false); + + //Verify + verifyMessageIsReceivedSuccessfully(); + } + + @Test + public void testClientSubscribeAfterNotRetainedQoS2IsSent() throws Exception { + LOG.info("*** testClientSubscribeAfterNotRetainedQoS2IsSent ***"); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.EXACTLY_ONCE, false); + subscribeToWithQos("/topic", QoS.EXACTLY_ONCE); + + //Verify + verifyNoMessageIsReceived(); + } + + @Test + public void testClientSubscribeBeforeRetainedQoS2IsSent() throws Exception { + LOG.info("*** testClientSubscribeBeforeRetainedQoS2IsSent ***"); + + subscribeToWithQos("/topic", QoS.EXACTLY_ONCE); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.EXACTLY_ONCE, false); + + //Verify + verifyMessageIsReceivedSuccessfully(); + } + + @Test + public void testClientSubscribeAfterRetainedQoS2IsSent() throws Exception { + LOG.info("*** testClientSubscribeAfterRetainedQoS2IsSent ***"); + + //Exercise + internalPublishToWithQosAndRetained("/topic", QOSType.EXACTLY_ONCE, false); + subscribeToWithQos("/topic", QoS.EXACTLY_ONCE); + + //Verify + verifyNoMessageIsReceived(); + } +} diff --git a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java index a6644514..be307f87 100644 --- a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java +++ b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java @@ -18,11 +18,14 @@ package io.moquette.testembedded; import io.moquette.interception.AbstractInterceptHandler; import io.moquette.interception.InterceptHandler; import io.moquette.interception.messages.*; +import io.moquette.proto.messages.AbstractMessage; +import io.moquette.proto.messages.PublishMessage; import io.moquette.server.Server; import io.moquette.server.config.IConfig; import io.moquette.server.config.ClasspathConfig; import java.io.IOException; +import java.nio.ByteBuffer; import java.util.List; import static java.util.Arrays.asList; @@ -42,7 +45,8 @@ import static java.util.Arrays.asList; final Server mqttBroker = new Server(); List userHandlers = asList(new PublisherListener()); mqttBroker.startServer(classPathConfig, userHandlers); - System.out.println("Broker started press [CRTL+C] to stop"); + + System.out.println("Broker started press [CTRL+C] to stop"); //Bind a shutdown hook Runtime.getRuntime().addShutdownHook(new Thread() { @Override @@ -52,5 +56,17 @@ import static java.util.Arrays.asList; System.out.println("Broker stopped"); } }); + + Thread.sleep(20000); + System.out.println("Before self publish"); + PublishMessage message = new PublishMessage(); + message.setTopicName("/exit"); + message.setRetainFlag(true); +// message.setQos(AbstractMessage.QOSType.MOST_ONE); +// message.setQos(AbstractMessage.QOSType.LEAST_ONE); + message.setQos(AbstractMessage.QOSType.EXACTLY_ONCE); + message.setPayload(ByteBuffer.wrap("Hello World!!".getBytes())); + mqttBroker.internalPublish(message); + System.out.println("After self publish"); } } \ No newline at end of file