From fedd2965ccf100e8fd8acc3684899593db3fafb8 Mon Sep 17 00:00:00 2001 From: andsel Date: Fri, 22 Jul 2016 12:09:10 +0200 Subject: [PATCH] Fixed Hazelcast itnegration tests, removing unusefull thread sleeps, but more important permitting the co existence of 2 Brokers in the same JVM --- .../HazelcastInterceptHandler.java | 8 +-- .../moquette/interception/HazelcastMsg.java | 5 +- .../io/moquette/server/HazelcastListener.java | 10 ++-- .../main/java/io/moquette/server/Server.java | 22 ++++---- .../moquette/spi/impl/ProtocolProcessor.java | 6 +- .../io/moquette/spi/impl/SimpleMessaging.java | 13 +++-- .../io/moquette/server/IntegrationUtils.java | 12 ++++ ...rationHazelcastHandlerInterceptorTest.java | 55 +++++++------------ 8 files changed, 60 insertions(+), 71 deletions(-) diff --git a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java index 75c58c3f..7792a424 100644 --- a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java @@ -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 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); } diff --git a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java index 6ba7deeb..4d400764 100644 --- a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java +++ b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java @@ -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; } diff --git a/broker/src/main/java/io/moquette/server/HazelcastListener.java b/broker/src/main/java/io/moquette/server/HazelcastListener.java index 7a3f13cf..13738441 100644 --- a/broker/src/main/java/io/moquette/server/HazelcastListener.java +++ b/broker/src/main/java/io/moquette/server/HazelcastListener.java @@ -22,13 +22,12 @@ public class HazelcastListener implements MessageListener { this.server = server; } - - @Override public void onMessage(Message 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 { 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); } } } diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index f57a9fd9..73c409c8 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -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"); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index ef0ae7d0..f7804e6a 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -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) { diff --git a/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java index b0c35701..380dc4e4 100644 --- a/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java @@ -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 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); diff --git a/broker/src/test/java/io/moquette/server/IntegrationUtils.java b/broker/src/test/java/io/moquette/server/IntegrationUtils.java index a065aa69..9559487e 100644 --- a/broker/src/test/java/io/moquette/server/IntegrationUtils.java +++ b/broker/src/test/java/io/moquette/server/IntegrationUtils.java @@ -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); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java index 6adba101..ca6e7174 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java @@ -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); - - } - - - }