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

Merge branch 'feat/cluster' of https://github.com/mackristof/moquette into mackristof-feat/cluster

Browse Source
This commit is contained in:
andsel
2016-07-21 16:17:15 +02:00
parent 1ef2a4fd3c 2dd18c2b70
commit 7886bf7103
25 changed files with 472 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/pom.xml
+6
View File
@@ -101,6 +101,12 @@
<version>2.1.4</version>
</dependency>
<dependency>
<groupId>com.hazelcast</groupId>
<artifactId>hazelcast</artifactId>
<version>3.5</version>
</dependency>
<!-- Test dependencies -->
<dependency>
<groupId>junit</groupId>
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
+33
View File
@@ -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<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
+44
View File
@@ -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;
}
}
broker/src/main/java/io/moquette/server/HazelcastListener.java
+45
View File
@@ -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<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(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);
}
}
}
broker/src/main/java/io/moquette/server/Server.java
+51 -1
View File
@@ -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<HazelcastMsg> 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");
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+27 -8
View File
@@ -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<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.
@@ -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);
broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java
+12 -4
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;
@@ -79,7 +81,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 +90,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 +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;
}
broker/src/main/java/io/moquette/spi/impl/security/AcceptAllAuthenticator.java
+1 -1
View File
@@ -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;
}
}
broker/src/main/java/io/moquette/spi/impl/security/DBAuthenticator.java
+1 -1
View File
@@ -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");
broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java
+3 -2
View File
@@ -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;
broker/src/main/java/io/moquette/spi/security/IAuthenticator.java
+1 -1
View File
@@ -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);
}
broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java
+1 -1
View File
@@ -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;
}
broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java
+146
View File
@@ -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);
}
}
broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java
+9 -3
View File
@@ -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<String> m_clientIds;
private Map<String, byte[]> m_userPwds;
MockAuthenticator(Map<String, byte[]> userPwds) {
MockAuthenticator(Set<String> clientIds, Map<String, byte[]> 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;
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+5 -2
View File
@@ -83,10 +83,13 @@ public class ProtocolProcessorTest {
m_messagesStore = memStorage.messagesStore();
m_sessionStore = memStorage.sessionsStore();
//m_messagesStore.initStore();
Set<String> clientIds = new HashSet<>();
clientIds.add(FAKE_CLIENT_ID);
clientIds.add(FAKE_CLIENT_ID2);
Map<String, byte[]> 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());
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java
+22 -18
View File
@@ -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<String> clientIds = new HashSet<>();
clientIds.add(ProtocolProcessorTest.FAKE_CLIENT_ID);
Map<String, byte[]> 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());
broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java
+3 -3
View File
@@ -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()));
}
broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java
+3 -3
View File
@@ -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()));
}
}
broker/src/test/resources/config/hazelcast.xml
+10
View File
@@ -0,0 +1,10 @@
<hazelcast
xsi:schemaLocation="http://www.hazelcast.com/schema/config
http://www.hazelcast.com/schema/config/hazelcast-config-3.5.xsd"
xmlns="http://www.hazelcast.com/schema/config"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<network>
<join><multicast enabled="true"/></join>
</network>
</hazelcast>
broker/src/test/resources/hazelcast.xml
+10
View File
@@ -0,0 +1,10 @@
<hazelcast
xsi:schemaLocation="http://www.hazelcast.com/schema/config
http://www.hazelcast.com/schema/config/hazelcast-config-3.5.xsd"
xmlns="http://www.hazelcast.com/schema/config"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<network>
<join><multicast enabled="true"/></join>
</network>
</hazelcast>
broker/src/test/resources/log4j.properties
+4
View File
@@ -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
distribution/src/main/assembly/assembly.xml
+5
View File
@@ -36,6 +36,11 @@
<source>src/main/resources/password_file.conf</source>
<outputDirectory>config/</outputDirectory>
</file>
<file>
<source>src/main/resources/hazelcast.xml</source>
<outputDirectory>config/</outputDirectory>
</file>
<file>
<source>src/main/resources/README.txt</source>
distribution/src/main/resources/hazelcast.xml
+10
View File
@@ -0,0 +1,10 @@
<hazelcast
xsi:schemaLocation="http://www.hazelcast.com/schema/config
http://www.hazelcast.com/schema/config/hazelcast-config-3.5.xsd"
xmlns="http://www.hazelcast.com/schema/config"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<network>
<join><multicast enabled="true"/></join>
</network>
</hazelcast>
netty_parser/src/main/java/io/moquette/parser/proto/messages/PublishMessage.java
+18
View File
@@ -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;
}
}