anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dropper Hazelcast because Moquette is not designed to be clusterizable

Browse Source
This commit is contained in:
andsel
2018-07-20 18:17:30 +02:00
parent baecbbcb31
commit 7546a509da
8 changed files with 4 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChangeLog.txt
+4
View File
@@ -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
broker/build.gradle
-1
View File
@@ -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'
broker/src/main/java/io/moquette/BrokerConstants.java
-1
View File
@@ -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";
broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java
-56
View File
@@ -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<HazelcastMsg> topic = hz.getTopic("moquette");
HazelcastMsg hazelcastMsg = new HazelcastMsg(msg);
topic.publish(hazelcastMsg);
}
}
broker/src/main/java/io/moquette/interception/HazelcastMsg.java
-53
View File
@@ -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;
}
}
broker/src/main/java/io/moquette/server/HazelcastListener.java
-62
View File
@@ -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<HazelcastMsg> {
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<HazelcastMsg> 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);
}
}
}
broker/src/main/java/io/moquette/server/Server.java
-54
View File
@@ -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<HazelcastMsg> 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().
broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java
-153
View File
@@ -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());
}
}