diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java b/broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java new file mode 100644 index 00000000..094aadac --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java @@ -0,0 +1,103 @@ +package org.dna.mqtt.moquette.server; + +import java.io.BufferedReader; +import java.io.File; +import java.io.FileNotFoundException; +import java.io.FileReader; +import java.io.IOException; +import java.io.Reader; +import java.text.ParseException; +import java.util.Properties; +import java.util.logging.Level; +import org.dna.mqtt.commons.Constants; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Mosquitto configuration parser. + * + * A line that at the very first has # is a comment + * Each line has key value format, where the separator used it the space. + * + * @author andrea + */ +class ConfigurationParser { + + private static final Logger LOG = LoggerFactory.getLogger(ConfigurationParser.class); + + private Properties m_properties = new Properties(); + + ConfigurationParser() { + m_properties.put("port", Integer.toString(Constants.PORT)); + m_properties.put("host", Constants.HOST); + } + + /** + * Parse the configuration from file. + */ + void parse(File file) throws ParseException { + if (file == null) { + LOG.warn("parsing NULL file, so fallback on default configuration!"); + return; + } + if (!file.exists()) { + LOG.warn(String.format("parsing not existing file %s, so fallback on default configuration!", file.getAbsolutePath())); + return; + } + try { + FileReader reader = new FileReader(file); + parse(reader); + } catch (FileNotFoundException fex) { + LOG.warn(String.format("parsing not existing file %s, so fallback on default configuration!", file.getAbsolutePath()), fex); + return; + } + } + + /** + * Parse the configuration + * + * @throws ParseException if the format is not compliant. + */ + void parse(Reader reader) throws ParseException { + if (reader == null) { + //just log and return default properties + LOG.warn("parsing NULL reader, so fallback on default configuration!"); + return; + } + + BufferedReader br = new BufferedReader(reader); + String line; + try { + while ((line = br.readLine()) != null) { + int commentMarker = line.indexOf('#'); + if (commentMarker != -1) { + if (commentMarker == 0) { + //skip its a comment + continue; + } else { + //it's a malformed comment + throw new ParseException(line, commentMarker); + } + } else { + if (line.isEmpty() || line.matches("^\\s*$")) { + //skip it's a black line + continue; + } + + //split till the first space + int deilimiterIdx = line.indexOf(' '); + String key = line.substring(0, deilimiterIdx).trim(); + String value = line.substring(deilimiterIdx).trim(); + + m_properties.put(key, value); + } + } + } catch (IOException ex) { + throw new ParseException("Failed to read", 1); + } + } + + Properties getProperties() { + return m_properties; + } +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java index 3a520e43..b3eac7d3 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -2,6 +2,7 @@ package org.dna.mqtt.moquette.server; import java.io.File; import java.io.IOException; +import java.text.ParseException; import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; import org.dna.mqtt.moquette.server.netty.NettyAcceptor; import org.slf4j.Logger; @@ -36,11 +37,18 @@ public class Server { } public void startServer() throws IOException { + ConfigurationParser confParser = new ConfigurationParser(); + try { + confParser.parse(new File("config/moquette.conf")); + } catch (ParseException pex) { + LOG.warn("An error occured in parsing configuration, fallback on deafult configuration", pex); + } + messaging = SimpleMessaging.getInstance(); messaging.init(); m_acceptor = new NettyAcceptor(); - m_acceptor.initialize(messaging); + m_acceptor.initialize(messaging, confParser.getProperties()); } public void stopServer() { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/ServerAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/ServerAcceptor.java index 7ce5e549..c0309985 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/ServerAcceptor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/ServerAcceptor.java @@ -1,6 +1,7 @@ package org.dna.mqtt.moquette.server; import java.io.IOException; +import java.util.Properties; import org.dna.mqtt.moquette.messaging.spi.IMessaging; /** @@ -9,7 +10,7 @@ import org.dna.mqtt.moquette.messaging.spi.IMessaging; */ public interface ServerAcceptor { - void initialize(IMessaging messaging) throws IOException; + void initialize(IMessaging messaging, Properties props) throws IOException; void close(); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java index da0cf1cd..f0ad2a2d 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java @@ -13,6 +13,7 @@ import io.netty.channel.socket.nio.NioServerSocketChannel; import java.io.IOException; import io.netty.handler.timeout.IdleStateHandler; +import java.util.Properties; import org.dna.mqtt.commons.Constants; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.parser.netty.MQTTDecoder; @@ -36,7 +37,7 @@ public class NettyAcceptor implements ServerAcceptor { MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector(); - public void initialize(IMessaging messaging) throws IOException { + public void initialize(IMessaging messaging, Properties props) throws IOException { m_bossGroup = new NioEventLoopGroup(); m_workerGroup = new NioEventLoopGroup(); @@ -65,7 +66,8 @@ public class NettyAcceptor implements ServerAcceptor { .childOption(ChannelOption.SO_KEEPALIVE, true); try { // Bind and start to accept incoming connections. - ChannelFuture f = b.bind(Constants.PORT); +// ChannelFuture f = b.bind(Constants.PORT); + ChannelFuture f = b.bind(props.getProperty("host"), Integer.parseInt(props.getProperty("port"))); LOG.info("Server binded"); f.sync(); } catch (InterruptedException ex) { diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ConfigurationParserTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ConfigurationParserTest.java new file mode 100644 index 00000000..29999015 --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ConfigurationParserTest.java @@ -0,0 +1,89 @@ +package org.dna.mqtt.moquette.server; + +import java.io.Reader; +import java.io.StringReader; +import java.text.ParseException; +import java.util.Properties; +import org.dna.mqtt.commons.Constants; +import org.junit.Test; +import static org.junit.Assert.*; +import org.junit.Before; + +/** + * + * @author andrea + */ +public class ConfigurationParserTest { + + ConfigurationParser m_parser; + + @Before + public void setUp() { + m_parser = new ConfigurationParser(); + } + + @Test + public void checkDefaultOptions() { + Properties props = m_parser.getProperties(); + + verifyDefaults(props); + } + + @Test + public void parseEmpty() throws ParseException { + Reader conf = new StringReader(" "); + m_parser.parse(conf); + + //Verify + verifyDefaults(m_parser.getProperties()); + } + + @Test + public void parseValidComment() throws ParseException { + Reader conf = new StringReader("#simple comment"); + m_parser.parse(conf); + + //Verify + verifyDefaults(m_parser.getProperties()); + } + + @Test(expected = ParseException.class) + public void parseInvalidComment() throws ParseException { + Reader conf = new StringReader(" #simple comment"); + m_parser.parse(conf); + } + + @Test + public void parseSingleVariable() throws ParseException { + Reader conf = new StringReader("port 1234"); + m_parser.parse(conf); + + //Verify + assertEquals("1234", m_parser.getProperties().getProperty("port")); + } + + @Test + public void parseCompleteFile() throws ParseException { + String content = "# This is initial config format \r\n" + + " \r\n" + + "port 1234 \r\n" + + "host localhost \r\n" + + "fake multi word string property\r\n"; + Reader conf = new StringReader(content); + m_parser.parse(conf); + + //Verify + Properties props = m_parser.getProperties(); + assertEquals("1234", props.getProperty("port")); + assertEquals("localhost", props.getProperty("host")); + assertEquals("multi word string property", props.getProperty("fake")); + } + + /** + * Helper method to verify default options. + */ + private void verifyDefaults(Properties props) { + assertEquals(Constants.PORT, Integer.parseInt(props.getProperty("port"))); + assertEquals(Constants.HOST, props.getProperty("host")); + } +} \ No newline at end of file diff --git a/distribution/src/main/assembly/assembly.xml b/distribution/src/main/assembly/assembly.xml index 6b87caf0..84544e9f 100644 --- a/distribution/src/main/assembly/assembly.xml +++ b/distribution/src/main/assembly/assembly.xml @@ -22,6 +22,11 @@ moquette-log.properties + + src/main/resources/moquette.conf + config/ + + src/main/resources/README.txt diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf new file mode 100644 index 00000000..00894f22 --- /dev/null +++ b/distribution/src/main/resources/moquette.conf @@ -0,0 +1,10 @@ +############################################## +# Moquette configuration file. +# +# The synthax is equals to mosquitto.conf +# +############################################## + +port 1883 + +host 0.0.0.0 \ No newline at end of file diff --git a/parser_commons/src/main/java/org/dna/mqtt/commons/Constants.java b/parser_commons/src/main/java/org/dna/mqtt/commons/Constants.java index 4b6f3259..5130dc22 100644 --- a/parser_commons/src/main/java/org/dna/mqtt/commons/Constants.java +++ b/parser_commons/src/main/java/org/dna/mqtt/commons/Constants.java @@ -5,5 +5,6 @@ package org.dna.mqtt.commons; */ public class Constants { public static final int PORT = 1883; + public static final String HOST = "0.0.0.0"; public static final int DEFAULT_CONNECT_TIMEOUT = 10; }