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; + } }