From f41f7dafb98a2b281b5cad97f5d829bc3f1b65af Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 8 Dec 2015 12:06:31 +0100 Subject: [PATCH] Added sample publish InterceptHandler implementation to EmbeddedLauncher (#104 publish listner part) --- .../AbstractInterceptHandler.java | 41 ++++++++++++++++ .../messages/InterceptAbstractMessage.java | 15 ++++++ .../main/java/io/moquette/server/Server.java | 17 ++++++- .../server/config/ClasspathConfig.java | 2 +- .../io/moquette/spi/impl/SimpleMessaging.java | 4 +- embedding_moquette/pom.xml | 1 - .../testembedded/EmbeddedLauncher.java | 49 ++++++++++++++++--- .../src/main/resources/config/moquette.conf | 22 +++++++++ 8 files changed, 139 insertions(+), 12 deletions(-) create mode 100644 broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java create mode 100644 embedding_moquette/src/main/resources/config/moquette.conf diff --git a/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java b/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java new file mode 100644 index 00000000..9b0fc56e --- /dev/null +++ b/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java @@ -0,0 +1,41 @@ +/* + * Copyright (c) 2012-2015 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.*; + +/** + * Basic abstract class usefull to avoid empty methods creation in subclasses. + * + * Created by andrea on 08/12/15. + */ +public abstract class AbstractInterceptHandler implements InterceptHandler { + + @Override + public void onConnect(InterceptConnectMessage msg) {} + + @Override + public void onDisconnect(InterceptDisconnectMessage msg) {} + + @Override + public void onPublish(InterceptPublishMessage msg) {} + + @Override + public void onSubscribe(InterceptSubscribeMessage msg) {} + + @Override + public void onUnsubscribe(InterceptUnsubscribeMessage msg) {} +} diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptAbstractMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptAbstractMessage.java index a3a7a159..88a4b136 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptAbstractMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptAbstractMessage.java @@ -1,3 +1,18 @@ +/* + * Copyright (c) 2012-2015 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.messages; import io.moquette.proto.messages.AbstractMessage; diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index bfd33389..24423214 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -15,6 +15,7 @@ */ package io.moquette.server; +import io.moquette.interception.InterceptHandler; import io.moquette.server.config.MemoryConfig; import io.moquette.spi.impl.SimpleMessaging; import io.moquette.server.config.FilesystemConfig; @@ -26,6 +27,9 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import java.util.Properties; /** @@ -87,12 +91,23 @@ public class Server { * Starts Moquette bringing the configuration files from the given Config implementation. */ public void startServer(IConfig config) throws IOException { + startServer(config, null); + } + + /** + * Starts Moquette with config provided by an implementation of IConfig class and with the set of InterceptHandler + * */ + public void startServer(IConfig config, List handlers) throws IOException { + if (handlers == null) { + handlers = Collections.emptyList(); + } + final String handlerProp = System.getProperty("intercept.handler"); if (handlerProp != null) { config.setProperty("intercept.handler", handlerProp); } LOG.info("Persistent store file: " + config.getProperty(io.moquette.commons.Constants.PERSISTENT_STORE_PROPERTY_NAME)); - final ProtocolProcessor processor = SimpleMessaging.getInstance().init(config); + final ProtocolProcessor processor = SimpleMessaging.getInstance().init(config, handlers); m_acceptor = new NettyAcceptor(); m_acceptor.initialize(processor, config); diff --git a/broker/src/main/java/io/moquette/server/config/ClasspathConfig.java b/broker/src/main/java/io/moquette/server/config/ClasspathConfig.java index 7180fa9d..b36ac4c4 100644 --- a/broker/src/main/java/io/moquette/server/config/ClasspathConfig.java +++ b/broker/src/main/java/io/moquette/server/config/ClasspathConfig.java @@ -39,7 +39,7 @@ public class ClasspathConfig implements IConfig { ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); InputStream is = classLoader.getResourceAsStream("config/moquette.conf"); if (is == null) { - throw new RuntimeException("Can't locate the resourse \"config/moquette.conf\""); + throw new RuntimeException("Can't locate the resource \"config/moquette.conf\""); } Reader configReader = new InputStreamReader(is); ConfigurationParser confParser = new ConfigurationParser(); 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 2a091057..83f74f1f 100644 --- a/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java @@ -66,7 +66,7 @@ public class SimpleMessaging { return INSTANCE; } - public ProtocolProcessor init(IConfig props) { + public ProtocolProcessor init(IConfig props, List embeddedObservers) { subscriptions = new SubscriptionsStore(); m_mapStorage = new MapDBPersistentStore(props); @@ -74,7 +74,7 @@ public class SimpleMessaging { IMessagesStore messagesStore = m_mapStorage.messagesStore(); ISessionsStore sessionsStore = m_mapStorage.sessionsStore(messagesStore); - List observers = new ArrayList<>(); + List observers = new ArrayList<>(embeddedObservers); String interceptorClassName = props.getProperty("intercept.handler"); if (interceptorClassName != null && !interceptorClassName.isEmpty()) { try { diff --git a/embedding_moquette/pom.xml b/embedding_moquette/pom.xml index a9f40233..73c6d9d6 100644 --- a/embedding_moquette/pom.xml +++ b/embedding_moquette/pom.xml @@ -1,4 +1,3 @@ - 4.0.0 diff --git a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java index 04c6c540..a6644514 100644 --- a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java +++ b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java @@ -1,21 +1,56 @@ + /* + * Copyright (c) 2012-2015 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.testembedded; +import io.moquette.interception.AbstractInterceptHandler; +import io.moquette.interception.InterceptHandler; +import io.moquette.interception.messages.*; import io.moquette.server.Server; import io.moquette.server.config.IConfig; import io.moquette.server.config.ClasspathConfig; import java.io.IOException; +import java.util.List; + +import static java.util.Arrays.asList; + + public class EmbeddedLauncher { + static class PublisherListener extends AbstractInterceptHandler { + + @Override + public void onPublish(InterceptPublishMessage msg) { + System.out.println("Received on topic: " + msg.getTopicName() + " content: " + new String(msg.getPayload().array())); + } + } -public class EmbeddedLauncher { public static void main(String[] args) throws InterruptedException, IOException { final IConfig classPathConfig = new ClasspathConfig(); final Server mqttBroker = new Server(); - mqttBroker.startServer(classPathConfig); - System.out.println("Broker started"); - Thread.sleep(1000); - System.out.println("Stopping broker"); - mqttBroker.stopServer(); - System.out.println("Broker stopped"); + List userHandlers = asList(new PublisherListener()); + mqttBroker.startServer(classPathConfig, userHandlers); + System.out.println("Broker started press [CRTL+C] to stop"); + //Bind a shutdown hook + Runtime.getRuntime().addShutdownHook(new Thread() { + @Override + public void run() { + System.out.println("Stopping broker"); + mqttBroker.stopServer(); + System.out.println("Broker stopped"); + } + }); } } \ No newline at end of file diff --git a/embedding_moquette/src/main/resources/config/moquette.conf b/embedding_moquette/src/main/resources/config/moquette.conf new file mode 100644 index 00000000..dcac3dcb --- /dev/null +++ b/embedding_moquette/src/main/resources/config/moquette.conf @@ -0,0 +1,22 @@ +############################################## +# Moquette configuration file. +# +# The synthax is equals to mosquitto.conf +# +############################################## + +port 1883 + +#websocket_port 8080 + +host 0.0.0.0 + +#Password file +#password_file password_file.conf + +#ssl_port 8883 +#jks_path serverkeystore.jks +#key_store_password passw0rdsrv +#key_manager_password passw0rdsrv + +allow_anonymous true