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

Added Client ID generator

Browse Source
This commit is contained in:
selva.andre
2012-08-09 23:14:29 +02:00
parent 937a7b7060
commit 5540805570
1 changed files with 75 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/client/Client.java
+75 -57
View File
@@ -1,6 +1,8 @@
package org.dna.mqtt.moquette.client;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
@@ -44,11 +46,11 @@ 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.
*
*
* 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.
*
* @author andrea
*/
public final class Client {
@@ -62,29 +64,27 @@ public final class Client {
private static final int NUM_SCHEDULER_TIMER_THREAD = 1;
final static int DEFAULT_RETRIES = 3;
private int m_connectRetries = DEFAULT_RETRIES;
private String m_hostname;
private int m_port;
//internal mangement used for conversation with the server
private IoConnector m_connector;
private IoSession m_session;
private CountDownLatch m_connectBarrier;
private CountDownLatch m_subscribeBarrier;
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;
final Runnable pingerDeamon = new Runnable() {
public void run() {
LOG.debug("Pingreq sent");
//send a ping req
m_session.write(new PingReqMessage());
}
};
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;
@@ -105,7 +105,7 @@ public final class Client {
encoder.addMessageEncoder(SubscribeMessage.class, new SubscribeEncoder());
encoder.addMessageEncoder(DisconnectMessage.class, new DisconnectEncoder());
encoder.addMessageEncoder(PingReqMessage.class, new PingReqEncoder());
m_connector = new NioSocketConnector();
// m_connector.getFilterChain().addLast("logger", new LoggingFilter());
@@ -114,47 +114,49 @@ public final class Client {
m_connector.setHandler(new ClientMQTTHandler(this));
m_connector.getSessionConfig().setReadBufferSize(2048);
m_connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, Server.DEFAULT_CONNECT_TIMEOUT);
m_scheduler = Executors.newScheduledThreadPool(NUM_SCHEDULER_TIMER_THREAD);
m_macAddress = readMACAddress();
}
public void connect() 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);
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);
connMsg.setClientID("FAKE_MSG_ID1"); //TODO create a generator for the ID
connMsg.setClientID(generateClientID());
m_session.write(connMsg);
//suspend until the server respond with CONN_ACK
boolean unlocked = false;
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;
@@ -179,22 +181,22 @@ public final class Client {
}
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{
public void publish(String topic, byte[] payload) throws PublishException {
PublishMessage msg = new PublishMessage();
msg.setTopicName(topic);
msg.setPayload(payload);
//Untill the server could handle all the Qos levels
msg.setQos(AbstractMessage.QOSType.MOST_ONE);
WriteFuture wf = m_session.write(msg);
try {
wf.await();
@@ -202,21 +204,21 @@ public final class Client {
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.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();
@@ -225,61 +227,57 @@ public final class Client {
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
m_subscribersList.put(topic, publishCallback);
//wait for the SubAck
m_subscribeBarrier = new CountDownLatch(1);
//suspend until the server respond with CONN_ACK
boolean unlocked = false;
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");
}
updatePinger();
//TODO check the ACK messageID
}
/**
* TODO extract this SPI method in a SPI
* 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() {
LOG.info("subscribeAckCallback invoked");
m_subscribeBarrier.countDown();
}
protected void publishCallback(String topic, byte[] payload) {
m_subscribersList.get(topic).published(topic, payload);
}
/**
* In the current pinger is not ye executed, then cancel it and schedule
* In the current pinger is not ye executed, then cancel it and schedule
* another by KEEPALIVE_SECS
*/
private void updatePinger() {
@@ -288,15 +286,35 @@ public final class Client {
}
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], (i < mac.length - 1) ? "-" : ""));
}
return sb.toString();
} catch (Exception ex) {
throw new MQTTException("Can't retrieve host MAC address", ex);
}
}
private String generateClientID() {
double rnd = Math.random();
return "Moquette-client_" + m_macAddress + Math.round(rnd*1000);
}
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();