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..6ba7deeb --- /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 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..7a3f13cf --- /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())); + 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 bc1098c3..ef0ae7d0 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -58,7 +58,7 @@ import org.slf4j.LoggerFactory; /** * Class responsible to handle the logic of MQTT protocol it's the director of - * the protocol execution. + * the protocol execution. * * Used by the front facing class SimpleMessaging. * @@ -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) { @@ -167,7 +177,7 @@ public class ProtocolProcessor { failedCredentials(channel); return; } - if (!m_authenticator.checkValid(msg.getUsername(), pwd)) { + if (!m_authenticator.checkValid(msg.getClientID(), msg.getUsername(), pwd)) { failedCredentials(channel); channel.close(); return; @@ -345,7 +355,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); @@ -354,11 +364,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 } @@ -397,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); 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/main/java/io/moquette/spi/impl/security/AcceptAllAuthenticator.java b/broker/src/main/java/io/moquette/spi/impl/security/AcceptAllAuthenticator.java index 0e39dd6d..39898665 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/AcceptAllAuthenticator.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/AcceptAllAuthenticator.java @@ -6,7 +6,7 @@ import io.moquette.spi.security.IAuthenticator; * Created by andrea on 8/23/14. */ public class AcceptAllAuthenticator implements IAuthenticator { - public boolean checkValid(String username, byte[] password) { + public boolean checkValid(String clientId, String username, byte[] password) { return true; } } diff --git a/broker/src/main/java/io/moquette/spi/impl/security/DBAuthenticator.java b/broker/src/main/java/io/moquette/spi/impl/security/DBAuthenticator.java index 0d92faa0..e2c8e849 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/DBAuthenticator.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/DBAuthenticator.java @@ -63,7 +63,7 @@ public class DBAuthenticator implements IAuthenticator { } @Override - public synchronized boolean checkValid(String username, byte[] password) { + public synchronized boolean checkValid(String clientId, String username, byte[] password) { // Check Username / Password in DB using sqlQuery if (username == null || password == null) { LOG.info("username or password was null"); diff --git a/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java b/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java index e2423417..7825e916 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java @@ -106,8 +106,9 @@ public class FileAuthenticator implements IAuthenticator { throw new ParseException("Failed to read", 1); } } - - public boolean checkValid(String username, byte[] password) { + + @Override + public boolean checkValid(String clientId, String username, byte[] password) { if (username == null || password == null) { LOG.info("username or password was null"); return false; diff --git a/broker/src/main/java/io/moquette/spi/security/IAuthenticator.java b/broker/src/main/java/io/moquette/spi/security/IAuthenticator.java index e760a3d2..0bd4f118 100644 --- a/broker/src/main/java/io/moquette/spi/security/IAuthenticator.java +++ b/broker/src/main/java/io/moquette/spi/security/IAuthenticator.java @@ -22,5 +22,5 @@ package io.moquette.spi.security; */ public interface IAuthenticator { - boolean checkValid(String username, byte[] password); + boolean checkValid(String clientId, String username, byte[] password); } diff --git a/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java b/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java index 68ac1f18..91636841 100644 --- a/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java +++ b/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java @@ -72,7 +72,7 @@ public class ConfigurationClassLoaderTest implements IAuthenticator, IAuthorizat } @Override - public boolean checkValid(String username, byte[] password) { + public boolean checkValid(String clientID, String username, byte[] password) { return true; } 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/java/io/moquette/spi/impl/MockAuthenticator.java b/broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java index 892b835e..296a4c90 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java +++ b/broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java @@ -16,6 +16,7 @@ package io.moquette.spi.impl; import java.util.Map; +import java.util.Set; import io.moquette.spi.security.IAuthenticator; @@ -25,14 +26,19 @@ import io.moquette.spi.security.IAuthenticator; * @author andrea */ class MockAuthenticator implements IAuthenticator { - + + private Set m_clientIds; private Map m_userPwds; - MockAuthenticator(Map userPwds) { + MockAuthenticator(Set clientIds, Map userPwds) { + m_clientIds = clientIds; m_userPwds = userPwds; } - public boolean checkValid(String username, byte[] password) { + public boolean checkValid(String clientId, String username, byte[] password) { + if (!m_clientIds.contains(clientId)) { + return false; + } if (!m_userPwds.containsKey(username)) { return false; } diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 9792ebdb..983dece9 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -83,10 +83,13 @@ public class ProtocolProcessorTest { m_messagesStore = memStorage.messagesStore(); m_sessionStore = memStorage.sessionsStore(); //m_messagesStore.initStore(); - + + Set clientIds = new HashSet<>(); + clientIds.add(FAKE_CLIENT_ID); + clientIds.add(FAKE_CLIENT_ID2); Map users = new HashMap<>(); users.put(TEST_USER, TEST_PWD); - m_mockAuthenticator = new MockAuthenticator(users); + m_mockAuthenticator = new MockAuthenticator(clientIds, users); subscriptions = new SubscriptionsStore(); subscriptions.init(memStorage.sessionsStore()); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 17692691..27d5b5ee 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -30,7 +30,9 @@ import org.junit.Before; import org.junit.Test; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; +import java.util.Set; import static io.moquette.parser.netty.Utils.VERSION_3_1_1; import static io.moquette.parser.proto.messages.ConnAckMessage.BAD_USERNAME_OR_PASSWORD; @@ -71,9 +73,11 @@ public class ProtocolProcessor_CONNECT_Test { m_sessionStore = memStorage.sessionsStore(); //m_messagesStore.initStore(); + Set clientIds = new HashSet<>(); + clientIds.add(ProtocolProcessorTest.FAKE_CLIENT_ID); Map users = new HashMap<>(); users.put(ProtocolProcessorTest.TEST_USER, ProtocolProcessorTest.TEST_PWD); - m_mockAuthenticator = new MockAuthenticator(users); + m_mockAuthenticator = new MockAuthenticator(clientIds, users); subscriptions = new SubscriptionsStore(); subscriptions.init(m_sessionStore); @@ -106,7 +110,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void testWill() { - connMsg.setClientID("123"); + connMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); connMsg.setWillFlag(true); connMsg.setWillTopic("topic"); connMsg.setWillMessage("Topic message".getBytes()); @@ -125,7 +129,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void validAuthentication() { - connMsg.setClientID("123"); + connMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); connMsg.setUserFlag(true); connMsg.setPasswordFlag(true); connMsg.setUsername(ProtocolProcessorTest.TEST_USER); @@ -140,7 +144,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void noPasswdAuthentication() { - connMsg.setClientID("123"); + connMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); connMsg.setUserFlag(true); connMsg.setPasswordFlag(false); connMsg.setUsername(ProtocolProcessorTest.TEST_USER); @@ -154,7 +158,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void invalidAuthentication() { - connMsg.setClientID("123"); + connMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); connMsg.setUserFlag(true); connMsg.setPasswordFlag(true); connMsg.setUsername(ProtocolProcessorTest.TEST_USER + "_fake"); @@ -169,7 +173,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void prohibitAnonymousClient() { - connMsg.setClientID("123"); + connMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); @@ -182,7 +186,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void prohibitAnonymousClient_providingUsername() { - connMsg.setClientID("123"); + connMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); connMsg.setUserFlag(true); connMsg.setUsername(ProtocolProcessorTest.TEST_USER + "_fake"); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, @@ -197,7 +201,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void acceptAnonymousClient() { - connMsg.setClientID("123"); + connMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); @@ -214,7 +218,7 @@ public class ProtocolProcessor_CONNECT_Test { new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); //first connect with clean session true - connMsg.setClientID("123"); + connMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); connMsg.setCleanSession(true); m_processor.processConnect(m_session, connMsg); assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); @@ -225,12 +229,12 @@ public class ProtocolProcessor_CONNECT_Test { m_session = new EmbeddedChannel(); ConnectMessage secondConnMsg = new ConnectMessage(); secondConnMsg.setProtocolVersion((byte) 0x03); - secondConnMsg.setClientID("123"); + secondConnMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); m_processor.processConnect(m_session, secondConnMsg); assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); //Verify client session is clean false - ClientSession session = m_sessionStore.sessionForClient("123"); + ClientSession session = m_sessionStore.sessionForClient(ProtocolProcessorTest.FAKE_CLIENT_ID); assertFalse(session.isCleanSession()); //Verify @@ -240,7 +244,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void connectWithSameClientIDBadCredentialsDoesntDropExistingClient() { //Connect a client1 - connMsg.setClientID("Client1"); + connMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); connMsg.setUserFlag(true); connMsg.setPasswordFlag(true); connMsg.setUsername(ProtocolProcessorTest.TEST_USER); @@ -251,7 +255,7 @@ public class ProtocolProcessor_CONNECT_Test { //create another connect same clientID but with bad credentials ConnectMessage evilClientConnMsg = new ConnectMessage(); evilClientConnMsg.setProtocolVersion((byte) 0x03); - evilClientConnMsg.setClientID("Client1"); + evilClientConnMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); evilClientConnMsg.setUserFlag(true); evilClientConnMsg.setPasswordFlag(true); evilClientConnMsg.setUsername(ProtocolProcessorTest.EVIL_TEST_USER); @@ -274,9 +278,9 @@ public class ProtocolProcessor_CONNECT_Test { public void testConnAckContainsSessionPresentFlag() throws InterruptedException { connMsg = new ConnectMessage(); connMsg.setProtocolVersion(VERSION_3_1_1); - connMsg.setClientID("CliID"); + connMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); connMsg.setCleanSession(false); - NettyUtils.clientID(m_session, "CliID"); + NettyUtils.clientID(m_session, ProtocolProcessorTest.FAKE_CLIENT_ID); NettyUtils.cleanSession(m_session, false); //Connect a first time @@ -298,7 +302,7 @@ public class ProtocolProcessor_CONNECT_Test { //connect with clean a false and subscribe to a topic connMsg = new ConnectMessage(); connMsg.setProtocolVersion(VERSION_3_1_1); - connMsg.setClientID("CliID"); + connMsg.setClientID(ProtocolProcessorTest.FAKE_CLIENT_ID); connMsg.setCleanSession(false); m_processor.processConnect(m_session, connMsg); assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); @@ -307,10 +311,10 @@ public class ProtocolProcessor_CONNECT_Test { SubscribeMessage subscribeMsg = new SubscribeMessage(); subscribeMsg.addSubscription(new SubscribeMessage.Couple((byte) AbstractMessage.QOSType.MOST_ONE.ordinal(), ProtocolProcessorTest.FAKE_TOPIC)); - NettyUtils.clientID(m_session, "CliID"); + NettyUtils.clientID(m_session, ProtocolProcessorTest.FAKE_CLIENT_ID); NettyUtils.cleanSession(m_session, false); m_processor.processSubscribe(m_session, subscribeMsg); - Subscription expectedSubscription = new Subscription("CliID", ProtocolProcessorTest.FAKE_TOPIC, + Subscription expectedSubscription = new Subscription(ProtocolProcessorTest.FAKE_CLIENT_ID, ProtocolProcessorTest.FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); assertTrue(subscriptions.contains(expectedSubscription)); assertEqualsSubAck(m_session.readOutbound()); diff --git a/broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java b/broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java index 1db62da6..885825d2 100644 --- a/broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java @@ -57,19 +57,19 @@ public class DBAuthenticatorTest { @Test public void Db_verifyValid() { final DBAuthenticator dbAuthenticator = new DBAuthenticator(ORG_H2_DRIVER, JDBC_H2_MEM_TEST, "SELECT PASSWORD FROM ACCOUNT WHERE LOGIN=?", SHA_256); - assertTrue(dbAuthenticator.checkValid("dbuser","password".getBytes())); + assertTrue(dbAuthenticator.checkValid(null, "dbuser","password".getBytes())); } @Test public void Db_verifyInvalidLogin(){ final DBAuthenticator dbAuthenticator = new DBAuthenticator(ORG_H2_DRIVER, JDBC_H2_MEM_TEST, "SELECT PASSWORD FROM ACCOUNT WHERE LOGIN=?", SHA_256); - assertFalse(dbAuthenticator.checkValid("dbuser2","password".getBytes())); + assertFalse(dbAuthenticator.checkValid(null, "dbuser2","password".getBytes())); } @Test public void Db_verifyInvalidPassword(){ final DBAuthenticator dbAuthenticator = new DBAuthenticator(ORG_H2_DRIVER, JDBC_H2_MEM_TEST, "SELECT PASSWORD FROM ACCOUNT WHERE LOGIN=?", SHA_256); - assertFalse(dbAuthenticator.checkValid("dbuser","wrongPassword".getBytes())); + assertFalse(dbAuthenticator.checkValid(null, "dbuser","wrongPassword".getBytes())); } diff --git a/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java b/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java index f18e2673..0e707e8e 100644 --- a/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java @@ -32,7 +32,7 @@ public class FileAuthenticatorTest { String file = getClass().getResource("/password_file.conf").getPath(); IAuthenticator auth = new FileAuthenticator(null, file); - assertTrue(auth.checkValid("testuser", "passwd".getBytes())); + assertTrue(auth.checkValid(null, "testuser", "passwd".getBytes())); } @Test @@ -40,14 +40,14 @@ public class FileAuthenticatorTest { String file = getClass().getResource("/password_file.conf").getPath(); IAuthenticator auth = new FileAuthenticator(null, file); - assertFalse(auth.checkValid("testuser2", "passwd".getBytes())); + assertFalse(auth.checkValid(null, "testuser2", "passwd".getBytes())); } @Test public void loadPasswordFile_verifyDirectoryRef() { IAuthenticator auth = new FileAuthenticator("", ""); - assertFalse(auth.checkValid("testuser2", "passwd".getBytes())); + assertFalse(auth.checkValid(null, "testuser2", "passwd".getBytes())); } } \ No newline at end of file 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; + } }