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

Added the 'disabled' keyword in config to disable a listener on a port

Browse Source
This commit is contained in:
andsel
2016-05-18 00:11:19 +02:00
parent 04303eda75
commit e41e55f5aa
11 changed files with 29 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/BrokerConstants.java
+1
View File
@@ -44,6 +44,7 @@ public class BrokerConstants {
public static final String DB_AUTHENTICATOR_DIGEST = "authenticator.db.digest"; public static final String DB_AUTHENTICATOR_DIGEST = "authenticator.db.digest";
public static final int PORT = 1883; public static final int PORT = 1883;
public static final int WEBSOCKET_PORT = 8080; public static final int WEBSOCKET_PORT = 8080;
public static final String DISABLED_PORT_BIND = "disabled";
public static final String HOST = "0.0.0.0"; public static final String HOST = "0.0.0.0";
public static final String NEED_CLIENT_AUTH = "need_client_auth"; public static final String NEED_CLIENT_AUTH = "need_client_auth";
} }
broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java
+12 -11
View File
@@ -16,6 +16,7 @@
package io.moquette.server.netty; package io.moquette.server.netty;
import io.moquette.BrokerConstants; import io.moquette.BrokerConstants;
import static io.moquette.BrokerConstants.*;
import io.moquette.parser.commons.Constants; import io.moquette.parser.commons.Constants;
import io.moquette.parser.netty.MQTTDecoder; import io.moquette.parser.netty.MQTTDecoder;
import io.moquette.parser.netty.MQTTEncoder; import io.moquette.parser.netty.MQTTEncoder;
@@ -146,9 +147,9 @@ public class NettyAcceptor implements ServerAcceptor {
private void initializePlainTCPTransport(final NettyMQTTHandler handler, IConfig props) throws IOException { private void initializePlainTCPTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler(); final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME); String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
String tcpPortProp = props.getProperty(BrokerConstants.PORT_PROPERTY_NAME); String tcpPortProp = props.getProperty(PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
if (tcpPortProp == null || "".equals(tcpPortProp)) { if (DISABLED_PORT_BIND.equals(tcpPortProp)) {
LOG.info("tcp MQTT is disabled because the value for the property with key BrokerConstants.PORT_PROPERTY_NAME is null"); LOG.info("tcp MQTT is disabled because the value for the property with key {}", BrokerConstants.PORT_PROPERTY_NAME);
return; return;
} }
int port = Integer.parseInt(tcpPortProp); int port = Integer.parseInt(tcpPortProp);
@@ -168,8 +169,8 @@ public class NettyAcceptor implements ServerAcceptor {
} }
private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException { private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
String webSocketPortProp = props.getProperty(BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME); String webSocketPortProp = props.getProperty(WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
if (webSocketPortProp == null || "".equals(webSocketPortProp)) { if (DISABLED_PORT_BIND.equals(webSocketPortProp)) {
//Do nothing no WebSocket configured //Do nothing no WebSocket configured
LOG.info("WebSocket is disabled"); LOG.info("WebSocket is disabled");
return; return;
@@ -200,10 +201,10 @@ public class NettyAcceptor implements ServerAcceptor {
} }
private void initializeSSLTCPTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext) throws IOException { private void initializeSSLTCPTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext) throws IOException {
String sslPortProp = props.getProperty(BrokerConstants.SSL_PORT_PROPERTY_NAME); String sslPortProp = props.getProperty(SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
if (sslPortProp == null || "".equals(sslPortProp)) { if (DISABLED_PORT_BIND.equals(sslPortProp)) {
//Do nothing no SSL configured //Do nothing no SSL configured
LOG.info("SSL MQTT is disabled because there is no value in properties for key BrokerConstants.SSL_PORT_PROPERTY_NAME"); LOG.info("SSL MQTT is disabled because there is no value in properties for key {}", BrokerConstants.SSL_PORT_PROPERTY_NAME);
return; return;
} }
@@ -231,10 +232,10 @@ public class NettyAcceptor implements ServerAcceptor {
} }
private void initializeWSSTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext) throws IOException { private void initializeWSSTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext) throws IOException {
String sslPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME); String sslPortProp = props.getProperty(WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
if (sslPortProp == null || "".equals(sslPortProp)) { if (DISABLED_PORT_BIND.equals(sslPortProp)) {
//Do nothing no SSL configured //Do nothing no SSL configured
LOG.info("SSL websocket is disabled because there is no value in properties for key BrokerConstants.WSS_PORT_PROPERTY_NAME"); LOG.info("SSL websocket is disabled because there is no value in properties for key {}", BrokerConstants.WSS_PORT_PROPERTY_NAME);
return; return;
} }
int sslPort = Integer.parseInt(sslPortProp); int sslPort = Integer.parseInt(sslPortProp);
broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java
+2 -2
View File
@@ -55,7 +55,7 @@ public class ConfigurationClassLoaderTest implements IAuthenticator, IAuthorizat
@Test @Test
public void loadAuthenticator() throws Exception { public void loadAuthenticator() throws Exception {
Properties props = new Properties(IntegrationUtils.prepareTestPropeties()); Properties props = new Properties(IntegrationUtils.prepareTestProperties());
props.setProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, "io.moquette.server.ConfigurationClassLoaderTest"); props.setProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, "io.moquette.server.ConfigurationClassLoaderTest");
startServer(props); startServer(props);
assertTrue(true); assertTrue(true);
@@ -64,7 +64,7 @@ public class ConfigurationClassLoaderTest implements IAuthenticator, IAuthorizat
@Test @Test
public void loadAuthorizator() throws Exception { public void loadAuthorizator() throws Exception {
Properties props = new Properties(IntegrationUtils.prepareTestPropeties()); Properties props = new Properties(IntegrationUtils.prepareTestProperties());
props.setProperty(BrokerConstants.AUTHORIZATOR_CLASS_NAME, "io.moquette.server.ConfigurationClassLoaderTest"); props.setProperty(BrokerConstants.AUTHORIZATOR_CLASS_NAME, "io.moquette.server.ConfigurationClassLoaderTest");
startServer(props); startServer(props);
assertTrue(true); assertTrue(true);
broker/src/test/java/io/moquette/server/IntegrationUtils.java
+3 -1
View File
@@ -22,6 +22,7 @@ import java.util.Properties;
import static io.moquette.BrokerConstants.DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; import static io.moquette.BrokerConstants.DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME;
import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME; import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME;
import static io.moquette.BrokerConstants.PORT_PROPERTY_NAME;
import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertFalse;
/** /**
@@ -35,9 +36,10 @@ public class IntegrationUtils {
return currentDir + File.separator + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; return currentDir + File.separator + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME;
} }
public static Properties prepareTestPropeties() { public static Properties prepareTestProperties() {
Properties testProperties = new Properties(); Properties testProperties = new Properties();
testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath()); testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath());
testProperties.put(PORT_PROPERTY_NAME, "1883");
return testProperties; return testProperties;
} }
broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java
+1 -1
View File
@@ -49,7 +49,7 @@ public class ServerIntegrationEmbeddedPublishTest {
protected void startServer() throws IOException { protected void startServer() throws IOException {
m_server = new Server(); m_server = new Server();
final Properties configProps = IntegrationUtils.prepareTestPropeties(); final Properties configProps = IntegrationUtils.prepareTestProperties();
m_config = new MemoryConfig(configProps); m_config = new MemoryConfig(configProps);
m_server.startServer(m_config); m_server.startServer(m_config);
} }
broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java
+1 -1
View File
@@ -45,7 +45,7 @@ public class ServerIntegrationFuseTest {
protected void startServer() throws IOException { protected void startServer() throws IOException {
m_server = new Server(); m_server = new Server();
final Properties configProps = IntegrationUtils.prepareTestPropeties(); final Properties configProps = IntegrationUtils.prepareTestProperties();
m_config = new MemoryConfig(configProps); m_config = new MemoryConfig(configProps);
m_server.startServer(m_config); m_server.startServer(m_config);
} }
broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java
+2 -2
View File
@@ -54,7 +54,7 @@ public class ServerIntegrationPahoTest {
protected void startServer() throws IOException { protected void startServer() throws IOException {
m_server = new Server(); m_server = new Server();
final Properties configProps = IntegrationUtils.prepareTestPropeties(); final Properties configProps = IntegrationUtils.prepareTestProperties();
m_config = new MemoryConfig(configProps); m_config = new MemoryConfig(configProps);
m_server.startServer(m_config); m_server.startServer(m_config);
} }
@@ -193,7 +193,7 @@ public class ServerIntegrationPahoTest {
m_server.stopServer(); m_server.stopServer();
m_server.startServer(IntegrationUtils.prepareTestPropeties()); m_server.startServer(IntegrationUtils.prepareTestProperties());
//reconnect and publish //reconnect and publish
m_client.connect(options); m_client.connect(options);
broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java
+1 -1
View File
@@ -62,7 +62,7 @@ public class ServerIntegrationQoSValidationTest {
protected void startServer() throws IOException { protected void startServer() throws IOException {
m_server = new Server(); m_server = new Server();
final Properties configProps = IntegrationUtils.prepareTestPropeties(); final Properties configProps = IntegrationUtils.prepareTestProperties();
m_config = new MemoryConfig(configProps); m_config = new MemoryConfig(configProps);
m_server.startServer(m_config); m_server.startServer(m_config);
} }
broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java
+1 -1
View File
@@ -48,7 +48,7 @@ public class ServerIntegrationWebSocketTest {
protected void startServer() throws IOException { protected void startServer() throws IOException {
m_server = new Server(); m_server = new Server();
final Properties configProps = IntegrationUtils.prepareTestPropeties(); final Properties configProps = IntegrationUtils.prepareTestProperties();
configProps.put(BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, Integer.toString(BrokerConstants.WEBSOCKET_PORT)); configProps.put(BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, Integer.toString(BrokerConstants.WEBSOCKET_PORT));
m_config = new MemoryConfig(configProps); m_config = new MemoryConfig(configProps);
m_server.startServer(m_config); m_server.startServer(m_config);
broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java
+1 -2
View File
@@ -17,7 +17,6 @@ package io.moquette.server;
import io.moquette.parser.proto.messages.ConnectMessage; import io.moquette.parser.proto.messages.ConnectMessage;
import io.moquette.parser.proto.messages.AbstractMessage; import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.parser.proto.messages.AbstractMessage.QOSType;
import io.moquette.parser.proto.messages.ConnAckMessage; import io.moquette.parser.proto.messages.ConnAckMessage;
import io.moquette.server.config.IConfig; import io.moquette.server.config.IConfig;
import io.moquette.server.config.MemoryConfig; import io.moquette.server.config.MemoryConfig;
@@ -50,7 +49,7 @@ public class ServerLowlevelMessagesIntegrationTests {
protected void startServer() throws IOException { protected void startServer() throws IOException {
m_server = new Server(); m_server = new Server();
final Properties configProps = IntegrationUtils.prepareTestPropeties(); final Properties configProps = IntegrationUtils.prepareTestProperties();
m_config = new MemoryConfig(configProps); m_config = new MemoryConfig(configProps);
m_server.startServer(m_config); m_server.startServer(m_config);
} }
broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java
+4 -4
View File
@@ -48,7 +48,7 @@ public class ServerRestartIntegrationTest {
protected void startServer() throws IOException { protected void startServer() throws IOException {
m_server = new Server(); m_server = new Server();
final Properties configProps = IntegrationUtils.prepareTestPropeties(); final Properties configProps = IntegrationUtils.prepareTestProperties();
m_config = new MemoryConfig(configProps); m_config = new MemoryConfig(configProps);
m_server.startServer(m_config); m_server.startServer(m_config);
} }
@@ -101,7 +101,7 @@ public class ServerRestartIntegrationTest {
IntegrationUtils.cleanPersistenceFile(m_config); IntegrationUtils.cleanPersistenceFile(m_config);
//restart the server //restart the server
m_server.startServer(IntegrationUtils.prepareTestPropeties()); m_server.startServer(IntegrationUtils.prepareTestProperties());
//reconnect the Subscriber subscribing to the same /topic but different QoS //reconnect the Subscriber subscribing to the same /topic but different QoS
m_subscriber.connect(CLEAN_SESSION_OPT); m_subscriber.connect(CLEAN_SESSION_OPT);
@@ -128,7 +128,7 @@ public class ServerRestartIntegrationTest {
m_server.stopServer(); m_server.stopServer();
//restart the server //restart the server
m_server.startServer(IntegrationUtils.prepareTestPropeties()); m_server.startServer(IntegrationUtils.prepareTestProperties());
m_publisher.connect(); m_publisher.connect();
m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), 0, false); m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), 0, false);
@@ -148,7 +148,7 @@ public class ServerRestartIntegrationTest {
m_server.stopServer(); m_server.stopServer();
//restart the server //restart the server
m_server.startServer(IntegrationUtils.prepareTestPropeties()); m_server.startServer(IntegrationUtils.prepareTestProperties());
//subscriber reconnects //subscriber reconnects
m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_dataStore); m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_dataStore);
m_subscriber.setCallback(m_messageCollector); m_subscriber.setCallback(m_messageCollector);