{"action":"created","comment":{"url":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/comments/53628031","id":53628031,"diff_hunk":"@@ -0,0 +1,356 @@\n+package org.apache.axis2.transport.mqtt;\n+/*\n+*  Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.\n+*\n+*  WSO2 Inc. licenses this file to you under the Apache License,\n+*  Version 2.0 (the \"License\"); you may not use this file except\n+*  in compliance with the License.\n+*  You may obtain a copy of the License at\n+*\n+*    http://www.apache.org/licenses/LICENSE-2.0\n+*\n+* Unless required by applicable law or agreed to in writing,\n+* software distributed under the License is distributed on an\n+* \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n+* KIND, either express or implied.  See the License for the\n+* specific language governing permissions and limitations\n+* under the License.\n+*/\n+\n+import org.apache.commons.logging.Log;\n+import org.apache.commons.logging.LogFactory;\n+import org.eclipse.paho.client.mqttv3.*;\n+\n+import java.io.IOException;\n+import java.sql.Timestamp;\n+\n+public class MqttAsyncCallback implements MqttCallback {\n+\n+    int state = BEGIN;\n+\n+    static final int BEGIN = 0;\n+    static final int CONNECTED = 1;\n+    static final int PUBLISHED = 2;\n+    static final int SUBSCRIBED = 3;\n+    static final int DISCONNECTED = 4;\n+    static final int FINISH = 5;\n+    static final int ERROR = 6;\n+    static final int DISCONNECT = 7;\n+\n+    private MqttConnectOptions conOpt;\n+    private Log log = LogFactory.getLog(MqttAsyncCallback.class);\n+\n+    // Private instance variables\n+    private MqttAsyncClient client;\n+    private String brokerUrl;\n+    private Throwable ex = null;\n+    private final Object waiter = new Object();\n+    private boolean donext = false;\n+\n+    public void setConOpt(MqttConnectOptions conOpt) {\n+        this.conOpt = conOpt;\n+    }\n+\n+    public MqttAsyncCallback(MqttAsyncClient clientAsync) throws MqttException {\n+        client = clientAsync;\n+        // Set this wrapper as the callback handler\n+        client.setCallback(this);\n+    }\n+\n+    /**\n+     * Publish / send a message to an MQTT server\n+     *\n+     * @param topicName the name of the topic to publish to\n+     * @param message   the set of bytes to send to the MQTT server\n+     * @throws MqttException\n+     */\n+    public void publish(String topicName, MqttMessage message) throws Throwable {\n+        // Use a state machine to decide which step to do next. State change occurs\n+        // when a notification is received that an MQTT action has completed\n+        while (state != FINISH) {\n+            switch (state) {\n+                case BEGIN:\n+                    // Connect using a non-blocking connect\n+                    MqttConnector con = new MqttConnector();\n+                    con.doConnect();\n+                    break;\n+                case CONNECTED:\n+                    // Publish using a non-blocking publisher\n+                    Publisher pub = new Publisher();\n+                    pub.doPublish(topicName, message);\n+                    break;\n+                case PUBLISHED:\n+                    state = DISCONNECT;\n+                    donext = true;\n+                    break;\n+                case DISCONNECT:\n+                    Disconnector disc = new Disconnector();\n+                    disc.doDisconnect();\n+                    break;\n+                case ERROR:\n+                    throw ex;\n+                case DISCONNECTED:\n+                    state = FINISH;\n+                    donext = true;\n+                    break;\n+            }\n+            waitForStateChange(10000);\n+        }\n+    }\n+\n+    /**\n+     * Wait for a maximum amount of time for a state change event to occur\n+     *\n+     * @param maxTTW maximum time to wait in milliseconds\n+     * @throws MqttException\n+     */\n+    private void waitForStateChange(int maxTTW) throws MqttException {\n+        synchronized (waiter) {\n+            if (!donext) {\n+                try {\n+                    waiter.wait(maxTTW);\n+                } catch (InterruptedException e) {\n+                    log.info(\"timed out\");\n+                    e.printStackTrace();\n+                }\n+                if (ex != null) {\n+                    throw (MqttException) ex;\n+                }\n+            }\n+            donext = false;\n+        }\n+    }\n+\n+    /**\n+     * Subscribe to a topic on an MQTT server\n+     * Once subscribed this method waits for the messages to arrive from the server\n+     * that match the subscription. It continues listening for messages until the enter key is\n+     * pressed.\n+     *\n+     * @param topicName to subscribe to (can be wild carded)\n+     * @param qos       the maximum quality of service to receive messages at for this subscription\n+     * @throws MqttException\n+     */\n+    public void subscribe(String topicName, int qos) throws Throwable {\n+        // Use a state machine to decide which step to do next. State change occurs\n+        // when a notification is received that an MQTT action has completed\n+        while (state != FINISH) {\n+            switch (state) {\n+                case BEGIN:\n+                    // Connect using a non-blocking connect\n+                    MqttConnector con = new MqttConnector();\n+                    con.doConnect();\n+                    break;\n+                case CONNECTED:\n+                    // Subscribe using a non-blocking subscribe\n+                    Subscriber sub = new Subscriber();\n+                    sub.doSubscribe(topicName, qos);\n+                    break;\n+                case SUBSCRIBED:\n+                    // Block until Enter is pressed allowing messages to arrive\n+                    log.info(\"Press <Enter> to exit\");\n+                    try {\n+                        System.in.read();\n+                    } catch (IOException e) {\n+                        //If we can't read we'll just exit\n+                    }\n+                    state = DISCONNECT;\n+                    donext = true;\n+                    break;\n+                case DISCONNECT:\n+                    Disconnector disc = new Disconnector();\n+                    disc.doDisconnect();\n+                    break;\n+                case ERROR:\n+                    throw ex;\n+                case DISCONNECTED:\n+                    state = FINISH;\n+                    donext = true;\n+                    break;\n+            }\n+            waitForStateChange(10000);\n+        }\n+    }\n+\n+\n+    public void connectionLost(Throwable throwable) {\n+        //ignoring for the moment...\n+    }\n+\n+    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {\n+        throw new IllegalStateException();\n+    }\n+\n+    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {\n+        log.info(\"message delivered .. : \" + iMqttDeliveryToken.toString());\n+    }\n+\n+    /**\n+     * Connect in a non-blocking way and then sit back and wait to be\n+     * notified that the action has completed.\n+     */\n+    public class MqttConnector {\n+\n+        public MqttConnector() {\n+        }\n+        public void doConnect() {\n+            // Connect to the server\n+            // Get a token and setup an asynchronous listener on the token which\n+            // will be notified once the connect completes\n+            log.info(\"Connecting to \" + brokerUrl + \" with client ID \" + client.getClientId());\n+\n+            IMqttActionListener conListener = new IMqttActionListener() {\n+                public void onSuccess(IMqttToken asyncActionToken) {\n+                    log.info(\"Connected\");\n+                    state = CONNECTED;\n+                    carryOn();\n+                }\n+                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {\n+                    ex = exception;\n+                    state = ERROR;\n+                    log.info(\"connect failed\" + exception);\n+                    carryOn();\n+                }\n+                public void carryOn() {\n+                    synchronized (waiter) {\n+                        donext = true;\n+                        waiter.notifyAll();\n+                    }\n+                }\n+            };\n+\n+            try {\n+                // Connect using a non-blocking connect\n+                client.connect(conOpt, \"Connect sample context\", conListener);\n+\n+            } catch (MqttException e) {\n+                // If though it is a non-blocking connect an exception can be\n+                // thrown if validation of parms fails or other checks such\n+                // as already connected fail.\n+                state = ERROR;\n+                donext = true;\n+                ex = e;\n+            }\n+        }\n+    }\n+\n+    /**\n+     * Publish in a non-blocking way and then sit back and wait to be\n+     * notified that the action has completed.\n+     */\n+    public class Publisher {\n+        public void doPublish(String topicName, MqttMessage message) {\n+            // Send / publish a message to the server\n+            // Get a token and setup an asynchronous listener on the token which\n+            // will be notified once the message has been delivered\n+            // MqttMessage message = new MqttMessage(payload);\n+            //message.setQos(qos);\n+            String time = new Timestamp(System.currentTimeMillis()).toString();\n+\n+            // Setup a listener object to be notified when the publish completes.\n+            IMqttActionListener pubListener = new IMqttActionListener() {\n+                public void onSuccess(IMqttToken asyncActionToken) {\n+                    log.info(\"Publish Completed\");\n+                    state = PUBLISHED;\n+                    carryOn();\n+                }\n+                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {\n+                    ex = exception;\n+                    state = ERROR;\n+                    log.info(\"Publish failed\" + exception);\n+                    carryOn();\n+                }\n+                public void carryOn() {","path":"modules/mqtt/src/main/java/org/apache/axis2/transport/mqtt/MqttAsyncCallback.java","position":263,"original_position":263,"commit_id":"4c5add1eda9af8eeb86fb49788e873b3cbba8f07","original_commit_id":"4c5add1eda9af8eeb86fb49788e873b3cbba8f07","user":{"login":"ashalya","id":8655189,"avatar_url":"https://avatars.githubusercontent.com/u/8655189?v=3","gravatar_id":"","url":"https://api.github.com/users/ashalya","html_url":"https://github.com/ashalya","followers_url":"https://api.github.com/users/ashalya/followers","following_url":"https://api.github.com/users/ashalya/following{/other_user}","gists_url":"https://api.github.com/users/ashalya/gists{/gist_id}","starred_url":"https://api.github.com/users/ashalya/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ashalya/subscriptions","organizations_url":"https://api.github.com/users/ashalya/orgs","repos_url":"https://api.github.com/users/ashalya/repos","events_url":"https://api.github.com/users/ashalya/events{/privacy}","received_events_url":"https://api.github.com/users/ashalya/received_events","type":"User","site_admin":false},"body":"fixed in https://github.com/wso2/wso2-axis2-transports/pull/66","created_at":"2016-02-22T14:04:06Z","updated_at":"2016-02-22T14:04:06Z","html_url":"https://github.com/wso2/wso2-axis2-transports/pull/38#discussion_r53628031","pull_request_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/38","_links":{"self":{"href":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/comments/53628031"},"html":{"href":"https://github.com/wso2/wso2-axis2-transports/pull/38#discussion_r53628031"},"pull_request":{"href":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/38"}}},"pull_request":{"url":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/38","id":36129989,"html_url":"https://github.com/wso2/wso2-axis2-transports/pull/38","diff_url":"https://github.com/wso2/wso2-axis2-transports/pull/38.diff","patch_url":"https://github.com/wso2/wso2-axis2-transports/pull/38.patch","issue_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/issues/38","number":38,"state":"closed","locked":false,"title":"mqtt axis2 transport","user":{"login":"ashalya","id":8655189,"avatar_url":"https://avatars.githubusercontent.com/u/8655189?v=3","gravatar_id":"","url":"https://api.github.com/users/ashalya","html_url":"https://github.com/ashalya","followers_url":"https://api.github.com/users/ashalya/followers","following_url":"https://api.github.com/users/ashalya/following{/other_user}","gists_url":"https://api.github.com/users/ashalya/gists{/gist_id}","starred_url":"https://api.github.com/users/ashalya/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ashalya/subscriptions","organizations_url":"https://api.github.com/users/ashalya/orgs","repos_url":"https://api.github.com/users/ashalya/repos","events_url":"https://api.github.com/users/ashalya/events{/privacy}","received_events_url":"https://api.github.com/users/ashalya/received_events","type":"User","site_admin":false},"body":"","created_at":"2015-05-25T10:09:17Z","updated_at":"2016-02-22T14:04:06Z","closed_at":"2015-07-07T04:28:13Z","merged_at":"2015-07-07T04:28:13Z","merge_commit_sha":"94878a38be45fe02613f4ef1d9fcb5c3037ac770","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/38/commits","review_comments_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/38/comments","review_comment_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/comments{/number}","comments_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/issues/38/comments","statuses_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/statuses/4c5add1eda9af8eeb86fb49788e873b3cbba8f07","head":{"label":"ashalya:master","ref":"master","sha":"4c5add1eda9af8eeb86fb49788e873b3cbba8f07","user":{"login":"ashalya","id":8655189,"avatar_url":"https://avatars.githubusercontent.com/u/8655189?v=3","gravatar_id":"","url":"https://api.github.com/users/ashalya","html_url":"https://github.com/ashalya","followers_url":"https://api.github.com/users/ashalya/followers","following_url":"https://api.github.com/users/ashalya/following{/other_user}","gists_url":"https://api.github.com/users/ashalya/gists{/gist_id}","starred_url":"https://api.github.com/users/ashalya/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ashalya/subscriptions","organizations_url":"https://api.github.com/users/ashalya/orgs","repos_url":"https://api.github.com/users/ashalya/repos","events_url":"https://api.github.com/users/ashalya/events{/privacy}","received_events_url":"https://api.github.com/users/ashalya/received_events","type":"User","site_admin":false},"repo":null},"base":{"label":"wso2:master","ref":"master","sha":"fbb0c5086f4a2482940da392ff9f84443683c054","user":{"login":"wso2","id":533043,"avatar_url":"https://avatars.githubusercontent.com/u/533043?v=3","gravatar_id":"","url":"https://api.github.com/users/wso2","html_url":"https://github.com/wso2","followers_url":"https://api.github.com/users/wso2/followers","following_url":"https://api.github.com/users/wso2/following{/other_user}","gists_url":"https://api.github.com/users/wso2/gists{/gist_id}","starred_url":"https://api.github.com/users/wso2/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/wso2/subscriptions","organizations_url":"https://api.github.com/users/wso2/orgs","repos_url":"https://api.github.com/users/wso2/repos","events_url":"https://api.github.com/users/wso2/events{/privacy}","received_events_url":"https://api.github.com/users/wso2/received_events","type":"Organization","site_admin":false},"repo":{"id":25507523,"name":"wso2-axis2-transports","full_name":"wso2/wso2-axis2-transports","owner":{"login":"wso2","id":533043,"avatar_url":"https://avatars.githubusercontent.com/u/533043?v=3","gravatar_id":"","url":"https://api.github.com/users/wso2","html_url":"https://github.com/wso2","followers_url":"https://api.github.com/users/wso2/followers","following_url":"https://api.github.com/users/wso2/following{/other_user}","gists_url":"https://api.github.com/users/wso2/gists{/gist_id}","starred_url":"https://api.github.com/users/wso2/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/wso2/subscriptions","organizations_url":"https://api.github.com/users/wso2/orgs","repos_url":"https://api.github.com/users/wso2/repos","events_url":"https://api.github.com/users/wso2/events{/privacy}","received_events_url":"https://api.github.com/users/wso2/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/wso2/wso2-axis2-transports","description":"","fork":false,"url":"https://api.github.com/repos/wso2/wso2-axis2-transports","forks_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/forks","keys_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/keys{/key_id}","collaborators_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/teams","hooks_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/hooks","issue_events_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/issues/events{/number}","events_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/events","assignees_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/assignees{/user}","branches_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/branches{/branch}","tags_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/tags","blobs_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/git/refs{/sha}","trees_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/git/trees{/sha}","statuses_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/statuses/{sha}","languages_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/languages","stargazers_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/stargazers","contributors_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/contributors","subscribers_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/subscribers","subscription_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/subscription","commits_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/commits{/sha}","git_commits_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/git/commits{/sha}","comments_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/comments{/number}","issue_comment_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/issues/comments{/number}","contents_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/contents/{+path}","compare_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/compare/{base}...{head}","merges_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/merges","archive_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/downloads","issues_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/issues{/number}","pulls_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls{/number}","milestones_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/milestones{/number}","notifications_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/labels{/name}","releases_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/releases{/id}","deployments_url":"https://api.github.com/repos/wso2/wso2-axis2-transports/deployments","created_at":"2014-10-21T07:03:40Z","updated_at":"2016-02-19T10:58:44Z","pushed_at":"2016-02-22T08:02:03Z","git_url":"git://github.com/wso2/wso2-axis2-transports.git","ssh_url":"git@github.com:wso2/wso2-axis2-transports.git","clone_url":"https://github.com/wso2/wso2-axis2-transports.git","svn_url":"https://github.com/wso2/wso2-axis2-transports","homepage":null,"size":887,"stargazers_count":4,"watchers_count":4,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":41,"mirror_url":null,"open_issues_count":3,"forks":41,"open_issues":3,"watchers":4,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/38"},"html":{"href":"https://github.com/wso2/wso2-axis2-transports/pull/38"},"issue":{"href":"https://api.github.com/repos/wso2/wso2-axis2-transports/issues/38"},"comments":{"href":"https://api.github.com/repos/wso2/wso2-axis2-transports/issues/38/comments"},"review_comments":{"href":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/38/comments"},"review_comment":{"href":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/wso2/wso2-axis2-transports/pulls/38/commits"},"statuses":{"href":"https://api.github.com/repos/wso2/wso2-axis2-transports/statuses/4c5add1eda9af8eeb86fb49788e873b3cbba8f07"}}}}