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

Added Codecs to handle subscription messages

Browse Source
This commit is contained in:
selva.andre
2012-07-27 00:16:32 +02:00
parent a65026dd03
commit 397aea8e6a
1 changed files with 17 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
+17 -2
View File
@@ -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) );