From 5540805570b5df5043727ba81cbed84318d534cb Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Thu, 9 Aug 2012 23:14:29 +0200 Subject: [PATCH] Added Client ID generator --- .../org/dna/mqtt/moquette/client/Client.java | 132 ++++++++++-------- 1 file changed, 75 insertions(+), 57 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java index 5c6467e5..0714b203 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java @@ -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 m_subscribersList = new HashMap(); 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();