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

Fixed Hazelcast itnegration tests, removing unusefull thread sleeps, but more important permitting the co existence of 2 Brokers in the same JVM

Browse Source
This commit is contained in:
andsel
2016-07-22 12:09:10 +02:00
parent 7886bf7103
commit fedd2965cc
8 changed files with 60 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java
+3 -5
View File
@@ -15,18 +15,16 @@ 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())));
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());
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
+2 -3
View File
@@ -5,7 +5,7 @@ import java.io.Serializable;
/**
* Created by mackristof on 28/05/2016.
*/
public class HazelcastMsg implements Serializable{
public class HazelcastMsg implements Serializable {
private final String clientId;
private final String userName;
@@ -13,7 +13,7 @@ public class HazelcastMsg implements Serializable{
private final byte[] payload;
private final String topic;
public HazelcastMsg(String clientId, String topic , byte qos, byte[] payload, String userName) {
public HazelcastMsg(String clientId, String topic, byte qos, byte[] payload, String userName) {
this.clientId = clientId;
this.qos = qos;
this.payload = payload;
@@ -21,7 +21,6 @@ public class HazelcastMsg implements Serializable{
this.topic = topic;
}
public String getClientId() {
return clientId;
}
broker/src/main/java/io/moquette/server/HazelcastListener.java
+4 -6
View File
@@ -22,13 +22,12 @@ public class HazelcastListener implements MessageListener<HazelcastMsg> {
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()));
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()));
@@ -36,10 +35,9 @@ public class HazelcastListener implements MessageListener<HazelcastMsg> {
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);
} catch (Exception ex) {
LOG.error("error polling hazelcast msg queue", ex);
}
}
}
broker/src/main/java/io/moquette/server/Server.java
+10 -12
View File
@@ -31,10 +31,8 @@ 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;
@@ -63,6 +61,8 @@ public class Server {
private HazelcastInstance hazelcastInstance;
private SimpleMessaging m_messaging;
public static void main(String[] args) throws IOException {
final Server server = new Server();
@@ -131,9 +131,9 @@ 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));
if (config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME) !=null && config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME).equals(HazelcastInterceptHandler.class.getCanonicalName())) {
@@ -152,7 +152,8 @@ public class Server {
}
listenOnHazelCastMsg();
}
final ProtocolProcessor processor = SimpleMessaging.getInstance().init(config, handlers, authenticator, authorizator, this);
m_messaging = SimpleMessaging.getInstance();
final ProtocolProcessor processor = m_messaging.init(config, handlers, authenticator, authorizator, this);
if (sslCtxCreator == null) {
sslCtxCreator = new DefaultMoquetteSslContextCreator(config);
@@ -162,8 +163,6 @@ public class Server {
m_acceptor.initialize(processor, config, sslCtxCreator);
m_processor = processor;
m_initialized = true;
}
private void listenOnHazelCastMsg() {
@@ -193,15 +192,14 @@ 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 ){
if (hazelcastInstance != null) {
try {
hazelcastInstance.shutdown();
} catch (HazelcastInstanceNotActiveException e){
LOG.info("hazelcast allready shutdown");
} catch (HazelcastInstanceNotActiveException e) {
LOG.info("hazelcast already shutdown");
}
}
LOG.info("Server stopped");
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+3 -3
View File
@@ -364,11 +364,10 @@ public class ProtocolProcessor {
route2Subscribers(toStoreMsg);
} else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1
route2Subscribers(toStoreMsg);
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);
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);
if(msg.isLocal()) {
@@ -717,7 +716,7 @@ 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(clientSession);
@@ -778,6 +777,7 @@ public class ProtocolProcessor {
}
});
LOG.debug("Found {} messages to republish", messages.size());
ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId());
verifyToActivate(targetSession);
for (IMessagesStore.StoredMessage storedMsg : messages) {
broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java
+7 -6
View File
@@ -55,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();
@@ -63,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();
}
/**
@@ -97,7 +98,7 @@ public class SimpleMessaging {
try {
final Constructor<? extends InterceptHandler> constructor = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).getConstructor(Server.class);
handler = constructor.newInstance(server);
}catch (NoSuchMethodException nsme){
} catch (NoSuchMethodException nsme){
handler = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).newInstance();
}
observers.add(handler);
broker/src/test/java/io/moquette/server/IntegrationUtils.java
+12
View File
@@ -36,6 +36,11 @@ public class IntegrationUtils {
return currentDir + File.separator + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME;
}
static String localClusterMapDBPath(int port) {
String currentDir = System.getProperty("user.dir");
return currentDir + File.separator + port + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME;
}
public static Properties prepareTestProperties() {
Properties testProperties = new Properties();
testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath());
@@ -43,6 +48,13 @@ public class IntegrationUtils {
return testProperties;
}
public static Properties prepareTestClusterProperties(int port) {
Properties testProperties = new Properties();
testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localClusterMapDBPath(port));
testProperties.put(PORT_PROPERTY_NAME, Integer.toString(port));
return testProperties;
}
public static void cleanPersistenceFile(IConfig config) {
String fileName = config.getProperty(PERSISTENT_STORE_PROPERTY_NAME);
cleanPersistenceFile(fileName);
broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java
+19 -36
View File
@@ -38,32 +38,25 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest{
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 {
protected Server startServer(int port, 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_config.setProperty(BrokerConstants.PORT_PROPERTY_NAME, Integer.toString(port));
m_server.startServer(m_config);
return m_server;
}
private void stopServer(Server m_server, IConfig m_config) {
private void stopServer(Server m_server) {
m_server.stopServer();
}
private Properties addHazelCastConf(Properties properties,int port, String hazelcastConfigurationFile) {
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);
@@ -75,8 +68,16 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest{
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);
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();
@@ -85,12 +86,8 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest{
m_listener.setCallback(m_messagesCollector);
m_listener.connect();
}
@After
public void tearDown() throws Exception {
if (m_listener != null && m_listener.isConnected()) {
@@ -101,23 +98,16 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest{
m_publisher.disconnect();
}
stopServer(server1883, m_config_1883);
stopServer(server1884, m_config_1884);
}
@AfterClass
public static void shutdown(){
stopServer(server1883);
stopServer(server1884);
}
@Test
public void checkPublishOnHazelCastQueue() throws Exception {
LOG.info("*** checkPublishOnHazelCastQueue ***");
m_listener.subscribe("/topic", 0);
public void checkPublishPassThroughCluster_Qos0() throws Exception {
LOG.info("*** checkPublishPassThroughCluster_Qos0 ***");
m_listener.subscribe("/topic", 1);
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());
@@ -135,12 +125,5 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest{
MqttMessage messageQos2 = m_messagesCollector.getMessage(true);
assertEquals("Hello world MQTT QoS2", messageQos2.toString());
assertEquals(2, messageQos2.getQos());
Thread.sleep(5000);
}
}