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

Added sample publish InterceptHandler implementation to EmbeddedLauncher (#104 publish listner part)

Browse Source
This commit is contained in:
andsel
2015-12-08 12:07:03 +01:00
parent 5a1cd97b71
commit f41f7dafb9
8 changed files with 139 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java
+41
View File
@@ -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) {}
}
broker/src/main/java/io/moquette/interception/messages/InterceptAbstractMessage.java
+15
View File
@@ -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;
broker/src/main/java/io/moquette/server/Server.java
+16 -1
View File
@@ -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<? extends InterceptHandler> 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);
broker/src/main/java/io/moquette/server/config/ClasspathConfig.java
+1 -1
View File
@@ -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();
broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java
+2 -2
View File
@@ -66,7 +66,7 @@ public class SimpleMessaging {
return INSTANCE;
}
public ProtocolProcessor init(IConfig props) {
public ProtocolProcessor init(IConfig props, List<? extends InterceptHandler> 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<InterceptHandler> observers = new ArrayList<>();
List<InterceptHandler> observers = new ArrayList<>(embeddedObservers);
String interceptorClassName = props.getProperty("intercept.handler");
if (interceptorClassName != null && !interceptorClassName.isEmpty()) {
try {
embedding_moquette/pom.xml
-1
View File
@@ -1,4 +1,3 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java
+42 -7
View File
@@ -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<? extends InterceptHandler> 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");
}
});
}
}
embedding_moquette/src/main/resources/config/moquette.conf
+22
View File
@@ -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