diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java index 66289cf3..cb74e75c 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -8,12 +8,17 @@ import org.apache.mina.core.session.IdleStatus; import org.apache.mina.filter.codec.ProtocolCodecFilter; import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder; import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder; +import org.apache.mina.filter.logging.LoggingFilter; import org.apache.mina.transport.socket.nio.NioSocketAcceptor; +import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; import org.dna.mqtt.moquette.proto.ConnAckEncoder; import org.dna.mqtt.moquette.proto.ConnectDecoder; import org.dna.mqtt.moquette.proto.DisconnectDecoder; import org.dna.mqtt.moquette.proto.PublishDecoder; +import org.dna.mqtt.moquette.proto.SubAckEncoder; +import org.dna.mqtt.moquette.proto.SubscribeDecoder; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +import org.dna.mqtt.moquette.proto.messages.SubAckMessage; /** * Launch a configured version of the server. * @author andrea @@ -34,18 +39,28 @@ public class Server { DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); decoder.addMessageDecoder(new ConnectDecoder()); decoder.addMessageDecoder(new PublishDecoder()); + decoder.addMessageDecoder(new SubscribeDecoder()); decoder.addMessageDecoder(new DisconnectDecoder()); DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); // encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder()); + encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder()); IoAcceptor acceptor = new NioSocketAcceptor(); -// acceptor.getFilterChain().addLast( "logger", new LoggingFilter("SERVER LOG") ); + acceptor.getFilterChain().addLast( "logger", new LoggingFilter("SERVER LOG") ); acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder)); - acceptor.setHandler( new MQTTHandler() ); + MQTTHandler handler = new MQTTHandler(); + SimpleMessaging messaging = new SimpleMessaging(); + //TODO fix this hugly wiring + handler.setMessaging(messaging); + messaging.setNotifier(handler); + + + acceptor.setHandler(handler); + ((NioSocketAcceptor)acceptor).getSessionConfig().setReuseAddress(true); acceptor.getSessionConfig().setReadBufferSize( 2048 ); acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, DEFAULT_CONNECT_TIMEOUT ); acceptor.bind( new InetSocketAddress(PORT) );