Modified Client to:
- accept a user provided clientID in the constructor - handle the cleanSession in connect method - let the user register a collback to a topic without do any subscription on server side - divide the closing from a server from the shutdown of all the client instance
This commit is contained in:
@@ -78,6 +78,9 @@ public final class Client {
|
|||||||
private ScheduledExecutorService m_scheduler;
|
private ScheduledExecutorService m_scheduler;
|
||||||
private ScheduledFuture m_pingerHandler;
|
private ScheduledFuture m_pingerHandler;
|
||||||
private String m_macAddress;
|
private String m_macAddress;
|
||||||
|
|
||||||
|
private String m_clientID;
|
||||||
|
|
||||||
final Runnable pingerDeamon = new Runnable() {
|
final Runnable pingerDeamon = new Runnable() {
|
||||||
public void run() {
|
public void run() {
|
||||||
LOG.debug("Pingreq sent");
|
LOG.debug("Pingreq sent");
|
||||||
@@ -92,6 +95,14 @@ public final class Client {
|
|||||||
init();
|
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() {
|
protected void init() {
|
||||||
DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
|
DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
|
||||||
decoder.addMessageDecoder(new ConnAckDecoder());
|
decoder.addMessageDecoder(new ConnAckDecoder());
|
||||||
@@ -120,7 +131,16 @@ public final class Client {
|
|||||||
m_macAddress = readMACAddress();
|
m_macAddress = readMACAddress();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Connects to the server with clean session set to true, do not maintains
|
||||||
|
* client topic subscriptions
|
||||||
|
*/
|
||||||
public void connect() throws MQTTException {
|
public void connect() throws MQTTException {
|
||||||
|
connect(true);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
public void connect(boolean cleanSession) throws MQTTException {
|
||||||
int retries = 0;
|
int retries = 0;
|
||||||
|
|
||||||
try {
|
try {
|
||||||
@@ -141,7 +161,12 @@ public final class Client {
|
|||||||
//send a message over the session
|
//send a message over the session
|
||||||
ConnectMessage connMsg = new ConnectMessage();
|
ConnectMessage connMsg = new ConnectMessage();
|
||||||
connMsg.setKeepAlive(KEEPALIVE_SECS);
|
connMsg.setKeepAlive(KEEPALIVE_SECS);
|
||||||
connMsg.setClientID(generateClientID());
|
if (m_clientID == null) {
|
||||||
|
m_clientID = generateClientID();
|
||||||
|
}
|
||||||
|
connMsg.setClientID(m_clientID);
|
||||||
|
connMsg.setCleanSession(cleanSession);
|
||||||
|
|
||||||
m_session.write(connMsg);
|
m_session.write(connMsg);
|
||||||
|
|
||||||
//suspend until the server respond with CONN_ACK
|
//suspend until the server respond with CONN_ACK
|
||||||
@@ -233,8 +258,7 @@ public final class Client {
|
|||||||
throw new SubscribeException(ex);
|
throw new SubscribeException(ex);
|
||||||
}
|
}
|
||||||
|
|
||||||
//register the publishCallback in some registry to be notified
|
register(topic, publishCallback);
|
||||||
m_subscribersList.put(topic, publishCallback);
|
|
||||||
|
|
||||||
//wait for the SubAck
|
//wait for the SubAck
|
||||||
m_subscribeBarrier = new CountDownLatch(1);
|
m_subscribeBarrier = new CountDownLatch(1);
|
||||||
@@ -295,7 +319,7 @@ public final class Client {
|
|||||||
|
|
||||||
StringBuilder sb = new StringBuilder();
|
StringBuilder sb = new StringBuilder();
|
||||||
for (int i = 0; i < mac.length; i++) {
|
for (int i = 0; i < mac.length; i++) {
|
||||||
sb.append(String.format("%02X%s", mac[i], (i < mac.length - 1) ? "-" : ""));
|
sb.append(String.format("%02X%s", mac[i], ""));
|
||||||
}
|
}
|
||||||
return sb.toString();
|
return sb.toString();
|
||||||
} catch (Exception ex) {
|
} catch (Exception ex) {
|
||||||
@@ -305,7 +329,9 @@ public final class Client {
|
|||||||
|
|
||||||
private String generateClientID() {
|
private String generateClientID() {
|
||||||
double rnd = Math.random();
|
double rnd = Math.random();
|
||||||
return "Moquette-client_" + m_macAddress + Math.round(rnd*1000);
|
String id = "Moque" + m_macAddress + Math.round(rnd*1000);
|
||||||
|
LOG.debug("Generated ClientID " + id);
|
||||||
|
return id;
|
||||||
}
|
}
|
||||||
|
|
||||||
public void close() {
|
public void close() {
|
||||||
@@ -317,7 +343,21 @@ public final class Client {
|
|||||||
|
|
||||||
// wait until the summation is done
|
// wait until the summation is done
|
||||||
m_session.getCloseFuture().awaitUninterruptibly();
|
m_session.getCloseFuture().awaitUninterruptibly();
|
||||||
|
// m_connector.dispose();
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
public void shutdown() {
|
||||||
m_connector.dispose();
|
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);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user