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..7792a424 --- /dev/null +++ b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java @@ -0,0 +1,31 @@ +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("{} 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()); + 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..4d400764 --- /dev/null +++ b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java @@ -0,0 +1,43 @@ +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 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 = payload; + this.userName = userName; + this.topic = topic; + } + + public String getClientId() { + return clientId; + } + + public byte getQos() { + return qos; + } + + public byte[] 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..13738441 --- /dev/null +++ b/broker/src/main/java/io/moquette/server/HazelcastListener.java @@ -0,0 +1,43 @@ +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("{} 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())); + publishMessage.setPayload(ByteBuffer.wrap(msg.getMessageObject().getPayload())); + publishMessage.setLocal(false); + publishMessage.setClientId(msg.getMessageObject().getClientId()); + server.internalPublish(publishMessage); + } + } 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 b53bdf8b..7d672866 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -15,7 +15,16 @@ */ 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; @@ -50,6 +59,11 @@ public class Server { private ProtocolProcessor m_processor; + private HazelcastInstance hazelcastInstance; + + private SimpleMessaging m_messaging; + + public static void main(String[] args) throws IOException { final Server server = new Server(); server.startServer(); @@ -117,12 +131,29 @@ 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)); - 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(); + } + m_messaging = SimpleMessaging.getInstance(); + final ProtocolProcessor processor = m_messaging.init(config, handlers, authenticator, authorizator, this); if (sslCtxCreator == null) { sslCtxCreator = new DefaultMoquetteSslContextCreator(config); @@ -134,6 +165,16 @@ public class Server { m_initialized = true; } + private void listenOnHazelCastMsg() { + HazelcastInstance hz = getHazelcastInstance(); + ITopic topic = hz.getTopic("moquette"); + topic.addMessageListener(new HazelcastListener(this)); + } + + public HazelcastInstance getHazelcastInstance(){ + return hazelcastInstance; + } + /** * 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. @@ -151,8 +192,15 @@ 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) { + try { + hazelcastInstance.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 6981f4ea..dc85dc58 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -108,12 +108,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. @@ -124,11 +132,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; @@ -138,6 +147,7 @@ public class ProtocolProcessor { m_authenticator = authenticator; m_messagesStore = storageService; m_sessionsStore = sessionsStore; + m_server_port = serverPort; } public void processConnect(Channel channel, ConnectMessage msg) { @@ -302,7 +312,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); @@ -311,7 +321,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(); } @@ -342,7 +356,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); @@ -351,11 +365,15 @@ 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); + 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 } @@ -394,7 +412,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); @@ -460,7 +482,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()); @@ -567,7 +589,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); @@ -594,7 +616,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); @@ -613,7 +635,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(); @@ -669,7 +691,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) { @@ -695,10 +717,10 @@ 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(clientID, clientSession); + verifyToActivate(clientSession); //ack the client SubAckMessage ackMessage = new SubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); @@ -756,8 +778,9 @@ public class ProtocolProcessor { } }); + LOG.debug("Found {} messages to republish", messages.size()); 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()); 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..380dc4e4 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; @@ -53,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(); @@ -61,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(); } /** @@ -79,7 +82,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 +91,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 +152,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/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 new file mode 100644 index 00000000..ca6e7174 --- /dev/null +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java @@ -0,0 +1,129 @@ +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, IConfig m_config) throws IOException { + Server m_server = new Server(); + m_config.setProperty(BrokerConstants.PORT_PROPERTY_NAME, Integer.toString(port)); + m_server.startServer(m_config); + return m_server; + } + + private void stopServer(Server m_server) { + 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); + + 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(); + 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); + stopServer(server1884); + } + + @Test + 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); + 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()); + } +} 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; + } }