anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added internal publish once the broker is embedded in hosting applications

Browse Source
This commit is contained in:
andsel
2015-12-08 18:12:00 +01:00
parent 6805cf89ee
commit 16a5556fa3
5 changed files with 374 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChangeLog.txt
+1
View File
@@ -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
broker/src/main/java/io/moquette/server/Server.java
+23
View File
@@ -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");
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+82 -1
View File
@@ -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.
broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java
+251
View File
@@ -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();
}
}
embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java
+17 -1
View File
@@ -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<? extends InterceptHandler> 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");
}
}