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

Added skeleton for client side of subscription

Browse Source
This commit is contained in:
selva.andre
2012-07-27 00:17:43 +02:00
parent 397aea8e6a
commit b32bbe4629
7 changed files with 114 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/SubscribeException.java
+16
View File
@@ -0,0 +1,16 @@
package org.dna.mqtt.moquette;
/**
*
* @author andrea
*/
public class SubscribeException extends ConnectionException {
public SubscribeException(String msg) {
super(msg);
}
public SubscribeException(Throwable e) {
super(e);
}
}
parser/src/main/java/org/dna/mqtt/moquette/client/Client.java
+59 -2
View File
@@ -19,15 +19,19 @@ import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.dna.mqtt.moquette.ConnectionException;
import org.dna.mqtt.moquette.MQTTException;
import org.dna.mqtt.moquette.PublishException;
import org.dna.mqtt.moquette.SubscribeException;
import org.dna.mqtt.moquette.proto.ConnAckDecoder;
import org.dna.mqtt.moquette.proto.ConnectEncoder;
import org.dna.mqtt.moquette.proto.DisconnectEncoder;
import org.dna.mqtt.moquette.proto.PublishEncoder;
import org.dna.mqtt.moquette.proto.SubAckDecoder;
import org.dna.mqtt.moquette.proto.SubscribeEncoder;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.dna.mqtt.moquette.proto.messages.DisconnectMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.dna.mqtt.moquette.server.Server;
/**
@@ -44,7 +48,8 @@ public final class Client {
private static final Logger LOG = Logger.getLogger(Client.class.getName());
// private static final String HOSTNAME = /*"localhost"*/ "127.0.0.1";
// private static final int PORT = Server.PORT;
private static final long CONNECT_TIMEOUT = 3 * 1000L; // 30 seconds
private static final long CONNECT_TIMEOUT = 3 * 1000L; // 3 seconds
private static final long SUBACK_TIMEOUT = 4 * 1000L;
final static int DEFAULT_RETRIES = 3;
private int m_connectRetries = DEFAULT_RETRIES;
@@ -55,6 +60,7 @@ public final class Client {
private IoConnector m_connector;
private IoSession m_session;
private CountDownLatch m_connectBarrier;
private CountDownLatch m_subscribeBarrier;
private byte m_returnCode;
public Client(String host, int port) {
@@ -66,15 +72,17 @@ public final class Client {
protected void init() {
DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
decoder.addMessageDecoder(new ConnAckDecoder());
decoder.addMessageDecoder(new SubAckDecoder());
DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder());
encoder.addMessageEncoder(SubscribeMessage.class, new SubscribeEncoder());
encoder.addMessageEncoder(DisconnectMessage.class, new DisconnectEncoder());
m_connector = new NioSocketConnector();
m_connector.getFilterChain().addLast("logger", new LoggingFilter());
// m_connector.getFilterChain().addLast("logger", new LoggingFilter());
m_connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(encoder, decoder));
m_connector.setHandler(new ClientMQTTHandler(this));
@@ -180,6 +188,48 @@ public final class Client {
}
}
public void subscribe(String topic, IPublishCallback publishCallback) {
LOG.info("subscribe invoked");
SubscribeMessage msg = new SubscribeMessage();
msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), topic));
msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
WriteFuture wf = m_session.write(msg);
try {
wf.await();
} catch (InterruptedException ex) {
LOG.log(Level.SEVERE, null, ex);
throw new SubscribeException(ex);
}
LOG.info("subscribe message sent");
Throwable ex = wf.getException();
if (ex != null) {
throw new SubscribeException(ex);
}
//TODO register the publishCallback in some registry to be notified
//wait for the SubAck
m_subscribeBarrier = new CountDownLatch(1);
//suspend until the server respond with CONN_ACK
boolean unlocked = false;
try {
LOG.info("subscribe waiting for suback");
unlocked = m_subscribeBarrier.await(SUBACK_TIMEOUT, TimeUnit.MILLISECONDS); //TODO parametrize
} catch (InterruptedException iex) {
throw new SubscribeException(iex);
}
//if not arrive into certain limit, raise an error
if (!unlocked) {
throw new SubscribeException("Subscribe timeout elapsed unless server responded with a SUB_ACK");
}
//TODO check the ACK messageID
}
/**
* TODO extract this SPI method in a SPI
@@ -190,6 +240,13 @@ public final class Client {
m_connectBarrier.countDown();
}
protected void subscribeAckCallback() {
LOG.info("subscribeAckCallback invoked");
m_subscribeBarrier.countDown();
}
public void close() {
//send the CLOSE message
m_session.write(new DisconnectMessage());
parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java
+8
View File
@@ -6,6 +6,7 @@ import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
/**
*
@@ -29,6 +30,9 @@ public class ClientMQTTHandler extends IoHandlerAdapter {
case CONNACK:
handleConnectAck(session, (ConnAckMessage) msg);
break;
case SUBACK:
handleSubscribeAck(session, (SubAckMessage) msg);
break;
// case SUBSCRIBE:
// handleSubscribe(session, (SubscribeMessage) msg);
// break;
@@ -41,4 +45,8 @@ public class ClientMQTTHandler extends IoHandlerAdapter {
private void handleConnectAck(IoSession session, ConnAckMessage connAckMessage) {
m_callback.connectionAckCallback(connAckMessage.getReturnCode());
}
private void handleSubscribeAck(IoSession session, SubAckMessage subAckMessage) {
m_callback.subscribeAckCallback();
}
}
parser/src/main/java/org/dna/mqtt/moquette/client/IPublishCallback.java
+10
View File
@@ -0,0 +1,10 @@
package org.dna.mqtt.moquette.client;
/**
*
* @author andrea
*/
public interface IPublishCallback {
void published(String topic, byte[] message);
}
parser/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java
+5
View File
@@ -29,6 +29,11 @@ public class SubscribeMessage extends MessageIDMessage {
}
private List<Couple> m_subscriptions = new ArrayList<Couple>();
public SubscribeMessage() {
//Subscribe has always QoS 1
m_qos = AbstractMessage.QOSType.LEAST_ONE;
}
public List<Couple> subscriptions() {
return m_subscriptions;
}
parser/src/main/resources/log4j.properties
+9
View File
@@ -0,0 +1,9 @@
# Set root logger level to DEBUG and its only appender to A1.
log4j.rootLogger=DEBUG, A1
# A1 is set to be a ConsoleAppender.
log4j.appender.A1=org.apache.log4j.ConsoleAppender
# A1 uses PatternLayout.
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n
parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java
+7 -1
View File
@@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.server;
import java.io.IOException;
import org.apache.mina.core.service.IoConnector;
import org.dna.mqtt.moquette.client.Client;
import org.dna.mqtt.moquette.client.IPublishCallback;
import org.junit.Test;
/**
@@ -24,7 +25,12 @@ public class ServerIntegrationTest {
client.connect();
client.publish("/topic", "Test my payload".getBytes());
//TODO client.subscribe("/topic", callback)
client.subscribe("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
throw new UnsupportedOperationException("Not supported yet.");
}
});
client.close();
}