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

Merge pull request #189 from andsel/mackristof-feat/cluster

Browse Source 
Mackristof feat/cluster
This commit is contained in:
Andrea Selva
2016-07-22 12:13:16 +02:00
committed by GitHub
parent 6a4aa28489 fedd2965cc
commit 3085ee54d8
16 changed files with 432 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/BrokerConstants.java
+2
View File
@@ -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";
}
broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java
+31
View File
@@ -0,0 +1,31 @@
package io.moquette.interception;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ITopic;
import io.moquette.interception.messages.InterceptPublishMessage;
import io.moquette.server.Server;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Created by mackristof on 28/05/2016.
*/
public class HazelcastInterceptHandler extends AbstractInterceptHandler {
private static final Logger LOG = LoggerFactory.getLogger(HazelcastInterceptHandler.class);
private final HazelcastInstance hz;
public HazelcastInterceptHandler (Server server){
this.hz = server.getHazelcastInstance();
}
@Override
public void onPublish(InterceptPublishMessage msg) {
LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), new String(msg.getPayload().array()));
ITopic<HazelcastMsg> topic = hz.getTopic("moquette");
HazelcastMsg hazelcastMsg = new HazelcastMsg(msg.getClientID(), msg.getTopicName(), msg.getQos().byteValue(),
msg.getPayload().array(), msg.getUsername());
topic.publish(hazelcastMsg);
}
}
broker/src/main/java/io/moquette/interception/HazelcastMsg.java
+43
View File
@@ -0,0 +1,43 @@
package io.moquette.interception;
import java.io.Serializable;
/**
* Created by mackristof on 28/05/2016.
*/
public class HazelcastMsg implements Serializable {
private final String clientId;
private final String userName;
private final byte qos;
private final byte[] payload;
private final String topic;
public HazelcastMsg(String clientId, String topic, byte qos, byte[] payload, String userName) {
this.clientId = clientId;
this.qos = qos;
this.payload = payload;
this.userName = userName;
this.topic = topic;
}
public String getClientId() {
return clientId;
}
public byte getQos() {
return qos;
}
public byte[] getPayload() {
return payload;
}
public String getUserName() {
return userName;
}
public String getTopic() {
return topic;
}
}
broker/src/main/java/io/moquette/server/HazelcastListener.java
+43
View File
@@ -0,0 +1,43 @@
package io.moquette.server;
import com.hazelcast.core.Message;
import com.hazelcast.core.MessageListener;
import io.moquette.interception.HazelcastMsg;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.parser.proto.messages.PublishMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
/**
* Created by mackristof on 28/05/2016.
*/
public class HazelcastListener implements MessageListener<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) {
try {
if (!msg.getPublishingMember().equals(server.getHazelcastInstance().getCluster().getLocalMember())) {
LOG.info("{} received from hazelcast for topic {} message: {}", msg.getMessageObject().getClientId(),
msg.getMessageObject().getTopic(), msg.getMessageObject().getPayload());
PublishMessage publishMessage = new PublishMessage();
publishMessage.setTopicName(msg.getMessageObject().getTopic());
publishMessage.setQos(AbstractMessage.QOSType.valueOf(msg.getMessageObject().getQos()));
publishMessage.setPayload(ByteBuffer.wrap(msg.getMessageObject().getPayload()));
publishMessage.setLocal(false);
publishMessage.setClientId(msg.getMessageObject().getClientId());
server.internalPublish(publishMessage);
}
} catch (Exception ex) {
LOG.error("error polling hazelcast msg queue", ex);
}
}
}
broker/src/main/java/io/moquette/server/Server.java
+52 -4
View File
@@ -15,7 +15,16 @@
*/
package io.moquette.server;
import com.hazelcast.config.ClasspathXmlConfig;
import com.hazelcast.config.Config;
import com.hazelcast.config.FileSystemXmlConfig;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.HazelcastInstanceNotActiveException;
import com.hazelcast.core.ITopic;
import io.moquette.BrokerConstants;
import io.moquette.interception.HazelcastInterceptHandler;
import io.moquette.interception.HazelcastMsg;
import io.moquette.interception.InterceptHandler;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.config.MemoryConfig;
@@ -50,6 +59,11 @@ public class Server {
private ProtocolProcessor m_processor;
private HazelcastInstance hazelcastInstance;
private SimpleMessaging m_messaging;
public static void main(String[] args) throws IOException {
final Server server = new Server();
server.startServer();
@@ -117,12 +131,29 @@ public class Server {
handlers = Collections.emptyList();
}
final String handlerProp = System.getProperty("intercept.handler");
final String handlerProp = System.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
if (handlerProp != null) {
config.setProperty("intercept.handler", handlerProp);
config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp);
}
LOG.info("Persistent store file: " + config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME));
final ProtocolProcessor processor = SimpleMessaging.getInstance().init(config, handlers, authenticator, authorizator);
if (config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME) !=null && config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME).equals(HazelcastInterceptHandler.class.getCanonicalName())) {
if (config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION) !=null){
Config hzconfig = null;
if (this.getClass().getClassLoader().getResource(config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION)) != null) {
hzconfig = new ClasspathXmlConfig(config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION));
} else {
hzconfig = new FileSystemXmlConfig(config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION));
}
LOG.info(String.format("starting server with hazelcast configuration file : %s",hzconfig));
hazelcastInstance = Hazelcast.newHazelcastInstance(hzconfig);
} else {
LOG.info("starting server with hazelcast default file");
hazelcastInstance = Hazelcast.newHazelcastInstance();
}
listenOnHazelCastMsg();
}
m_messaging = SimpleMessaging.getInstance();
final ProtocolProcessor processor = m_messaging.init(config, handlers, authenticator, authorizator, this);
if (sslCtxCreator == null) {
sslCtxCreator = new DefaultMoquetteSslContextCreator(config);
@@ -134,6 +165,16 @@ public class Server {
m_initialized = true;
}
private void listenOnHazelCastMsg() {
HazelcastInstance hz = getHazelcastInstance();
ITopic<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 only after the server is correctly started with startServer.
@@ -151,8 +192,15 @@ public class Server {
public void stopServer() {
LOG.info("Server stopping...");
m_acceptor.close();
SimpleMessaging.getInstance().shutdown();
m_messaging.shutdown();
m_initialized = false;
if (hazelcastInstance != null) {
try {
hazelcastInstance.shutdown();
} catch (HazelcastInstanceNotActiveException e) {
LOG.info("hazelcast already shutdown");
}
}
LOG.info("Server stopped");
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+39 -16
View File
@@ -108,12 +108,20 @@ public class ProtocolProcessor {
private ISessionsStore m_sessionsStore;
private IAuthenticator m_authenticator;
private BrokerInterceptor m_interceptor;
private String m_server_port;
//maps clientID to Will testament, if specified on CONNECT
private ConcurrentMap<String, WillMessage> m_willStore = new ConcurrentHashMap<>();
ProtocolProcessor() {}
public void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) {
init(subscriptions,storageService,sessionsStore,authenticator,allowAnonymous,authorizator,interceptor,null);
}
/**
* @param subscriptions the subscription store where are stored all the existing
* clients subscriptions.
@@ -124,11 +132,12 @@ public class ProtocolProcessor {
* @param allowAnonymous true connection to clients without credentials.
* @param authorizator used to apply ACL policies to publishes and subscriptions.
* @param interceptor to notify events to an intercept handler
* @param serverPort
*/
void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) {
boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) {
this.m_clientIDs = new ConcurrentHashMap<>();
this.m_interceptor = interceptor;
this.subscriptions = subscriptions;
@@ -138,6 +147,7 @@ public class ProtocolProcessor {
m_authenticator = authenticator;
m_messagesStore = storageService;
m_sessionsStore = sessionsStore;
m_server_port = serverPort;
}
public void processConnect(Channel channel, ConnectMessage msg) {
@@ -302,7 +312,7 @@ public class ProtocolProcessor {
//Remove the message from message store
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
verifyToActivate(targetSession);
StoredMessage inflightMsg = targetSession.getInflightMessage(messageID);
targetSession.inFlightAcknowledged(messageID);
@@ -311,7 +321,11 @@ public class ProtocolProcessor {
m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username));
}
private void verifyToActivate(String clientID, ClientSession targetSession) {
private void verifyToActivate(ClientSession targetSession) {
if (targetSession == null) {
return;
}
String clientID = targetSession.clientID;
if (m_clientIDs.containsKey(clientID)) {
targetSession.activate();
}
@@ -342,7 +356,7 @@ public class ProtocolProcessor {
}
final AbstractMessage.QOSType qos = msg.getQos();
final Integer messageID = msg.getMessageID();
LOG.info("PUBLISH from clientID <{}> on topic <{}> with QoS {}", clientID, topic, qos);
LOG.info("PUBLISH on server {} from clientID <{}> on topic <{}> with QoS {}", m_server_port, clientID, topic, qos);
String guid = null;
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
@@ -351,11 +365,15 @@ public class ProtocolProcessor {
route2Subscribers(toStoreMsg);
} else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1
route2Subscribers(toStoreMsg);
sendPubAck(clientID, messageID);
LOG.debug("replying with PubAck to MSG ID {}", messageID);
if (msg.isLocal()) {
sendPubAck(clientID, messageID);
}
LOG.info("server {} replying with PubAck to MSG ID {}", m_server_port, messageID);
} else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
sendPubRec(clientID, messageID);
if(msg.isLocal()) {
sendPubRec(clientID, messageID);
}
//Next the client will send us a pub rel
//NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
}
@@ -394,7 +412,11 @@ public class ProtocolProcessor {
String guid = null;
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
toStoreMsg.setClientID("BROKER_SELF");
if (msg.getClientId() == null || msg.getClientId().isEmpty()) {
toStoreMsg.setClientID("BROKER_SELF");
} else {
toStoreMsg.setClientID(msg.getClientId());
}
toStoreMsg.setMessageID(1);
if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
@@ -460,7 +482,7 @@ public class ProtocolProcessor {
qos = sub.getRequestedQos();
}
ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());
verifyToActivate(sub.getClientId(), targetSession);
verifyToActivate(targetSession);
LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
sub.getClientId(), sub.getTopicFilter(), qos, targetSession.isActive());
@@ -567,7 +589,7 @@ public class ProtocolProcessor {
int messageID = msg.getMessageID();
LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
verifyToActivate(targetSession);
IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID);
route2Subscribers(evt);
@@ -594,7 +616,7 @@ public class ProtocolProcessor {
public void processPubRec(Channel channel, PubRecMessage msg) {
String clientID = NettyUtils.clientID(channel);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
verifyToActivate(targetSession);
//remove from the inflight and move to the QoS2 second phase queue
int messageID = msg.getMessageID();
targetSession.moveInFlightToSecondPhaseAckWaiting(messageID);
@@ -613,7 +635,7 @@ public class ProtocolProcessor {
LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
//once received the PUBCOMP then remove the message from the temp memory
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
verifyToActivate(targetSession);
StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID);
String username = NettyUtils.userName(channel);
String topic = inflightMsg.getTopic();
@@ -669,7 +691,7 @@ public class ProtocolProcessor {
LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, clientSession);
verifyToActivate(clientSession);
for (String topic : topics) {
boolean validTopic = SubscriptionsStore.validate(topic);
if (!validTopic) {
@@ -695,10 +717,10 @@ public class ProtocolProcessor {
public void processSubscribe(Channel channel, SubscribeMessage msg) {
String clientID = NettyUtils.clientID(channel);
LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID());
LOG.debug("SUBSCRIBE client <{}> on server {} packetID {}", clientID, m_server_port, msg.getMessageID());
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, clientSession);
verifyToActivate(clientSession);
//ack the client
SubAckMessage ackMessage = new SubAckMessage();
ackMessage.setMessageID(msg.getMessageID());
@@ -756,8 +778,9 @@ public class ProtocolProcessor {
}
});
LOG.debug("Found {} messages to republish", messages.size());
ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId());
verifyToActivate(newSubscription.getClientId(), targetSession);
verifyToActivate(targetSession);
for (IMessagesStore.StoredMessage storedMsg : messages) {
//fire as retained the message
LOG.trace("send publish message for topic {}", newSubscription.getTopicFilter());
broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java
+18 -9
View File
@@ -16,6 +16,7 @@
package io.moquette.spi.impl;
import io.moquette.BrokerConstants;
import io.moquette.server.Server;
import io.moquette.spi.IMessagesStore;
import io.moquette.interception.InterceptHandler;
import io.moquette.server.config.IConfig;
@@ -29,6 +30,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.text.ParseException;
@@ -53,7 +55,7 @@ public class SimpleMessaging {
private BrokerInterceptor m_interceptor;
private static SimpleMessaging INSTANCE;
// private static SimpleMessaging INSTANCE;
private final ProtocolProcessor m_processor = new ProtocolProcessor();
@@ -61,10 +63,11 @@ public class SimpleMessaging {
}
public static SimpleMessaging getInstance() {
if (INSTANCE == null) {
INSTANCE = new SimpleMessaging();
}
return INSTANCE;
// if (INSTANCE == null) {
// INSTANCE = new SimpleMessaging();
// }
// return INSTANCE;
return new SimpleMessaging();
}
/**
@@ -79,7 +82,7 @@ public class SimpleMessaging {
* and fallback on the default one (permit all).
* */
public ProtocolProcessor init(IConfig props, List<? extends InterceptHandler> embeddedObservers,
IAuthenticator authenticator, IAuthorizator authorizator) {
IAuthenticator authenticator, IAuthorizator authorizator, Server server) {
subscriptions = new SubscriptionsStore();
m_mapStorage = new MapDBPersistentStore(props);
@@ -88,10 +91,16 @@ public class SimpleMessaging {
ISessionsStore sessionsStore = m_mapStorage.sessionsStore(messagesStore);
List<InterceptHandler> 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<? extends InterceptHandler> constructor = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).getConstructor(Server.class);
handler = constructor.newInstance(server);
} catch (NoSuchMethodException nsme){
handler = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).newInstance();
}
observers.add(handler);
} catch (Throwable ex) {
LOG.error("Can't load the intercept handler {}", ex);
@@ -143,7 +152,7 @@ public class SimpleMessaging {
}
boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
m_processor.init(subscriptions, messagesStore, sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor);
m_processor.init(subscriptions, messagesStore, sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
return m_processor;
}