Removed custom client side library
This commit is contained in:
selva.andre
@@ -1,348 +0,0 @@
|
||||
package org.dna.mqtt.moquette.server;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.dna.mqtt.moquette.client.Client;
|
||||
import org.dna.mqtt.moquette.client.IPublishCallback;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.dna.mqtt.commons.Constants;
|
||||
import org.junit.After;
|
||||
import static org.junit.Assert.*;
|
||||
import org.junit.Before;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class ServerIntegrationTest {
|
||||
|
||||
boolean received;
|
||||
Server server;
|
||||
|
||||
protected void startServer() throws IOException {
|
||||
server = new Server();
|
||||
server.startServer();
|
||||
}
|
||||
|
||||
@Before
|
||||
public void setUp() throws IOException {
|
||||
startServer();
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() {
|
||||
server.stopServer();
|
||||
File dbFile = new File(Server.STORAGE_FILE_PATH);
|
||||
if (dbFile.exists()) {
|
||||
dbFile.delete();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSubscribe() throws IOException, InterruptedException {
|
||||
// startServer();
|
||||
Client client = new Client("localhost", Constants.PORT);
|
||||
// Client client = new Client("test.mosquitto.org", 1883);
|
||||
client.connect();
|
||||
|
||||
|
||||
client.subscribe("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
}
|
||||
});
|
||||
|
||||
client.publish("/topic", "Test my payload".getBytes());
|
||||
|
||||
client.close();
|
||||
client.shutdown();
|
||||
|
||||
assertTrue(received);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testCleanSession_maintainClientSubscriptions() {
|
||||
Client client = new Client("localhost", Constants.PORT/*, "CLID_123"*/);
|
||||
client.connect(false); //without session cleanup
|
||||
|
||||
client.subscribe("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
}
|
||||
});
|
||||
client.close();
|
||||
|
||||
//Exercise that the client maintain the subscriptions
|
||||
client.connect(false);
|
||||
client.publish("/topic", "Test my payload".getBytes());
|
||||
client.close();
|
||||
|
||||
//Verify
|
||||
assertTrue(received);
|
||||
|
||||
//TearDown
|
||||
client.shutdown();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() {
|
||||
Client client = new Client("localhost", Constants.PORT, "CLID_123");
|
||||
client.connect(false); //without session cleanup
|
||||
|
||||
client.subscribe("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
}
|
||||
});
|
||||
client.close();
|
||||
client.shutdown();
|
||||
|
||||
//Exercise that the client maintain the subscriptions
|
||||
client = new Client("localhost", Constants.PORT, "CLID_123");
|
||||
client.register("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
}
|
||||
});
|
||||
client.connect(false);
|
||||
client.publish("/topic", "Test my payload".getBytes());
|
||||
client.close();
|
||||
|
||||
//Verify
|
||||
assertTrue(received);
|
||||
|
||||
//TearDown
|
||||
client.shutdown();
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Check that after a client has connected with clean session false, subscribed
|
||||
* to some topic and exited, if it reconnect with clean session true, the m_server
|
||||
* correctly cleanup every previous subscription
|
||||
*/
|
||||
@Test
|
||||
public void testCleanSession_correctlyClientSubscriptions() {
|
||||
Client client = new Client("localhost", Constants.PORT, "CLID_123");
|
||||
client.connect(false); //without session cleanup
|
||||
|
||||
client.subscribe("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
}
|
||||
});
|
||||
client.close();
|
||||
client.shutdown();
|
||||
|
||||
//Exercise that the client maintain the subscriptions
|
||||
client = new Client("localhost", Constants.PORT, "CLID_123");
|
||||
client.register("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
}
|
||||
});
|
||||
client.connect(true);
|
||||
client.publish("/topic", "Test my payload".getBytes());
|
||||
client.close();
|
||||
|
||||
//Verify
|
||||
assertFalse(received);
|
||||
|
||||
//TearDown
|
||||
client.shutdown();
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws IOException, InterruptedException {
|
||||
final CountDownLatch barrier = new CountDownLatch(1);
|
||||
Client client = new Client("localhost", Constants.PORT, "CLID_123");
|
||||
client.connect(false); //without session cleanup
|
||||
|
||||
client.subscribe("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
barrier.countDown();
|
||||
}
|
||||
});
|
||||
//never called because no notification is pending
|
||||
assertFalse(barrier.await(1, TimeUnit.SECONDS));
|
||||
|
||||
client.close();
|
||||
client.shutdown();
|
||||
server.stopServer();
|
||||
|
||||
server.startServer();
|
||||
|
||||
final CountDownLatch barrier2 = new CountDownLatch(1);
|
||||
|
||||
//Exercise that the client maintain the subscriptions
|
||||
client = new Client("localhost", Constants.PORT, "CLID_123");
|
||||
client.register("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
barrier2.countDown();
|
||||
}
|
||||
});
|
||||
client.connect(false);
|
||||
client.publish("/topic", "Test my payload".getBytes());
|
||||
client.close();
|
||||
|
||||
//Verify
|
||||
assertTrue(barrier2.await(1, TimeUnit.SECONDS));
|
||||
assertTrue(received);
|
||||
|
||||
//TearDown
|
||||
client.shutdown();
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testRetain_maintainMessage_againstClientDestruction() throws InterruptedException {
|
||||
final CountDownLatch barrier = new CountDownLatch(1);
|
||||
Client client = new Client("localhost", Constants.PORT, "CLID_123");
|
||||
client.connect();
|
||||
|
||||
client.publish("/topic", "Test my payload".getBytes(), true);
|
||||
client.close();
|
||||
client.shutdown();
|
||||
|
||||
//Exercise that the client maintain the subscriptions
|
||||
client = new Client("localhost", Constants.PORT, "CLID_123");
|
||||
client.connect();
|
||||
client.subscribe("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
barrier.countDown();
|
||||
}
|
||||
});
|
||||
client.close();
|
||||
|
||||
//Verify
|
||||
barrier.await(1, TimeUnit.SECONDS);
|
||||
assertTrue(received);
|
||||
|
||||
//TearDown
|
||||
client.shutdown();
|
||||
}
|
||||
|
||||
|
||||
@Ignore
|
||||
public void testUnsubscribe_do_not_notify_anymore_same_session() throws InterruptedException {
|
||||
Client client = new Client("localhost", Constants.PORT, "CLID_123");
|
||||
// Client client = new Client("test.mosquitto.org", 1883);
|
||||
client.connect();
|
||||
|
||||
final CountDownLatch barrier = new CountDownLatch(1);
|
||||
|
||||
client.subscribe("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
barrier.countDown();
|
||||
}
|
||||
});
|
||||
|
||||
client.publish("/topic", "Test my payload".getBytes());
|
||||
|
||||
//wait 1 second to receive the published message
|
||||
boolean unlocked = barrier.await(1000, TimeUnit.MILLISECONDS);
|
||||
assertTrue(unlocked); //we were unlocked by the message reception
|
||||
assertTrue(received);
|
||||
|
||||
//reinit the flag
|
||||
received = false;
|
||||
//unsubscrbe and republish to check no notification is raised up
|
||||
client.unsubscribe("/topic");
|
||||
client.publish("/topic", "Test my payload".getBytes());
|
||||
assertFalse(received);
|
||||
|
||||
client.close();
|
||||
client.shutdown();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testUnsubscribe_do_not_notify_anymore_new_session() throws InterruptedException {
|
||||
Client client = new Client("localhost", Constants.PORT, "CLID_123");
|
||||
// Client client = new Client("test.mosquitto.org", 1883);
|
||||
client.connect();
|
||||
|
||||
final CountDownLatch barrier = new CountDownLatch(1);
|
||||
|
||||
client.subscribe("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
barrier.countDown();
|
||||
}
|
||||
});
|
||||
|
||||
client.publish("/topic", "Test my payload".getBytes());
|
||||
|
||||
//wait 1 second to receive the published message
|
||||
boolean unlocked = barrier.await(1000, TimeUnit.MILLISECONDS);
|
||||
assertTrue(unlocked); //we were unlocked by the message reception
|
||||
assertTrue(received);
|
||||
|
||||
//unsubscrbe and republish to check no notification is raised up
|
||||
client.unsubscribe("/topic");
|
||||
|
||||
client.close();
|
||||
client.shutdown();
|
||||
|
||||
//Exercise that the client maintain the subscriptions
|
||||
client = new Client("localhost", Constants.PORT, "CLID_123");
|
||||
client.connect();
|
||||
|
||||
//reinit the flag
|
||||
received = false;
|
||||
final CountDownLatch barrier2 = new CountDownLatch(1);
|
||||
|
||||
client.publish("/topic", "Test my payload".getBytes());
|
||||
unlocked = barrier2.await(1000, TimeUnit.MILLISECONDS);
|
||||
assertFalse(unlocked); //we were unlocked by the timeout exipration, no message received
|
||||
assertFalse(received);
|
||||
|
||||
client.close();
|
||||
client.shutdown();
|
||||
}
|
||||
|
||||
/*-------------- QoS 1 tests ----*/
|
||||
@Test
|
||||
public void testPublishWithQoS1() throws IOException, InterruptedException {
|
||||
// startServer();
|
||||
Client client = new Client("localhost", Constants.PORT);
|
||||
// Client client = new Client("test.mosquitto.org", 1883);
|
||||
client.connect();
|
||||
|
||||
|
||||
client.subscribe("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
}
|
||||
});
|
||||
|
||||
client.publish("/topic", "Test my payload".getBytes(), AbstractMessage.QOSType.LEAST_ONE, false);
|
||||
|
||||
client.close();
|
||||
client.shutdown();
|
||||
|
||||
assertTrue(received);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,98 +0,0 @@
|
||||
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
<artifactId>moquette-parent</artifactId>
|
||||
<groupId>org.dna.mqtt</groupId>
|
||||
<version>0.1</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>moquette-client</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<version>0.1</version>
|
||||
<name>Moquette - client</name>
|
||||
|
||||
<properties>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.dna.mqtt</groupId>
|
||||
<artifactId>moquette-parser</artifactId>
|
||||
<version>0.1</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.apache.mina</groupId>
|
||||
<artifactId>mina-core</artifactId>
|
||||
<version>2.0.4</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-log4j12</artifactId>
|
||||
<version>1.6.4</version>
|
||||
</dependency>
|
||||
|
||||
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<version>4.8.2</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<!-- needs extra dependencies: objenesis & hamcrest -->
|
||||
<dependency>
|
||||
<groupId>org.mockito</groupId>
|
||||
<artifactId>mockito-core</artifactId>
|
||||
<version>1.9.0</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<configuration>
|
||||
<source>1.5</source>
|
||||
<target>1.5</target>
|
||||
</configuration>
|
||||
</plugin>
|
||||
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-surefire-plugin</artifactId>
|
||||
<configuration>
|
||||
<excludes>
|
||||
<exclude>org/dna/mqtt/moquette/proto/TestUtils.java</exclude>
|
||||
<exclude>org/dna/mqtt/moquette/proto/TestUtils$*</exclude>
|
||||
</excludes>
|
||||
</configuration>
|
||||
</plugin>
|
||||
|
||||
|
||||
<plugin>
|
||||
<artifactId>maven-assembly-plugin</artifactId>
|
||||
<configuration>
|
||||
<archive>
|
||||
<manifest>
|
||||
<mainClass>org.dna.mqtt.moquette.client.Main</mainClass>
|
||||
</manifest>
|
||||
</archive>
|
||||
<descriptorRefs>
|
||||
<descriptorRef>jar-with-dependencies</descriptorRef>
|
||||
</descriptorRefs>
|
||||
</configuration>
|
||||
</plugin>
|
||||
|
||||
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
@@ -1,463 +0,0 @@
|
||||
package org.dna.mqtt.moquette.client;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.NetworkInterface;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.apache.mina.core.RuntimeIoException;
|
||||
import org.apache.mina.core.future.ConnectFuture;
|
||||
import org.apache.mina.core.future.WriteFuture;
|
||||
import org.apache.mina.core.service.IoConnector;
|
||||
import org.apache.mina.core.session.IdleStatus;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
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.transport.socket.nio.NioSocketConnector;
|
||||
import org.dna.mqtt.commons.Constants;
|
||||
import org.dna.mqtt.commons.MessageIDGenerator;
|
||||
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.*;
|
||||
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.MessageIDMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.PingReqMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* The API to connect to a MQTT server.
|
||||
*
|
||||
* Pass to the constructor the host and port to connect, invoke the connect
|
||||
* message, invoke publish to send notifications messages, invoke subscribe with
|
||||
* a callback to be notified when something happen on the desired topic.
|
||||
*
|
||||
* NB this class is not thread safe so MUST be used by only one thread at time.
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public final class Client {
|
||||
|
||||
final static int DEFAULT_RETRIES = 3;
|
||||
final static int RETRIES_QOS_GT0 = 3;
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Client.class);
|
||||
// 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; // 3 seconds
|
||||
private static final long SUBACK_TIMEOUT = 4 * 1000L;
|
||||
private static final int KEEPALIVE_SECS = 3;
|
||||
private static final int NUM_SCHEDULER_TIMER_THREAD = 1;
|
||||
private int m_connectRetries = DEFAULT_RETRIES;
|
||||
private String m_hostname;
|
||||
private int m_port;
|
||||
//internal management used for conversation with the server
|
||||
private IoConnector m_connector;
|
||||
private IoSession m_session;
|
||||
private CountDownLatch m_connectBarrier;
|
||||
private CountDownLatch m_subscribeBarrier;
|
||||
private int m_receivedSubAckMessageID;
|
||||
private byte m_returnCode;
|
||||
//TODO synchronize the access
|
||||
//Refact the da model should be a list of callback for each topic
|
||||
private Map<String, IPublishCallback> m_subscribersList = new HashMap<String, IPublishCallback>();
|
||||
private ScheduledExecutorService m_scheduler;
|
||||
private ScheduledFuture m_pingerHandler;
|
||||
private String m_macAddress;
|
||||
private MessageIDGenerator m_messageIDGenerator = new MessageIDGenerator();
|
||||
/**Maintain the list of in flight messages for QoS > 0*/
|
||||
// private Set<Integer> m_inflightIDs = new HashSet<Integer>();
|
||||
|
||||
private String m_clientID;
|
||||
|
||||
final Runnable pingerDeamon = new Runnable() {
|
||||
public void run() {
|
||||
LOG.debug("Pingreq sent");
|
||||
//send a ping req
|
||||
m_session.write(new PingReqMessage());
|
||||
}
|
||||
};
|
||||
|
||||
public Client(String host, int port) {
|
||||
m_hostname = host;
|
||||
m_port = port;
|
||||
init();
|
||||
}
|
||||
|
||||
/**
|
||||
* Constructor in which the user could provide it's own ClientID
|
||||
*/
|
||||
public Client(String host, int port, String clientID) {
|
||||
this(host, port);
|
||||
m_clientID = clientID;
|
||||
}
|
||||
|
||||
protected void init() {
|
||||
DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
|
||||
decoder.addMessageDecoder(new ConnAckDecoder());
|
||||
decoder.addMessageDecoder(new SubAckDecoder());
|
||||
decoder.addMessageDecoder(new UnsubAckDecoder());
|
||||
decoder.addMessageDecoder(new PublishDecoder());
|
||||
decoder.addMessageDecoder(new PubAckDecoder());
|
||||
decoder.addMessageDecoder(new PingRespDecoder());
|
||||
|
||||
DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
|
||||
encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
|
||||
encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder());
|
||||
encoder.addMessageEncoder(SubscribeMessage.class, new SubscribeEncoder());
|
||||
encoder.addMessageEncoder(UnsubscribeMessage.class, new UnsubscribeEncoder());
|
||||
encoder.addMessageEncoder(DisconnectMessage.class, new DisconnectEncoder());
|
||||
encoder.addMessageEncoder(PingReqMessage.class, new PingReqEncoder());
|
||||
|
||||
m_connector = new NioSocketConnector();
|
||||
|
||||
// m_connector.getFilterChain().addLast("logger", new LoggingFilter());
|
||||
m_connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(encoder, decoder));
|
||||
|
||||
m_connector.setHandler(new ClientMQTTHandler(this));
|
||||
m_connector.getSessionConfig().setReadBufferSize(2048);
|
||||
m_connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, Constants.DEFAULT_CONNECT_TIMEOUT);
|
||||
|
||||
m_scheduler = Executors.newScheduledThreadPool(NUM_SCHEDULER_TIMER_THREAD);
|
||||
|
||||
m_macAddress = readMACAddress();
|
||||
}
|
||||
|
||||
/**
|
||||
* Connects to the server with clean session set to true, do not maintains
|
||||
* client topic subscriptions
|
||||
*/
|
||||
public void connect() throws MQTTException {
|
||||
connect(true);
|
||||
}
|
||||
|
||||
|
||||
public void connect(boolean cleanSession) throws MQTTException {
|
||||
int retries = 0;
|
||||
|
||||
try {
|
||||
ConnectFuture future = m_connector.connect(new InetSocketAddress(m_hostname, m_port));
|
||||
LOG.debug("Client waiting to connect to server");
|
||||
future.awaitUninterruptibly();
|
||||
m_session = future.getSession();
|
||||
} catch (RuntimeIoException e) {
|
||||
LOG.debug("Failed to connect, retry " + retries + " of (" + m_connectRetries + ")", e);
|
||||
}
|
||||
|
||||
if (retries == m_connectRetries) {
|
||||
throw new MQTTException("Can't connect to the server after " + retries + "retries");
|
||||
}
|
||||
|
||||
m_connectBarrier = new CountDownLatch(1);
|
||||
|
||||
//send a message over the session
|
||||
ConnectMessage connMsg = new ConnectMessage();
|
||||
connMsg.setKeepAlive(KEEPALIVE_SECS);
|
||||
if (m_clientID == null) {
|
||||
m_clientID = generateClientID();
|
||||
}
|
||||
connMsg.setClientID(m_clientID);
|
||||
connMsg.setCleanSession(cleanSession);
|
||||
|
||||
m_session.write(connMsg);
|
||||
|
||||
//suspend until the server respond with CONN_ACK
|
||||
boolean unlocked = false;
|
||||
try {
|
||||
unlocked = m_connectBarrier.await(CONNECT_TIMEOUT, TimeUnit.MILLISECONDS); //TODO parametrize
|
||||
} catch (InterruptedException ex) {
|
||||
throw new ConnectionException(ex);
|
||||
}
|
||||
|
||||
//if not arrive into certain limit, raise an error
|
||||
if (!unlocked) {
|
||||
throw new ConnectionException("Connection timeout elapsed unless server responded with a CONN_ACK");
|
||||
}
|
||||
|
||||
//also raise an error when CONN_ACK is received with some error code inside
|
||||
if (m_returnCode != ConnAckMessage.CONNECTION_ACCEPTED) {
|
||||
String errMsg;
|
||||
switch (m_returnCode) {
|
||||
case ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION:
|
||||
errMsg = "Unacceptable protocol version";
|
||||
break;
|
||||
case ConnAckMessage.IDENTIFIER_REJECTED:
|
||||
errMsg = "Identifier rejected";
|
||||
break;
|
||||
case ConnAckMessage.SERVER_UNAVAILABLE:
|
||||
errMsg = "Server unavailable";
|
||||
break;
|
||||
case ConnAckMessage.BAD_USERNAME_OR_PASSWORD:
|
||||
errMsg = "Bad username or password";
|
||||
break;
|
||||
case ConnAckMessage.NOT_AUTHORIZED:
|
||||
errMsg = "Not authorized";
|
||||
break;
|
||||
default:
|
||||
errMsg = "Not idetified erro code " + m_returnCode;
|
||||
}
|
||||
throw new ConnectionException(errMsg);
|
||||
}
|
||||
|
||||
updatePinger();
|
||||
}
|
||||
|
||||
/**
|
||||
* Publish to the connected server the payload message to the given topic.
|
||||
* It's admitted to publish a 0 -length payload.
|
||||
*/
|
||||
public void publish(String topic, byte[] payload) throws PublishException {
|
||||
publish(topic, payload, false);
|
||||
}
|
||||
|
||||
/**
|
||||
* Publish by default with QoS 0
|
||||
* */
|
||||
public void publish(String topic, byte[] payload, boolean retain) throws PublishException {
|
||||
publish(topic, payload, AbstractMessage.QOSType.MOST_ONE, retain);
|
||||
}
|
||||
|
||||
public void publish(String topic, byte[] payload, AbstractMessage.QOSType qos, boolean retain) throws PublishException {
|
||||
PublishMessage msg = new PublishMessage();
|
||||
msg.setRetainFlag(retain);
|
||||
msg.setTopicName(topic);
|
||||
msg.setPayload(payload);
|
||||
|
||||
//Untill the server could handle all the Qos 2 level
|
||||
if (qos != AbstractMessage.QOSType.MOST_ONE) {
|
||||
msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
|
||||
int messageID = m_messageIDGenerator.next();
|
||||
msg.setMessageID(messageID);
|
||||
|
||||
try {
|
||||
manageSendQoS1(msg);
|
||||
} catch (Throwable ex) {
|
||||
throw new MQTTException(ex);
|
||||
}
|
||||
} else {
|
||||
//QoS 0 case
|
||||
msg.setQos(AbstractMessage.QOSType.MOST_ONE);
|
||||
WriteFuture wf = m_session.write(msg);
|
||||
try {
|
||||
wf.await();
|
||||
} catch (InterruptedException ex) {
|
||||
LOG.debug(null, ex);
|
||||
throw new PublishException(ex);
|
||||
}
|
||||
|
||||
Throwable ex = wf.getException();
|
||||
if (ex != null) {
|
||||
throw new PublishException(ex);
|
||||
}
|
||||
}
|
||||
|
||||
updatePinger();
|
||||
}
|
||||
|
||||
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);
|
||||
int messageID = m_messageIDGenerator.next();
|
||||
msg.setMessageID(messageID);
|
||||
// m_inflightIDs.add(messageID);
|
||||
register(topic, publishCallback);
|
||||
|
||||
try {
|
||||
manageSendQoS1(msg);
|
||||
} catch(Throwable ex) {
|
||||
//in case errors arise, remove the registration because the subscription
|
||||
// hasn't get well
|
||||
unregister(topic);
|
||||
throw new MQTTException(ex);
|
||||
}
|
||||
|
||||
updatePinger();
|
||||
}
|
||||
|
||||
|
||||
public void unsubscribe(String... topics) {
|
||||
LOG.info("unsubscribe invoked");
|
||||
UnsubscribeMessage msg = new UnsubscribeMessage();
|
||||
for (String topic : topics) {
|
||||
msg.addTopic(topic);
|
||||
}
|
||||
msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
|
||||
int messageID = m_messageIDGenerator.next();
|
||||
msg.setMessageID(messageID);
|
||||
// m_inflightIDs.add(messageID);
|
||||
// register(topic, publishCallback);
|
||||
try {
|
||||
manageSendQoS1(msg);
|
||||
} catch(Throwable ex) {
|
||||
//in case errors arise, remove the registration because the subscription
|
||||
// hasn't get well
|
||||
// unregister(topic);
|
||||
throw new MQTTException(ex);
|
||||
}
|
||||
|
||||
for (String topic : topics) {
|
||||
unregister(topic);
|
||||
}
|
||||
|
||||
// register(topic, publishCallback);
|
||||
updatePinger();
|
||||
}
|
||||
|
||||
private void manageSendQoS1(MessageIDMessage msg) throws Throwable{
|
||||
int messageID = msg.getMessageID();
|
||||
boolean unlocked = false;
|
||||
for (int retries = 0; retries < RETRIES_QOS_GT0 || !unlocked; retries++) {
|
||||
LOG.debug("manageSendQoS1 retry " + retries);
|
||||
if (retries > 0) {
|
||||
msg.setDupFlag(true);
|
||||
}
|
||||
|
||||
WriteFuture wf = m_session.write(msg);
|
||||
wf.await();
|
||||
LOG.info("message sent");
|
||||
|
||||
Throwable ex = wf.getException();
|
||||
if (ex != null) {
|
||||
throw ex;
|
||||
}
|
||||
|
||||
//wait for the SubAck
|
||||
m_subscribeBarrier = new CountDownLatch(1);
|
||||
|
||||
//suspend until the server respond with CONN_ACK
|
||||
LOG.info("subscribe waiting for suback");
|
||||
unlocked = m_subscribeBarrier.await(SUBACK_TIMEOUT, TimeUnit.MILLISECONDS); //TODO parametrize
|
||||
}
|
||||
|
||||
//if not arrive into certain limit, raise an error
|
||||
if (!unlocked) {
|
||||
throw new SubscribeException(String.format("Server doesn't replyed with a SUB_ACK after %d replies", RETRIES_QOS_GT0));
|
||||
} else {
|
||||
//check if message ID match
|
||||
if (m_receivedSubAckMessageID != messageID) {
|
||||
throw new SubscribeException(String.format("Server replyed with "
|
||||
+ "a broken MessageID in SUB_ACK, expected %d but received %d",
|
||||
messageID, m_receivedSubAckMessageID));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* TODO extract this SPI method in a SPI
|
||||
*/
|
||||
protected void connectionAckCallback(byte returnCode) {
|
||||
LOG.info("connectionAckCallback invoked");
|
||||
m_returnCode = returnCode;
|
||||
m_connectBarrier.countDown();
|
||||
}
|
||||
|
||||
protected void subscribeAckCallback(int messageID) {
|
||||
LOG.info("subscribeAckCallback invoked");
|
||||
m_subscribeBarrier.countDown();
|
||||
m_receivedSubAckMessageID = messageID;
|
||||
}
|
||||
|
||||
void unsubscribeAckCallback(int messageID) {
|
||||
LOG.info("unsubscribeAckCallback invoked");
|
||||
//NB we share the barrier because in futur will be a single barrier for all
|
||||
m_subscribeBarrier.countDown();
|
||||
m_receivedSubAckMessageID = messageID;
|
||||
}
|
||||
|
||||
void publishAckCallback(Integer messageID) {
|
||||
LOG.info("publishAckCallback invoked");
|
||||
m_subscribeBarrier.countDown();
|
||||
m_receivedSubAckMessageID = messageID;
|
||||
}
|
||||
|
||||
protected void publishCallback(String topic, byte[] payload) {
|
||||
IPublishCallback callback = m_subscribersList.get(topic);
|
||||
if (callback == null) {
|
||||
String msg = String.format("Can't find any publish callback fr topic %s", topic);
|
||||
LOG.error(msg);
|
||||
throw new MQTTException(msg);
|
||||
}
|
||||
callback.published(topic, payload);
|
||||
}
|
||||
|
||||
/**
|
||||
* In the current pinger is not ye executed, then cancel it and schedule
|
||||
* another by KEEPALIVE_SECS
|
||||
*/
|
||||
private void updatePinger() {
|
||||
if (m_pingerHandler != null) {
|
||||
m_pingerHandler.cancel(false);
|
||||
}
|
||||
m_pingerHandler = m_scheduler.scheduleWithFixedDelay(pingerDeamon, KEEPALIVE_SECS, KEEPALIVE_SECS, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
private String readMACAddress() {
|
||||
try {
|
||||
NetworkInterface network = NetworkInterface.getNetworkInterfaces().nextElement();
|
||||
|
||||
byte[] mac = network.getHardwareAddress();
|
||||
|
||||
StringBuilder sb = new StringBuilder();
|
||||
for (int i = 0; i < mac.length; i++) {
|
||||
sb.append(String.format("%02X%s", mac[i], ""));
|
||||
}
|
||||
return sb.toString();
|
||||
} catch (Exception ex) {
|
||||
throw new MQTTException("Can't retrieve host MAC address", ex);
|
||||
}
|
||||
}
|
||||
|
||||
private String generateClientID() {
|
||||
double rnd = Math.random();
|
||||
String id = "Moque" + m_macAddress + Math.round(rnd*1000);
|
||||
LOG.debug("Generated ClientID " + id);
|
||||
return id;
|
||||
}
|
||||
|
||||
public void close() {
|
||||
//stop the pinger
|
||||
m_pingerHandler.cancel(false);
|
||||
|
||||
//send the CLOSE message
|
||||
m_session.write(new DisconnectMessage());
|
||||
|
||||
// wait until the summation is done
|
||||
m_session.getCloseFuture().awaitUninterruptibly();
|
||||
// m_connector.dispose();
|
||||
}
|
||||
|
||||
|
||||
public void shutdown() {
|
||||
m_connector.dispose();
|
||||
}
|
||||
|
||||
/**
|
||||
* Used only to re-register the callback with the topic, not to send any
|
||||
* subscription message to the server, used when a client reconnect to an
|
||||
* existing session on the server
|
||||
*/
|
||||
public void register(String topic, IPublishCallback publishCallback) {
|
||||
//register the publishCallback in some registry to be notified
|
||||
m_subscribersList.put(topic, publishCallback);
|
||||
}
|
||||
|
||||
|
||||
//Remove the registration of the callback from the topic
|
||||
private void unregister(String topic) {
|
||||
m_subscribersList.remove(topic);
|
||||
}
|
||||
}
|
||||
@@ -1,73 +0,0 @@
|
||||
package org.dna.mqtt.moquette.client;
|
||||
|
||||
import org.apache.mina.core.service.IoHandlerAdapter;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.dna.mqtt.moquette.proto.messages.*;
|
||||
|
||||
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class ClientMQTTHandler extends IoHandlerAdapter {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(ClientMQTTHandler.class);
|
||||
|
||||
Client m_callback;
|
||||
|
||||
ClientMQTTHandler(Client callback) {
|
||||
m_callback = callback;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void messageReceived(IoSession session, Object message) throws Exception {
|
||||
AbstractMessage msg = (AbstractMessage) message;
|
||||
LOG.debug("Received a message of type " + msg.getMessageType());
|
||||
switch (msg.getMessageType()) {
|
||||
case CONNACK:
|
||||
handleConnectAck(session, (ConnAckMessage) msg);
|
||||
break;
|
||||
case SUBACK:
|
||||
handleSubscribeAck(session, (SubAckMessage) msg);
|
||||
break;
|
||||
case UNSUBACK:
|
||||
handleUnsubscribeAck(session, (UnsubAckMessage) msg);
|
||||
break;
|
||||
// case SUBSCRIBE:
|
||||
// handleSubscribe(session, (SubscribeMessage) msg);
|
||||
// break;
|
||||
case PUBLISH:
|
||||
handlePublish(session, (PublishMessage) msg);
|
||||
break;
|
||||
case PUBACK:
|
||||
handlePublishAck(session, (PubAckMessage) msg);
|
||||
break;
|
||||
case PINGRESP:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private void handlePublishAck(IoSession session, PubAckMessage msg) {
|
||||
m_callback.publishAckCallback(msg.getMessageID());
|
||||
}
|
||||
|
||||
private void handleConnectAck(IoSession session, ConnAckMessage connAckMessage) {
|
||||
m_callback.connectionAckCallback(connAckMessage.getReturnCode());
|
||||
}
|
||||
|
||||
private void handleSubscribeAck(IoSession session, SubAckMessage subAckMessage) {
|
||||
m_callback.subscribeAckCallback(subAckMessage.getMessageID());
|
||||
}
|
||||
|
||||
private void handlePublish(IoSession session, PublishMessage pubMessage) {
|
||||
m_callback.publishCallback(pubMessage.getTopicName(), pubMessage.getPayload());
|
||||
}
|
||||
|
||||
private void handleUnsubscribeAck(IoSession session, UnsubAckMessage unsubAckMessage) {
|
||||
m_callback.unsubscribeAckCallback(unsubAckMessage.getMessageID());
|
||||
}
|
||||
}
|
||||
@@ -1,10 +0,0 @@
|
||||
package org.dna.mqtt.moquette.client;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public interface IPublishCallback {
|
||||
|
||||
void published(String topic, byte[] message/*, boolean retained*/);
|
||||
}
|
||||
@@ -1,34 +0,0 @@
|
||||
package org.dna.mqtt.moquette.client;
|
||||
|
||||
import java.io.IOException;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class Main {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Main.class);
|
||||
|
||||
static boolean received;
|
||||
|
||||
public static void main(String[] args) throws IOException, InterruptedException {
|
||||
Client client = new Client("test.mosquitto.org", 1883);
|
||||
client.connect();
|
||||
|
||||
client.subscribe("/topic", new IPublishCallback() {
|
||||
|
||||
public void published(String topic, byte[] message) {
|
||||
received = true;
|
||||
}
|
||||
});
|
||||
|
||||
Thread.sleep(5000);
|
||||
|
||||
client.publish("/topic", "Test my payload".getBytes());
|
||||
|
||||
client.close();
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user