From 7546a509da9e5888be33650e5f8f1a52cc6bd4be Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 19 Jul 2018 17:57:52 +0200 Subject: [PATCH] Dropper Hazelcast because Moquette is not designed to be clusterizable --- ChangeLog.txt | 4 + broker/build.gradle | 1 - .../java/io/moquette/BrokerConstants.java | 1 - .../HazelcastInterceptHandler.java | 56 ------- .../moquette/interception/HazelcastMsg.java | 53 ------ .../io/moquette/server/HazelcastListener.java | 62 ------- .../main/java/io/moquette/server/Server.java | 54 ------- ...rationHazelcastHandlerInterceptorTest.java | 153 ------------------ 8 files changed, 4 insertions(+), 380 deletions(-) delete mode 100644 broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java delete mode 100644 broker/src/main/java/io/moquette/interception/HazelcastMsg.java delete mode 100644 broker/src/main/java/io/moquette/server/HazelcastListener.java delete mode 100644 broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java diff --git a/ChangeLog.txt b/ChangeLog.txt index a4b5f9b5..3a84a44f 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -1,3 +1,7 @@ +Version 0.12: + Fixed issue #415 + Dropped Hazelcast because at the moment Moquette is not designed to be clusterizable + Version 0.10: Netty epoll system added Separated in memory and MapDB storage implementations diff --git a/broker/build.gradle b/broker/build.gradle index 439ef515..af90eb96 100644 --- a/broker/build.gradle +++ b/broker/build.gradle @@ -24,7 +24,6 @@ dependencies { compile group: 'commons-codec', name: 'commons-codec', version:'1.10' // compile group: 'org.hdrhistogram', name: 'HdrHistogram', version:'2.1.4' - compile group: 'com.hazelcast', name: 'hazelcast', version:'3.5.4' compile group: 'org.slf4j', name: 'slf4j-api', version:'1.7.5' testCompile group: 'org.slf4j', name: 'slf4j-log4j12', version:'1.7.5' diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 5eee7edc..ffa93727 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -60,7 +60,6 @@ public final 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"; public static final String NETTY_SO_BACKLOG_PROPERTY_NAME = "netty.so_backlog"; public static final String NETTY_SO_REUSEADDR_PROPERTY_NAME = "netty.so_reuseaddr"; public static final String NETTY_TCP_NODELAY_PROPERTY_NAME = "netty.tcp_nodelay"; diff --git a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java deleted file mode 100644 index 3d2908ee..00000000 --- a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java +++ /dev/null @@ -1,56 +0,0 @@ -/* - * Copyright (c) 2012-2018 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ - -package io.moquette.interception; - -import com.hazelcast.core.HazelcastInstance; -import com.hazelcast.core.ITopic; -import io.moquette.interception.messages.InterceptPublishMessage; -import io.moquette.server.Server; -import io.netty.buffer.ByteBuf; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import static io.moquette.spi.impl.Utils.readBytesAndRewind; -import static java.nio.charset.StandardCharsets.UTF_8; - -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 String getID() { - return HazelcastInterceptHandler.class.getName() + "@" + hz.getName(); - } - - @Override - public void onPublish(InterceptPublishMessage msg) { - // TODO ugly, too much array copy - ByteBuf payload = msg.getPayload(); - byte[] payloadContent = readBytesAndRewind(payload); - - LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), - new String(payloadContent, UTF_8)); - ITopic topic = hz.getTopic("moquette"); - HazelcastMsg hazelcastMsg = new HazelcastMsg(msg); - 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 deleted file mode 100644 index 3af14642..00000000 --- a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java +++ /dev/null @@ -1,53 +0,0 @@ -/* - * Copyright (c) 2012-2018 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ - -package io.moquette.interception; - -import io.moquette.interception.messages.InterceptPublishMessage; -import java.io.Serializable; -import static io.moquette.spi.impl.Utils.readBytesAndRewind; - -public class HazelcastMsg implements Serializable { - - private static final long serialVersionUID = -1431584750134928273L; - private final String clientId; - private final int qos; - private final byte[] payload; - private final String topic; - - public HazelcastMsg(InterceptPublishMessage msg) { - this.clientId = msg.getClientID(); - this.topic = msg.getTopicName(); - this.qos = msg.getQos().value(); - this.payload = readBytesAndRewind(msg.getPayload()); - } - - public String getClientId() { - return clientId; - } - - public int getQos() { - return qos; - } - - public byte[] getPayload() { - return payload; - } - - 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 deleted file mode 100644 index 5e8cce7f..00000000 --- a/broker/src/main/java/io/moquette/server/HazelcastListener.java +++ /dev/null @@ -1,62 +0,0 @@ -/* - * Copyright (c) 2012-2018 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ - -package io.moquette.server; - -import com.hazelcast.core.Message; -import com.hazelcast.core.MessageListener; -import io.moquette.interception.HazelcastMsg; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.Unpooled; -import io.netty.handler.codec.mqtt.*; -import io.netty.util.ReferenceCountUtil; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -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) { - ByteBuf payload = null; - try { - if (!msg.getPublishingMember().equals(server.getHazelcastInstance().getCluster().getLocalMember())) { - HazelcastMsg hzMsg = msg.getMessageObject(); - LOG.info("{} received from hazelcast for topic {} message: {}", hzMsg.getClientId(), hzMsg.getTopic(), - hzMsg.getPayload()); - // TODO pass forward this information in somehow publishMessage.setLocal(false); - - MqttQoS qos = MqttQoS.valueOf(hzMsg.getQos()); - MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, false, 0); - MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(hzMsg.getTopic(), 0); - payload = Unpooled.wrappedBuffer(hzMsg.getPayload()); - MqttPublishMessage publishMessage = new MqttPublishMessage(fixedHeader, varHeader, payload); - server.internalPublish(publishMessage, hzMsg.getClientId()); - } - } catch (Exception ex) { - LOG.error("error polling hazelcast msg queue", ex); - } finally { - ReferenceCountUtil.release(payload); - } - } -} diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index 55b794cb..e053dc2f 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -16,17 +16,8 @@ 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.connections.IConnectionsManager; -import io.moquette.interception.HazelcastInterceptHandler; -import io.moquette.interception.HazelcastMsg; import io.moquette.interception.InterceptHandler; import io.moquette.server.config.*; import io.moquette.server.netty.NettyAcceptor; @@ -41,7 +32,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; -import java.io.FileNotFoundException; import java.io.IOException; import java.util.Collections; import java.util.List; @@ -58,15 +48,12 @@ public class Server { private static final Logger LOG = LoggerFactory.getLogger(Server.class); - private static final String HZ_INTERCEPT_HANDLER = HazelcastInterceptHandler.class.getCanonicalName(); - private ServerAcceptor m_acceptor; private volatile boolean m_initialized; private ProtocolProcessor m_processor; - private HazelcastInstance hazelcastInstance; private ProtocolProcessorBootstrapper m_processorBootstrapper; @@ -167,7 +154,6 @@ public class Server { if (handlerProp != null) { config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp); } - configureCluster(config); final String persistencePath = config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME); LOG.debug("Configuring Using persistent store file, path={}", persistencePath); m_processorBootstrapper = new ProtocolProcessorBootstrapper(); @@ -188,38 +174,6 @@ public class Server { m_initialized = true; } - private void configureCluster(IConfig config) throws FileNotFoundException { - LOG.info("Configuring embedded Hazelcast instance"); - String interceptHandlerClassname = config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME); - if (interceptHandlerClassname == null || !HZ_INTERCEPT_HANDLER.equals(interceptHandlerClassname)) { - LOG.info("There are no Hazelcast intercept handlers. The server won't start a Hazelcast instance."); - return; - } - String hzConfigPath = config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION); - if (hzConfigPath != null) { - boolean isHzConfigOnClasspath = this.getClass().getClassLoader().getResource(hzConfigPath) != null; - Config hzconfig = isHzConfigOnClasspath - ? new ClasspathXmlConfig(hzConfigPath) - : new FileSystemXmlConfig(hzConfigPath); - LOG.info("Starting Hazelcast instance. ConfigurationFile={}", hzconfig); - hazelcastInstance = Hazelcast.newHazelcastInstance(hzconfig); - } else { - LOG.info("Starting Hazelcast instance with default configuration"); - hazelcastInstance = Hazelcast.newHazelcastInstance(); - } - listenOnHazelCastMsg(); - } - - private void listenOnHazelCastMsg() { - LOG.info("Subscribing to Hazelcast topic. TopicName={}", "moquette"); - 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 @@ -246,14 +200,6 @@ public class Server { LOG.trace("Stopping MQTT protocol processor"); m_processorBootstrapper.shutdown(); m_initialized = false; - if (hazelcastInstance != null) { - LOG.trace("Stopping embedded Hazelcast instance"); - try { - hazelcastInstance.shutdown(); - } catch (HazelcastInstanceNotActiveException e) { - LOG.warn("embedded Hazelcast instance is already shut down."); - } - } // calling shutdown() does not actually stop tasks that are not cancelled, // and SessionsRepository does not stop its tasks. Thus shutdownNow(). diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java deleted file mode 100644 index 2e0e3ed1..00000000 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java +++ /dev/null @@ -1,153 +0,0 @@ -/* - * Copyright (c) 2012-2018 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ - -package io.moquette.server; - -import io.moquette.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 io.moquette.BrokerConstants.*; -import static java.nio.charset.StandardCharsets.UTF_8; -import static org.junit.Assert.assertEquals; - -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(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(PORT_PROPERTY_NAME, port); - properties.put(INTERCEPT_HANDLER_PROPERTY_NAME, HazelcastInterceptHandler.class.getCanonicalName()); - properties.put(HAZELCAST_CONFIGURATION, hazelcastConfigurationFile); - return properties; - } - - @Before - public void setUp() throws Exception { - 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(UTF_8), 0, false); - MqttMessage messageQos0 = m_messagesCollector.waitMessage(1); - assertEquals("Hello world MQTT QoS0", messageQos0.toString()); - assertEquals(0, messageQos0.getQos()); - } - - @Test - public void checkPublishPassThroughCluster_Qos1() throws Exception { - LOG.info("*** checkPublishPassThroughCluster_Qos1 ***"); - m_listener.subscribe("/topic", 1); - - m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(UTF_8), 1, false); - MqttMessage messageQos1 = m_messagesCollector.waitMessage(1); - assertEquals("Hello world MQTT QoS1", messageQos1.toString()); - assertEquals(1, messageQos1.getQos()); - } - - @Test - public void checkPublishPassThroughCluster_Qos2() throws Exception { - LOG.info("*** checkPublishPassThroughCluster_Qos2 ***"); - m_listener.subscribe("/topic", 2); - - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(UTF_8), 2, false); - MqttMessage messageQos2 = m_messagesCollector.waitMessage(1); - assertEquals("Hello world MQTT QoS2", messageQos2.toString()); - assertEquals(2, messageQos2.getQos()); - } -}