From eec4dc1779f89e5ef3ae4baf3e8ce1b55078e6ec Mon Sep 17 00:00:00 2001 From: mackristof Date: Tue, 28 Jun 2016 17:27:35 +0200 Subject: [PATCH 1/4] feat(cluster): add default clustering feature --- broker/pom.xml | 6 + .../java/io/moquette/BrokerConstants.java | 2 + .../HazelcastInterceptHandler.java | 33 ++++ .../moquette/interception/HazelcastMsg.java | 44 ++++++ .../io/moquette/server/HazelcastListener.java | 45 ++++++ .../main/java/io/moquette/server/Server.java | 52 ++++++- .../moquette/spi/impl/ProtocolProcessor.java | 31 +++- .../io/moquette/spi/impl/SimpleMessaging.java | 16 +- ...rationHazelcastHandlerInterceptorTest.java | 146 ++++++++++++++++++ .../src/test/resources/config/hazelcast.xml | 10 ++ broker/src/test/resources/hazelcast.xml | 10 ++ broker/src/test/resources/log4j.properties | 4 + distribution/src/main/assembly/assembly.xml | 5 + distribution/src/main/resources/hazelcast.xml | 10 ++ .../parser/proto/messages/PublishMessage.java | 18 +++ 15 files changed, 421 insertions(+), 11 deletions(-) create mode 100644 broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java create mode 100644 broker/src/main/java/io/moquette/interception/HazelcastMsg.java create mode 100644 broker/src/main/java/io/moquette/server/HazelcastListener.java create mode 100644 broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java create mode 100644 broker/src/test/resources/config/hazelcast.xml create mode 100644 broker/src/test/resources/hazelcast.xml create mode 100644 distribution/src/main/resources/hazelcast.xml diff --git a/broker/pom.xml b/broker/pom.xml index 2838e6e9..ab6dd37e 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -101,6 +101,12 @@ 2.1.4 + + com.hazelcast + hazelcast + 3.5 + + junit diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 3031b7fe..5a04e0a4 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -21,6 +21,7 @@ import java.io.File; * Created by andrea on 08/12/15. */ public class BrokerConstants { + public static final String INTERCEPT_HANDLER_PROPERTY_NAME ="intercept.handler"; public static final String PERSISTENT_STORE_PROPERTY_NAME = "persistent_store"; public static final String AUTOSAVE_INTERVAL_PROPERTY_NAME = "autosave_interval"; public static final String PASSWORD_FILE_PROPERTY_NAME = "password_file"; @@ -47,4 +48,5 @@ public class BrokerConstants { public static final String DISABLED_PORT_BIND = "disabled"; public static final String HOST = "0.0.0.0"; public static final String NEED_CLIENT_AUTH = "need_client_auth"; + public static final String HAZELCAST_CONFIGURATION = "hazelcast.configuration"; } diff --git a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java new file mode 100644 index 00000000..75c58c3f --- /dev/null +++ b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java @@ -0,0 +1,33 @@ +package io.moquette.interception; + +import com.hazelcast.core.HazelcastInstance; +import com.hazelcast.core.ITopic; +import io.moquette.interception.messages.InterceptPublishMessage; +import io.moquette.server.Server; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Created by mackristof on 28/05/2016. + */ +public class HazelcastInterceptHandler extends AbstractInterceptHandler { + + private static final Logger LOG = LoggerFactory.getLogger(HazelcastInterceptHandler.class); + private final HazelcastInstance hz; + + + public HazelcastInterceptHandler (Server server){ + this.hz = server.getHazelcastInstance(); + } + + + + @Override + public void onPublish(InterceptPublishMessage msg) { + LOG.info(String.format("%s publish on %s message : %s\n", msg.getClientID(), msg.getTopicName(), new String(msg.getPayload().array()))); + ITopic topic = hz.getTopic("moquette"); + HazelcastMsg hazelcastMsg = new HazelcastMsg(msg.getClientID(),msg.getTopicName(), msg.getQos().byteValue(), msg.getPayload().array(), msg.getUsername()); + topic.publish(hazelcastMsg); + } + +} diff --git a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java new file mode 100644 index 00000000..61f4ca20 --- /dev/null +++ b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java @@ -0,0 +1,44 @@ +package io.moquette.interception; + +import java.io.Serializable; + +/** + * Created by mackristof on 28/05/2016. + */ +public class HazelcastMsg implements Serializable{ + + private final String clientId; + private final String userName; + private final byte qos; + private final String payload; + private final String topic; + + public HazelcastMsg(String clientId, String topic , byte qos, byte[] payload, String userName) { + this.clientId = clientId; + this.qos = qos; + this.payload = new String(payload); + this.userName = userName; + this.topic = topic; + } + + + public String getClientId() { + return clientId; + } + + public byte getQos() { + return qos; + } + + public String getPayload() { + return payload; + } + + public String getUserName() { + return userName; + } + + public String getTopic() { + return topic; + } +} diff --git a/broker/src/main/java/io/moquette/server/HazelcastListener.java b/broker/src/main/java/io/moquette/server/HazelcastListener.java new file mode 100644 index 00000000..64acad59 --- /dev/null +++ b/broker/src/main/java/io/moquette/server/HazelcastListener.java @@ -0,0 +1,45 @@ +package io.moquette.server; + +import com.hazelcast.core.Message; +import com.hazelcast.core.MessageListener; +import io.moquette.interception.HazelcastMsg; +import io.moquette.parser.proto.messages.AbstractMessage; +import io.moquette.parser.proto.messages.PublishMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.ByteBuffer; + +/** + * Created by mackristof on 28/05/2016. + */ +public class HazelcastListener implements MessageListener { + private static final Logger LOG = LoggerFactory.getLogger(HazelcastListener.class); + + private final Server server; + + public HazelcastListener(Server server){ + this.server = server; + } + + + + @Override + public void onMessage(Message msg) { + try { + if (!msg.getPublishingMember().equals(server.getHazelcastInstance().getCluster().getLocalMember())) { + LOG.info(String.format("%s received from hazelcast for topic %s message : %s\n", msg.getMessageObject().getClientId(), msg.getMessageObject().getTopic(), msg.getMessageObject().getPayload())); + PublishMessage publishMessage = new PublishMessage(); + publishMessage.setTopicName(msg.getMessageObject().getTopic()); + publishMessage.setQos(AbstractMessage.QOSType.valueOf(msg.getMessageObject().getQos())); + publishMessage.setPayload(ByteBuffer.wrap(msg.getMessageObject().getPayload().getBytes("UTF-8"))); + publishMessage.setLocal(false); + publishMessage.setClientId(msg.getMessageObject().getClientId()); + server.internalPublish(publishMessage); + + } + } catch (Exception e) { + LOG.error("error while wait for polling hazelcast msg queue",e); + } + } +} diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index 92538f39..f57a9fd9 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -15,15 +15,26 @@ */ package io.moquette.server; +import com.hazelcast.config.ClasspathXmlConfig; +import com.hazelcast.config.Config; +import com.hazelcast.config.FileSystemXmlConfig; +import com.hazelcast.core.Hazelcast; +import com.hazelcast.core.HazelcastInstance; +import com.hazelcast.core.HazelcastInstanceNotActiveException; +import com.hazelcast.core.ITopic; import io.moquette.BrokerConstants; +import io.moquette.interception.HazelcastInterceptHandler; +import io.moquette.interception.HazelcastMsg; import io.moquette.interception.InterceptHandler; import io.moquette.parser.proto.messages.PublishMessage; import io.moquette.server.config.MemoryConfig; import io.moquette.spi.impl.SimpleMessaging; import io.moquette.server.config.FilesystemConfig; import io.moquette.server.config.IConfig; +import io.moquette.server.config.MemoryConfig; import io.moquette.server.netty.NettyAcceptor; import io.moquette.spi.impl.ProtocolProcessor; +import io.moquette.spi.impl.SimpleMessaging; import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthorizator; import io.moquette.spi.security.ISslContextCreator; @@ -50,6 +61,9 @@ public class Server { private ProtocolProcessor m_processor; + private HazelcastInstance hazelcastInstance; + + public static void main(String[] args) throws IOException { final Server server = new Server(); server.startServer(); @@ -122,7 +136,23 @@ public class Server { config.setProperty("intercept.handler", handlerProp); } LOG.info("Persistent store file: " + config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME)); - final ProtocolProcessor processor = SimpleMessaging.getInstance().init(config, handlers, authenticator, authorizator); + if (config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME) !=null && config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME).equals(HazelcastInterceptHandler.class.getCanonicalName())) { + if (config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION) !=null){ + Config hzconfig = null; + if (this.getClass().getClassLoader().getResource(config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION)) != null) { + hzconfig = new ClasspathXmlConfig(config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION)); + } else { + hzconfig = new FileSystemXmlConfig(config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION)); + } + LOG.info(String.format("starting server with hazelcast configuration file : %s",hzconfig)); + hazelcastInstance = Hazelcast.newHazelcastInstance(hzconfig); + } else { + LOG.info("starting server with hazelcast default file"); + hazelcastInstance = Hazelcast.newHazelcastInstance(); + } + listenOnHazelCastMsg(); + } + final ProtocolProcessor processor = SimpleMessaging.getInstance().init(config, handlers, authenticator, authorizator, this); if (sslCtxCreator == null) { sslCtxCreator = new DefaultMoquetteSslContextCreator(config); @@ -132,6 +162,18 @@ public class Server { m_acceptor.initialize(processor, config, sslCtxCreator); m_processor = processor; m_initialized = true; + + + } + + private void listenOnHazelCastMsg() { + HazelcastInstance hz = getHazelcastInstance(); + ITopic topic = hz.getTopic("moquette"); + topic.addMessageListener(new HazelcastListener(this)); + } + + public HazelcastInstance getHazelcastInstance(){ + return hazelcastInstance; } /** @@ -153,6 +195,14 @@ public class Server { m_acceptor.close(); SimpleMessaging.getInstance().shutdown(); m_initialized = false; + if (hazelcastInstance != null ){ + try { + hazelcastInstance.shutdown(); + } catch (HazelcastInstanceNotActiveException e){ + LOG.info("hazelcast allready shutdown"); + } + + } 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 dd5fa28f..848fa4fe 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -107,12 +107,20 @@ public class ProtocolProcessor { private ISessionsStore m_sessionsStore; private IAuthenticator m_authenticator; private BrokerInterceptor m_interceptor; + private String m_server_port; //maps clientID to Will testament, if specified on CONNECT private ConcurrentMap m_willStore = new ConcurrentHashMap<>(); ProtocolProcessor() {} + + public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, + ISessionsStore sessionsStore, + IAuthenticator authenticator, + boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) { + init(subscriptions,storageService,sessionsStore,authenticator,allowAnonymous,authorizator,interceptor,null); + } /** * @param subscriptions the subscription store where are stored all the existing * clients subscriptions. @@ -123,11 +131,12 @@ public class ProtocolProcessor { * @param allowAnonymous true connection to clients without credentials. * @param authorizator used to apply ACL policies to publishes and subscriptions. * @param interceptor to notify events to an intercept handler + * @param serverPort */ void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, - boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) { + boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { this.m_clientIDs = new ConcurrentHashMap<>(); this.m_interceptor = interceptor; this.subscriptions = subscriptions; @@ -137,6 +146,7 @@ public class ProtocolProcessor { m_authenticator = authenticator; m_messagesStore = storageService; m_sessionsStore = sessionsStore; + m_server_port = serverPort; } public void processConnect(Channel channel, ConnectMessage msg) { @@ -341,7 +351,7 @@ public class ProtocolProcessor { } final AbstractMessage.QOSType qos = msg.getQos(); final Integer messageID = msg.getMessageID(); - LOG.info("PUBLISH from clientID <{}> on topic <{}> with QoS {}", clientID, topic, qos); + LOG.info("PUBLISH on server {} from clientID <{}> on topic <{}> with QoS {}", m_server_port, clientID, topic, qos); String guid = null; IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); @@ -350,11 +360,16 @@ public class ProtocolProcessor { route2Subscribers(toStoreMsg); } else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1 route2Subscribers(toStoreMsg); - sendPubAck(clientID, messageID); - LOG.debug("replying with PubAck to MSG ID {}", messageID); + LOG.info("##### server {} trying to reply with PubAck to MSG ID {}", m_server_port, messageID); + if (msg.isLocal()) { + sendPubAck(clientID, messageID); + } + LOG.info("### server {} replying with PubAck to MSG ID {}", m_server_port, messageID); } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 guid = m_messagesStore.storePublishForFuture(toStoreMsg); - sendPubRec(clientID, messageID); + if(msg.isLocal()) { + sendPubRec(clientID, messageID); + } //Next the client will send us a pub rel //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher } @@ -393,7 +408,11 @@ public class ProtocolProcessor { String guid = null; IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); - toStoreMsg.setClientID("BROKER_SELF"); + if (msg.getClientId() == null || msg.getClientId().isEmpty()) { + toStoreMsg.setClientID("BROKER_SELF"); + } else { + toStoreMsg.setClientID(msg.getClientId()); + } toStoreMsg.setMessageID(1); if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 guid = m_messagesStore.storePublishForFuture(toStoreMsg); diff --git a/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java index 08bd76ce..b0c35701 100644 --- a/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java @@ -16,6 +16,7 @@ package io.moquette.spi.impl; import io.moquette.BrokerConstants; +import io.moquette.server.Server; import io.moquette.spi.IMessagesStore; import io.moquette.interception.InterceptHandler; import io.moquette.server.config.IConfig; @@ -29,6 +30,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; +import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.text.ParseException; @@ -79,7 +81,7 @@ public class SimpleMessaging { * and fallback on the default one (permit all). * */ public ProtocolProcessor init(IConfig props, List embeddedObservers, - IAuthenticator authenticator, IAuthorizator authorizator) { + IAuthenticator authenticator, IAuthorizator authorizator, Server server) { subscriptions = new SubscriptionsStore(); m_mapStorage = new MapDBPersistentStore(props); @@ -88,10 +90,16 @@ public class SimpleMessaging { ISessionsStore sessionsStore = m_mapStorage.sessionsStore(messagesStore); List observers = new ArrayList<>(embeddedObservers); - String interceptorClassName = props.getProperty("intercept.handler"); + String interceptorClassName = props.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME); if (interceptorClassName != null && !interceptorClassName.isEmpty()) { try { - InterceptHandler handler = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).newInstance(); + InterceptHandler handler; + try { + final Constructor constructor = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).getConstructor(Server.class); + handler = constructor.newInstance(server); + }catch (NoSuchMethodException nsme){ + handler = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).newInstance(); + } observers.add(handler); } catch (Throwable ex) { LOG.error("Can't load the intercept handler {}", ex); @@ -143,7 +151,7 @@ public class SimpleMessaging { } boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true")); - m_processor.init(subscriptions, messagesStore, sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor); + m_processor.init(subscriptions, messagesStore, sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME)); return m_processor; } diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java new file mode 100644 index 00000000..6adba101 --- /dev/null +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java @@ -0,0 +1,146 @@ +package io.moquette.server; + +import io.moquette.BrokerConstants; +import io.moquette.interception.HazelcastInterceptHandler; +import io.moquette.server.config.IConfig; +import io.moquette.server.config.MemoryConfig; +import org.eclipse.paho.client.mqttv3.IMqttClient; +import org.eclipse.paho.client.mqttv3.MqttClient; +import org.eclipse.paho.client.mqttv3.MqttClientPersistence; +import org.eclipse.paho.client.mqttv3.MqttMessage; +import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; +import org.junit.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.security.NoSuchAlgorithmException; +import java.sql.SQLException; +import java.util.Properties; + +import static org.junit.Assert.assertEquals; + +/** + * Created by mackristof on 13/05/2016. + */ +public class ServerIntegrationHazelcastHandlerInterceptorTest{ + private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationHazelcastHandlerInterceptorTest.class); + + static MqttClientPersistence s_dataStore; + static MqttClientPersistence s_pubDataStore; + + Server server1883; + Server server1884; + IMqttClient m_listener; + IMqttClient m_publisher; + MessageCollector m_messagesCollector; + IConfig m_config_1883; + IConfig m_config_1884; + + + + @BeforeClass + public static void beforeTests() throws NoSuchAlgorithmException, SQLException, ClassNotFoundException { + String tmpDir = System.getProperty("java.io.tmpdir"); + s_dataStore = new MqttDefaultFilePersistence(tmpDir); + s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher"); + + } + + protected Server startServer(int port, String hazelcastConfigurationFile, IConfig m_config) throws IOException { + Server m_server = new Server(); + final Properties configProps = addHazelCastConf(IntegrationUtils.prepareTestProperties(),port ,hazelcastConfigurationFile); + m_config = new MemoryConfig(configProps); + m_config.setProperty(BrokerConstants.PORT_PROPERTY_NAME,""+port); + m_server.startServer(m_config); + return m_server; + + } + + + private void stopServer(Server m_server, IConfig m_config) { + m_server.stopServer(); + } + + private Properties addHazelCastConf(Properties properties,int port, String hazelcastConfigurationFile) { + properties.put(BrokerConstants.PORT_PROPERTY_NAME, port); + properties.put(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, HazelcastInterceptHandler.class.getCanonicalName()); + properties.put(BrokerConstants.HAZELCAST_CONFIGURATION, hazelcastConfigurationFile); + return properties; + } + + @Before + public void setUp() throws Exception { + String dbPath = IntegrationUtils.localMapDBPath(); + IntegrationUtils.cleanPersistenceFile(dbPath); + + server1883 = startServer(1883,"config/hazelcast.xml", m_config_1883); + server1884 = startServer(1884,"config/hazelcast.xml", m_config_1884); + + m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); + m_publisher.connect(); + m_listener = new MqttClient("tcp://localhost:1884", "Listener", s_dataStore); + m_messagesCollector = new MessageCollector(); + m_listener.setCallback(m_messagesCollector); + + m_listener.connect(); + + + } + + + + @After + public void tearDown() throws Exception { + if (m_listener != null && m_listener.isConnected()) { + m_listener.disconnect(); + } + + if (m_publisher != null && m_publisher.isConnected()) { + m_publisher.disconnect(); + } + + stopServer(server1883, m_config_1883); + stopServer(server1884, m_config_1884); + } + + @AfterClass + public static void shutdown(){ + + } + + @Test + public void checkPublishOnHazelCastQueue() throws Exception { + + LOG.info("*** checkPublishOnHazelCastQueue ***"); + m_listener.subscribe("/topic", 0); + + m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(), 0, false); + Thread.sleep(1000); + MqttMessage messageQos0 = m_messagesCollector.getMessage(true); + assertEquals("Hello world MQTT QoS0", messageQos0.toString()); + assertEquals(0, messageQos0.getQos()); + + m_listener.subscribe("/topic", 1); + + m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(), 1, false); + MqttMessage messageQos1 = m_messagesCollector.getMessage(true); + assertEquals("Hello world MQTT QoS1", messageQos1.toString()); + assertEquals(1, messageQos1.getQos()); + + m_listener.subscribe("/topic", 2); + + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); + MqttMessage messageQos2 = m_messagesCollector.getMessage(true); + assertEquals("Hello world MQTT QoS2", messageQos2.toString()); + assertEquals(2, messageQos2.getQos()); + + Thread.sleep(5000); + + + } + + + +} diff --git a/broker/src/test/resources/config/hazelcast.xml b/broker/src/test/resources/config/hazelcast.xml new file mode 100644 index 00000000..e9e500f6 --- /dev/null +++ b/broker/src/test/resources/config/hazelcast.xml @@ -0,0 +1,10 @@ + + + + + + \ No newline at end of file diff --git a/broker/src/test/resources/hazelcast.xml b/broker/src/test/resources/hazelcast.xml new file mode 100644 index 00000000..e9e500f6 --- /dev/null +++ b/broker/src/test/resources/hazelcast.xml @@ -0,0 +1,10 @@ + + + + + + \ No newline at end of file diff --git a/broker/src/test/resources/log4j.properties b/broker/src/test/resources/log4j.properties index 2ce138c3..4957aeb8 100644 --- a/broker/src/test/resources/log4j.properties +++ b/broker/src/test/resources/log4j.properties @@ -19,6 +19,10 @@ log4j.logger.io.moquette.server.netty.NettyMQTTHandler=WARN log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=WARN log4j.logger.io.moquette.spi.persistence.MapDBSessionsStore=WARN +#HazelCast Handler +log4j.logger.io.moquette.interception.HazelcastInterceptHandler=INFO +log4j.logger.io.moquette.server.HazelcastListener=INFO + # stdout appender is set to be a ConsoleAppender. log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.Threshold=INFO diff --git a/distribution/src/main/assembly/assembly.xml b/distribution/src/main/assembly/assembly.xml index 5a0025b1..9a064d28 100644 --- a/distribution/src/main/assembly/assembly.xml +++ b/distribution/src/main/assembly/assembly.xml @@ -36,6 +36,11 @@ src/main/resources/password_file.conf config/ + + + src/main/resources/hazelcast.xml + config/ + src/main/resources/README.txt diff --git a/distribution/src/main/resources/hazelcast.xml b/distribution/src/main/resources/hazelcast.xml new file mode 100644 index 00000000..e9e500f6 --- /dev/null +++ b/distribution/src/main/resources/hazelcast.xml @@ -0,0 +1,10 @@ + + + + + + \ No newline at end of file diff --git a/netty_parser/src/main/java/io/moquette/parser/proto/messages/PublishMessage.java b/netty_parser/src/main/java/io/moquette/parser/proto/messages/PublishMessage.java index 2ae72431..0fe2ea6f 100644 --- a/netty_parser/src/main/java/io/moquette/parser/proto/messages/PublishMessage.java +++ b/netty_parser/src/main/java/io/moquette/parser/proto/messages/PublishMessage.java @@ -23,6 +23,9 @@ import java.nio.ByteBuffer; */ public class PublishMessage extends MessageIDMessage { + private boolean local = true; + private String clientId; + protected String m_topicName; protected ByteBuffer m_payload; @@ -46,4 +49,19 @@ public class PublishMessage extends MessageIDMessage { this.m_payload = payload; } + public boolean isLocal() { + return local; + } + + public void setLocal(boolean local) { + this.local = local; + } + + public String getClientId() { + return clientId; + } + + public void setClientId(String clientId) { + this.clientId = clientId; + } } From 2dd18c2b70d67cc0876a00eb8c4714eb5ff50a7f Mon Sep 17 00:00:00 2001 From: mackristof Date: Tue, 28 Jun 2016 22:07:54 +0200 Subject: [PATCH 2/4] fix(cluster): change HazelcastMsg payload to byte[] --- .../main/java/io/moquette/interception/HazelcastMsg.java | 6 +++--- .../src/main/java/io/moquette/server/HazelcastListener.java | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java index 61f4ca20..6ba7deeb 100644 --- a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java +++ b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java @@ -10,13 +10,13 @@ public class HazelcastMsg implements Serializable{ private final String clientId; private final String userName; private final byte qos; - private final String payload; + private final byte[] payload; private final String topic; public HazelcastMsg(String clientId, String topic , byte qos, byte[] payload, String userName) { this.clientId = clientId; this.qos = qos; - this.payload = new String(payload); + this.payload = payload; this.userName = userName; this.topic = topic; } @@ -30,7 +30,7 @@ public class HazelcastMsg implements Serializable{ return qos; } - public String getPayload() { + public byte[] getPayload() { return payload; } diff --git a/broker/src/main/java/io/moquette/server/HazelcastListener.java b/broker/src/main/java/io/moquette/server/HazelcastListener.java index 64acad59..7a3f13cf 100644 --- a/broker/src/main/java/io/moquette/server/HazelcastListener.java +++ b/broker/src/main/java/io/moquette/server/HazelcastListener.java @@ -32,7 +32,7 @@ public class HazelcastListener implements MessageListener { PublishMessage publishMessage = new PublishMessage(); publishMessage.setTopicName(msg.getMessageObject().getTopic()); publishMessage.setQos(AbstractMessage.QOSType.valueOf(msg.getMessageObject().getQos())); - publishMessage.setPayload(ByteBuffer.wrap(msg.getMessageObject().getPayload().getBytes("UTF-8"))); + publishMessage.setPayload(ByteBuffer.wrap(msg.getMessageObject().getPayload())); publishMessage.setLocal(false); publishMessage.setClientId(msg.getMessageObject().getClientId()); server.internalPublish(publishMessage); From 1ef2a4fd3ccea727b5a21037e4f572e689942049 Mon Sep 17 00:00:00 2001 From: andsel Date: Sat, 2 Jul 2016 12:30:19 +0200 Subject: [PATCH 3/4] Minor change, remove a redundant ClientID param from verifyToActivate method --- .../moquette/spi/impl/ProtocolProcessor.java | 22 +++++++++++-------- 1 file changed, 13 insertions(+), 9 deletions(-) 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 92f935b5..bc1098c3 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -301,7 +301,7 @@ public class ProtocolProcessor { //Remove the message from message store ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(clientID, targetSession); + verifyToActivate(targetSession); StoredMessage inflightMsg = targetSession.getInflightMessage(messageID); targetSession.inFlightAcknowledged(messageID); @@ -310,7 +310,11 @@ public class ProtocolProcessor { m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username)); } - private void verifyToActivate(String clientID, ClientSession targetSession) { + private void verifyToActivate(ClientSession targetSession) { + if (targetSession == null) { + return; + } + String clientID = targetSession.clientID; if (m_clientIDs.containsKey(clientID)) { targetSession.activate(); } @@ -459,7 +463,7 @@ public class ProtocolProcessor { qos = sub.getRequestedQos(); } ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId()); - verifyToActivate(sub.getClientId(), targetSession); + verifyToActivate(targetSession); LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", sub.getClientId(), sub.getTopicFilter(), qos, targetSession.isActive()); @@ -566,7 +570,7 @@ public class ProtocolProcessor { int messageID = msg.getMessageID(); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(clientID, targetSession); + verifyToActivate(targetSession); IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID); route2Subscribers(evt); @@ -593,7 +597,7 @@ public class ProtocolProcessor { public void processPubRec(Channel channel, PubRecMessage msg) { String clientID = NettyUtils.clientID(channel); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(clientID, targetSession); + verifyToActivate(targetSession); //remove from the inflight and move to the QoS2 second phase queue int messageID = msg.getMessageID(); targetSession.moveInFlightToSecondPhaseAckWaiting(messageID); @@ -612,7 +616,7 @@ public class ProtocolProcessor { 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 ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(clientID, targetSession); + verifyToActivate(targetSession); StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID); String username = NettyUtils.userName(channel); String topic = inflightMsg.getTopic(); @@ -668,7 +672,7 @@ public class ProtocolProcessor { LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(clientID, clientSession); + verifyToActivate(clientSession); for (String topic : topics) { boolean validTopic = SubscriptionsStore.validate(topic); if (!validTopic) { @@ -697,7 +701,7 @@ public class ProtocolProcessor { LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID()); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); - verifyToActivate(clientID, clientSession); + verifyToActivate(clientSession); //ack the client SubAckMessage ackMessage = new SubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); @@ -756,7 +760,7 @@ public class ProtocolProcessor { }); ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId()); - verifyToActivate(newSubscription.getClientId(), targetSession); + verifyToActivate(targetSession); for (IMessagesStore.StoredMessage storedMsg : messages) { //fire as retained the message LOG.trace("send publish message for topic {}", newSubscription.getTopicFilter()); From fedd2965ccf100e8fd8acc3684899593db3fafb8 Mon Sep 17 00:00:00 2001 From: andsel Date: Fri, 22 Jul 2016 12:09:10 +0200 Subject: [PATCH 4/4] Fixed Hazelcast itnegration tests, removing unusefull thread sleeps, but more important permitting the co existence of 2 Brokers in the same JVM --- .../HazelcastInterceptHandler.java | 8 +-- .../moquette/interception/HazelcastMsg.java | 5 +- .../io/moquette/server/HazelcastListener.java | 10 ++-- .../main/java/io/moquette/server/Server.java | 22 ++++---- .../moquette/spi/impl/ProtocolProcessor.java | 6 +- .../io/moquette/spi/impl/SimpleMessaging.java | 13 +++-- .../io/moquette/server/IntegrationUtils.java | 12 ++++ ...rationHazelcastHandlerInterceptorTest.java | 55 +++++++------------ 8 files changed, 60 insertions(+), 71 deletions(-) diff --git a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java index 75c58c3f..7792a424 100644 --- a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java @@ -15,18 +15,16 @@ public class HazelcastInterceptHandler extends AbstractInterceptHandler { private static final Logger LOG = LoggerFactory.getLogger(HazelcastInterceptHandler.class); private final HazelcastInstance hz; - public HazelcastInterceptHandler (Server server){ this.hz = server.getHazelcastInstance(); } - - @Override public void onPublish(InterceptPublishMessage msg) { - LOG.info(String.format("%s publish on %s message : %s\n", msg.getClientID(), msg.getTopicName(), new String(msg.getPayload().array()))); + LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), new String(msg.getPayload().array())); ITopic topic = hz.getTopic("moquette"); - HazelcastMsg hazelcastMsg = new HazelcastMsg(msg.getClientID(),msg.getTopicName(), msg.getQos().byteValue(), msg.getPayload().array(), msg.getUsername()); + HazelcastMsg hazelcastMsg = new HazelcastMsg(msg.getClientID(), msg.getTopicName(), msg.getQos().byteValue(), + msg.getPayload().array(), msg.getUsername()); topic.publish(hazelcastMsg); } diff --git a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java index 6ba7deeb..4d400764 100644 --- a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java +++ b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java @@ -5,7 +5,7 @@ import java.io.Serializable; /** * Created by mackristof on 28/05/2016. */ -public class HazelcastMsg implements Serializable{ +public class HazelcastMsg implements Serializable { private final String clientId; private final String userName; @@ -13,7 +13,7 @@ public class HazelcastMsg implements Serializable{ private final byte[] payload; private final String topic; - public HazelcastMsg(String clientId, String topic , byte qos, byte[] payload, String userName) { + public HazelcastMsg(String clientId, String topic, byte qos, byte[] payload, String userName) { this.clientId = clientId; this.qos = qos; this.payload = payload; @@ -21,7 +21,6 @@ public class HazelcastMsg implements Serializable{ this.topic = topic; } - public String getClientId() { return clientId; } diff --git a/broker/src/main/java/io/moquette/server/HazelcastListener.java b/broker/src/main/java/io/moquette/server/HazelcastListener.java index 7a3f13cf..13738441 100644 --- a/broker/src/main/java/io/moquette/server/HazelcastListener.java +++ b/broker/src/main/java/io/moquette/server/HazelcastListener.java @@ -22,13 +22,12 @@ public class HazelcastListener implements MessageListener { this.server = server; } - - @Override public void onMessage(Message msg) { try { if (!msg.getPublishingMember().equals(server.getHazelcastInstance().getCluster().getLocalMember())) { - LOG.info(String.format("%s received from hazelcast for topic %s message : %s\n", msg.getMessageObject().getClientId(), msg.getMessageObject().getTopic(), msg.getMessageObject().getPayload())); + LOG.info("{} received from hazelcast for topic {} message: {}", msg.getMessageObject().getClientId(), + msg.getMessageObject().getTopic(), msg.getMessageObject().getPayload()); PublishMessage publishMessage = new PublishMessage(); publishMessage.setTopicName(msg.getMessageObject().getTopic()); publishMessage.setQos(AbstractMessage.QOSType.valueOf(msg.getMessageObject().getQos())); @@ -36,10 +35,9 @@ public class HazelcastListener implements MessageListener { publishMessage.setLocal(false); publishMessage.setClientId(msg.getMessageObject().getClientId()); server.internalPublish(publishMessage); - } - } catch (Exception e) { - LOG.error("error while wait for polling hazelcast msg queue",e); + } catch (Exception ex) { + LOG.error("error polling hazelcast msg queue", ex); } } } diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index f57a9fd9..73c409c8 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -31,10 +31,8 @@ import io.moquette.server.config.MemoryConfig; import io.moquette.spi.impl.SimpleMessaging; import io.moquette.server.config.FilesystemConfig; import io.moquette.server.config.IConfig; -import io.moquette.server.config.MemoryConfig; import io.moquette.server.netty.NettyAcceptor; import io.moquette.spi.impl.ProtocolProcessor; -import io.moquette.spi.impl.SimpleMessaging; import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthorizator; import io.moquette.spi.security.ISslContextCreator; @@ -63,6 +61,8 @@ public class Server { private HazelcastInstance hazelcastInstance; + private SimpleMessaging m_messaging; + public static void main(String[] args) throws IOException { final Server server = new Server(); @@ -131,9 +131,9 @@ public class Server { handlers = Collections.emptyList(); } - final String handlerProp = System.getProperty("intercept.handler"); + final String handlerProp = System.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME); if (handlerProp != null) { - config.setProperty("intercept.handler", handlerProp); + config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp); } LOG.info("Persistent store file: " + config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME)); if (config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME) !=null && config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME).equals(HazelcastInterceptHandler.class.getCanonicalName())) { @@ -152,7 +152,8 @@ public class Server { } listenOnHazelCastMsg(); } - final ProtocolProcessor processor = SimpleMessaging.getInstance().init(config, handlers, authenticator, authorizator, this); + m_messaging = SimpleMessaging.getInstance(); + final ProtocolProcessor processor = m_messaging.init(config, handlers, authenticator, authorizator, this); if (sslCtxCreator == null) { sslCtxCreator = new DefaultMoquetteSslContextCreator(config); @@ -162,8 +163,6 @@ public class Server { m_acceptor.initialize(processor, config, sslCtxCreator); m_processor = processor; m_initialized = true; - - } private void listenOnHazelCastMsg() { @@ -193,15 +192,14 @@ public class Server { public void stopServer() { LOG.info("Server stopping..."); m_acceptor.close(); - SimpleMessaging.getInstance().shutdown(); + m_messaging.shutdown(); m_initialized = false; - if (hazelcastInstance != null ){ + if (hazelcastInstance != null) { try { hazelcastInstance.shutdown(); - } catch (HazelcastInstanceNotActiveException e){ - LOG.info("hazelcast allready shutdown"); + } catch (HazelcastInstanceNotActiveException e) { + LOG.info("hazelcast already shutdown"); } - } 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 ef0ae7d0..f7804e6a 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -364,11 +364,10 @@ public class ProtocolProcessor { route2Subscribers(toStoreMsg); } else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1 route2Subscribers(toStoreMsg); - LOG.info("##### server {} trying to reply with PubAck to MSG ID {}", m_server_port, messageID); if (msg.isLocal()) { sendPubAck(clientID, messageID); } - LOG.info("### server {} replying with PubAck to MSG ID {}", m_server_port, messageID); + LOG.info("server {} replying with PubAck to MSG ID {}", m_server_port, messageID); } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 guid = m_messagesStore.storePublishForFuture(toStoreMsg); if(msg.isLocal()) { @@ -717,7 +716,7 @@ public class ProtocolProcessor { public void processSubscribe(Channel channel, SubscribeMessage msg) { String clientID = NettyUtils.clientID(channel); - LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID()); + LOG.debug("SUBSCRIBE client <{}> on server {} packetID {}", clientID, m_server_port, msg.getMessageID()); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); verifyToActivate(clientSession); @@ -778,6 +777,7 @@ public class ProtocolProcessor { } }); + LOG.debug("Found {} messages to republish", messages.size()); ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId()); verifyToActivate(targetSession); for (IMessagesStore.StoredMessage storedMsg : messages) { diff --git a/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java index b0c35701..380dc4e4 100644 --- a/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java @@ -55,7 +55,7 @@ public class SimpleMessaging { private BrokerInterceptor m_interceptor; - private static SimpleMessaging INSTANCE; +// private static SimpleMessaging INSTANCE; private final ProtocolProcessor m_processor = new ProtocolProcessor(); @@ -63,10 +63,11 @@ public class SimpleMessaging { } public static SimpleMessaging getInstance() { - if (INSTANCE == null) { - INSTANCE = new SimpleMessaging(); - } - return INSTANCE; +// if (INSTANCE == null) { +// INSTANCE = new SimpleMessaging(); +// } +// return INSTANCE; + return new SimpleMessaging(); } /** @@ -97,7 +98,7 @@ public class SimpleMessaging { try { final Constructor constructor = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).getConstructor(Server.class); handler = constructor.newInstance(server); - }catch (NoSuchMethodException nsme){ + } catch (NoSuchMethodException nsme){ handler = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).newInstance(); } observers.add(handler); diff --git a/broker/src/test/java/io/moquette/server/IntegrationUtils.java b/broker/src/test/java/io/moquette/server/IntegrationUtils.java index a065aa69..9559487e 100644 --- a/broker/src/test/java/io/moquette/server/IntegrationUtils.java +++ b/broker/src/test/java/io/moquette/server/IntegrationUtils.java @@ -36,6 +36,11 @@ public class IntegrationUtils { return currentDir + File.separator + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; } + static String localClusterMapDBPath(int port) { + String currentDir = System.getProperty("user.dir"); + return currentDir + File.separator + port + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; + } + public static Properties prepareTestProperties() { Properties testProperties = new Properties(); testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath()); @@ -43,6 +48,13 @@ public class IntegrationUtils { return testProperties; } + public static Properties prepareTestClusterProperties(int port) { + Properties testProperties = new Properties(); + testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localClusterMapDBPath(port)); + testProperties.put(PORT_PROPERTY_NAME, Integer.toString(port)); + return testProperties; + } + public static void cleanPersistenceFile(IConfig config) { String fileName = config.getProperty(PERSISTENT_STORE_PROPERTY_NAME); cleanPersistenceFile(fileName); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java index 6adba101..ca6e7174 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java @@ -38,32 +38,25 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest{ IConfig m_config_1883; IConfig m_config_1884; - - @BeforeClass public static void beforeTests() throws NoSuchAlgorithmException, SQLException, ClassNotFoundException { String tmpDir = System.getProperty("java.io.tmpdir"); s_dataStore = new MqttDefaultFilePersistence(tmpDir); s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher"); - } - protected Server startServer(int port, String hazelcastConfigurationFile, IConfig m_config) throws IOException { + protected Server startServer(int port, IConfig m_config) throws IOException { Server m_server = new Server(); - final Properties configProps = addHazelCastConf(IntegrationUtils.prepareTestProperties(),port ,hazelcastConfigurationFile); - m_config = new MemoryConfig(configProps); - m_config.setProperty(BrokerConstants.PORT_PROPERTY_NAME,""+port); + m_config.setProperty(BrokerConstants.PORT_PROPERTY_NAME, Integer.toString(port)); m_server.startServer(m_config); return m_server; - } - - private void stopServer(Server m_server, IConfig m_config) { + private void stopServer(Server m_server) { m_server.stopServer(); } - private Properties addHazelCastConf(Properties properties,int port, String hazelcastConfigurationFile) { + private Properties addHazelCastConf(Properties properties, int port, String hazelcastConfigurationFile) { properties.put(BrokerConstants.PORT_PROPERTY_NAME, port); properties.put(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, HazelcastInterceptHandler.class.getCanonicalName()); properties.put(BrokerConstants.HAZELCAST_CONFIGURATION, hazelcastConfigurationFile); @@ -75,8 +68,16 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest{ String dbPath = IntegrationUtils.localMapDBPath(); IntegrationUtils.cleanPersistenceFile(dbPath); - server1883 = startServer(1883,"config/hazelcast.xml", m_config_1883); - server1884 = startServer(1884,"config/hazelcast.xml", m_config_1884); + final Properties configProps = addHazelCastConf(IntegrationUtils.prepareTestClusterProperties(1883), + 1883, "config/hazelcast.xml"); + m_config_1883 = new MemoryConfig(configProps); + + server1883 = startServer(1883, m_config_1883); + + final Properties configProps1884 = addHazelCastConf(IntegrationUtils.prepareTestClusterProperties(1884), + 1883, "config/hazelcast.xml"); + m_config_1884 = new MemoryConfig(configProps1884); + server1884 = startServer(1884, m_config_1884); m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); m_publisher.connect(); @@ -85,12 +86,8 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest{ m_listener.setCallback(m_messagesCollector); m_listener.connect(); - - } - - @After public void tearDown() throws Exception { if (m_listener != null && m_listener.isConnected()) { @@ -101,23 +98,16 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest{ m_publisher.disconnect(); } - stopServer(server1883, m_config_1883); - stopServer(server1884, m_config_1884); - } - - @AfterClass - public static void shutdown(){ - + stopServer(server1883); + stopServer(server1884); } @Test - public void checkPublishOnHazelCastQueue() throws Exception { - - LOG.info("*** checkPublishOnHazelCastQueue ***"); - m_listener.subscribe("/topic", 0); + public void checkPublishPassThroughCluster_Qos0() throws Exception { + LOG.info("*** checkPublishPassThroughCluster_Qos0 ***"); + m_listener.subscribe("/topic", 1); m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(), 0, false); - Thread.sleep(1000); MqttMessage messageQos0 = m_messagesCollector.getMessage(true); assertEquals("Hello world MQTT QoS0", messageQos0.toString()); assertEquals(0, messageQos0.getQos()); @@ -135,12 +125,5 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest{ MqttMessage messageQos2 = m_messagesCollector.getMessage(true); assertEquals("Hello world MQTT QoS2", messageQos2.toString()); assertEquals(2, messageQos2.getQos()); - - Thread.sleep(5000); - - } - - - }