Introduction of mosquitto.conf parser, fix #29

This commit is contained in:
andrea
2014-02-22 11:14:56 +01:00
parent bbb9924c4b
commit 9624815c82
8 changed files with 223 additions and 4 deletions
@@ -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;
}
}
@@ -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() {
@@ -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();
}
@@ -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) {