mqtt-connection.html

<link rel="import" href="../polymer/polymer.html">
<link rel="import" href="mqttjs.html">

<!--
The `<mqtt-connection>` element exposes MQTT connection functionality to the web.
Use one `<mqtt-connection>` element per connection to a mqtt broker e.g.
[mosca](http://www.mosca.io)
Currently the element supports `ws` and `wss` connections.

    <mqtt-connection
        url="ws://127.0.0.1:8080"
        options='{reconnectPeriod: 1000}'>
    </mqtt-connection>


With `auto` set to `true`, the element performs a MQTT.Connect whenever
the element is ready or the connection gets disconnected.

    <mqtt-connection auto></mqtt-connection>

Note: The `options` attribute must be double quoted JSON.

You can disconnect the mqtt-connection explicitly by calling `disconnect` on the
element. And manually connect if `auto` is set to `false` with `connect`

@demo demo/index.html
-->

<dom-module id="mqtt-connection">
  <template>
    <content id="mqttSubscriptions" select="mqtt-subscription"></content>
    <content id="mqttPublications" select="mqtt-publish"></content>
    <content id="content"></content>
  </template>
</dom-module>

<script>
  'use strict';
  (function (scope) {

    var MqttElements = scope.MqttElements = scope.MqttElements || {};

    MqttElements.MqttConnection = Polymer({
      is: 'mqtt-connection',

      /**
       * Fired when a connection is connected.
       *
       * @mqtt-subscription-unregister mqtt-connection-connected
       */

      /**
       * Fired when a connection is closed.
       *
       * @event mqtt-connection-close
       */

      /**
       * Fired when a connection is closed.
       *
       * @event mqtt-connection-offline
       */

      /**
       * Fired when a connection error occurs.
       *
       * @event mqtt-connection-error
       */

      /**
       * Fired when a mqtt message is received .
       *
       * @event mqtt-message
       */

      properties: {

        /**
         * When injecting the client object into different `<mqtt-connection>` element, each element will register to
         * listen for changes/ messages in the connection. To detect memory leaks the the `EventEmitter` raises an error
         * when more than `EventEmitter.maxListeners` (default 10) are registering to the client.
         *
         */
        maxListeners: {
          type: Number,
          value: 10,
          observer: "_maxListenersChanged"
        },

        /**
         * Set the clean flag to `false` to receive QOS 1 and 2 messages while offline. The messages are delivered if the
         * connection is re-establish with the broker.
         */
        clean: {
          type: Boolean,
          readOnly: false,
          value: true,
        },

        /**
         * The actual MQTT.js client that will be created on the first connect or injected
         * via `<mqtt-connection client="[[client]]">`
         */
        client: {
          type: Object,
          readOnly: false,
          notify: true,
          observer: '_clientChanged',
        },

        /**
         * The prefix of the option.clientId. If the option.clientId is not set a random ID will be generated prefixed
         * with clientIdPrefix
         */
        clientIdPrefix: {
          type: String,
          value: 'mqttjs_',
        },

        /**
         * Indicates weather the mqtt client is connected to the mqtt broker or not
         */
        connected: {
          type: Boolean,
          readOnly: true,
          notify: true,
          value: false,
          observer: '_connectedChanged',
        },

        /**
         * Indicates weather the mqtt client is currently disconnecting the mqtt connection or not
         */
        disconnecting: {
          type: Boolean,
          readOnly: true,
          notify: true,
          value: false,
        },

        /**
         * A reference to the MQEmitter that is used to route messages between the `mqtt-connection` element and the
         * `mqtt-subscription` elements.
         *
         * @attribute mqEmitter
         * @type MQEmitter
         * @default `new MQEmitter`
         */
        mqEmitter: {
          type: Object,
          notify: true,
          readOnly: true,
          value: function () {
            return require("MQEmitter")();
          },
        },

        /**
         * The options are directly passed to MQTT.js
         * For further documentation see https://github.com/mqttjs/MQTT.js/wiki/connection#mqttconnectionconnectoptions
         * * options.protocolId :String The mqtt protocolId tha is send to the mqtt broker DEFAULT: "MQTT"
         * * options.protocolVersion :number 4
         * * options.keepalive :number 120
         * * options.clientId :String
         * * options.clean :boolean
         * * options.username :String The username that is user to authenticate the connection with
         * * options.password :String The password that will be used to authenticate the `user` at the mqtt broker
         * * options.will :Object The clients will message that is published from the mqtt broker in behalf of the client
         * if the connection is from the client to the mqtt broker is lost.
         * * options.will.topic :String The topic of the will message of the client eg. "client/"
         * * options.will.qos :number The QOS of the will message of the client
         * * options.will.retain :boolean The retain flag of the will message of the client
         */
        options: {
          type: Object,
          readOnly: false,
          notify: true,
          value: function () {
            return {
              //MQIsdp
              protocolId: "MQTT",
              protocolVersion: 4,
              keepalive: 120,
              reconnectPeriod: 1000,
              clientId: "",
              encoding: "string",
            };
          },
        },

        username: {
          type: String,
          readOnly: false,
          notify: true,
          observer: "_usernameChanged"
        },

        password: {
          type: String,
          readOnly: false,
          notify: true,
          observer: "_passwordChanged"
        },


        /**
         * An array of all `<mqtt-publish>` elements that are placed within this `<mqtt-connection>` element.
         */
        publishers: {
          type: Array,
          notify: true,
          readOnly: true,
          value: function () {
            return [];
          },
        },

        /**
         * An array of all `<mqtt-subscription>` elements that are placed within this `<mqtt-connection>` element. Use
         * this to iterate over all subscriptions.
         */
        subscriptions: {
          type: Array,
          notify: true,
          readOnly: true,
          value: function () {
            return [];
          },
        },

        /**
         * The url of the mqtt broker and can be on the following protocols: 'mqtt', 'mqtts', 'tcp', 'tls', 'ws', 'wss'.
         * E.g ws://localhost:8080
         */
        url: {
          type: String,
          value: '',
        },

        withCredentials: {
          type: Boolean,
          readOnly: false,
          notify: true,
          value: false,
        },

        /**
         * Set to `true` to automatically create and connect the mqtt connection.
         */
        auto: {
          type: Boolean,
          readOnly: false,
          value: false,
        },

        packetQueue: {
          type: Array,
          value: function(){
            return [];
          }
        }

      },

      observers: [
        '_usernameOrPasswordChanged(username, options.username, password, options.password, withCredentials, auto)',
        '_autoChanged(auto, withCredentials)'
      ],

      listeners: {
        'mqtt-subscription-register': '_subscriptionElementRegistered',
        'mqtt-subscription-unregister': '_subscriptionElementUnregistered',
        'mqtt-publish-register': '_publishElementRegistered',
        'mqtt-publish-unregister': '_publishElementUnregistered',
      },

      attached: function () {
        if(!this.client && this.auto && !this.withCredentials) {
          this.connect();
        }
      },

      _usernameChanged: function (username, old) {
        this.set("options.username", username);
      },

      _passwordChanged: function (password, old) {
        this.set("options.password", password);
      },

      _autoChanged: function(auto, withCredentials){
        if (!this.client && auto && !withCredentials){
          this.connect();
        }
      },

      _usernameOrPasswordChanged: function (username, optUsername, password, optPassword, withCredentials, auto) {
        if (!this.client && auto && ( !withCredentials || ( withCredentials && ( username || optUsername )  && (password || optPassword )))){
          this.connect()
        }
      },

      _maxListenersChanged: function (maxListeners, old) {
        if(this.client && maxListeners) {
          this.client.setMaxListeners(maxListeners);
        }
      },

      /**
       * connect - connect to an MQTT broker.
       */
      connect: function () {
        // creates the client and automatically connects
        this._createClient();
      },

      /**
       * diconnect - disconnect for the MQTT broker
       */
      disconnect: function () {
        // check if we already have a client before calling end/ disconnect
        if(this.client) {
          this.client.end(true);
        }
      },

      /**
       * Publish to an MQTT topic
       * @param topic
       * @param message
       * @param opt_qos
       * @param opt_retained
       * @param opt_callback
       */
      publish: function (topic, message, opt_qos, opt_retained, opt_callback) {

        // the optional parameters to the publish
        var opts = {
          // default qos is `0`
          qos: opt_qos || 0,
          // default for retained is `false`
          retain: opt_retained || false
        };

        if(this.client) {
          // actually send the `publish` message to the MQTT broker
          this.client.publish(topic, message, opts, opt_callback);
        } else {
          var nop = function () {};
          var queuePacket = {
            cb: opt_callback || nop,
            packet: {
              cmd: 'publish',
              topic: topic,
              payload: message,
              qos: opt_qos || 0,
              retain: opt_retained || false,
              messageId: 1
            }
          };

          this.push("packetQueue", queuePacket);
        }
      },

      /**
       * Subscribe to an MQTT topic
       * @param topic that the subscription is made to
       * @param opt_qos [0,1,2] with the subscription is made with
       * @param opt_callback an callback that will be executed when the `suback` message is received
       */
      subscribe: function (topic, opt_qos, opt_callback) {
        var opts = {qos: opt_qos || 0};
        if(this.client) {
          this.client.subscribe(topic, opts, opt_callback);
        }
      },

      /**
       * Pushes an element onto an array. To notify the changes `Polymer.Base.push` is used to alter the array
       * @private
       * @param path to the array
       * @param element that will be put into the array
       */
      _addElementAndRegister: function (path, element) {
        // adding the element to the array
        this.push(path, element);
        // setting this connection as the parent on the connection
        this._setParentConnectionOnElement(element);
      },

      /**
       * A change listener to register event listeners on the `mqtt.client`
       * @private
       * @param client an instance of `MQTT.js#MqttClient`
       */
      _clientChanged: function (client) {
        if(client) {
          client
              .on("close", this._handelClose.bind(this))
              .on("connect", this._handelConnected.bind(this))
              .on("error", this._handelError.bind(this))
              .on("message", this._handelMessage.bind(this))
              .on("offline", this._handelOffline.bind(this))
              .on("reconnect", this._handelReconnect.bind(this));
        }
      },

      /**
       * A listener that is run each time the `client.connected` changes
       * @private
       * @param connected true if `client.connected` is connected
       * @param old is true if connected is false
       */
      _connectedChanged: function (connected, old) {
        if(connected) {
          // after the connection is established we need to subscribe/ resubscribe
          this._subscribeAllSubscriptions();
        } else if(old) {
          this._setUnsubscribedForAllSubscriptions();
        }
      },

      /**
       * Creates a new instance of MQTT.js#MqttClient
       * @private
       */
      _createClient: function () {
        // require mqtt from bundled browserified dependencies
        var mqtt = require("mqtt");
        if(mqtt) {
          this._createClientDebounce(mqtt);
//          return this.debounce('_createClient', this._createClientDebounce());
        } else {
          // failed to load mqtt.js
          // TODO SKO tell the user about this and maybe log environment
          this.fire("mqtt-connection-error", "<mqtt-connection> element could not load MQTT.js, pleas check your setup");
          console.error("<mqtt-connection> element could not load MQTT.js, pleas check your setup");
        }
      },

      _createClientDebounce: function(mqtt){
        if (this.options.clientId && this.options.clientId.length == 0){
          this.set("options.clientId", this._generateClientId());
        }
        this.options["queue"] = this.packetQueue;

        this.client = mqtt.connect(this.url, this.options);
        this.client.setMaxListeners(this.maxListeners);
      },

      /**
       * Creates a subscription for a `<mqtt-subscription>` element
       * @private
       * @param sub the `<mqtt-subscription>` element for that the subscription should be created
       */
      _createSubscription: function (sub) {
        if(this.client) {
          // subscribe the element at the mqtt-broker
          this.subscribe(sub.topic, sub.qos, sub._handelSubscriptionSubscribedFunc);

          // add the element to the mqEmitter to receive messages on the topic
          this.mqEmitter.on(sub.topic, sub._handelMessageFunc);
        }
      },

      /**
       * Generates a random mqtt client ID
       * @returns {string} the X digit mqtt client ID prefixed with 'mqttjs_'. Set `clientIdPrefix` prefix the clientID
       * with something else
       * @private
       */
      _generateClientId: function () {
        return this.clientIdPrefix + Math.random().toString(16).substr(2, 8);
      },

      /**
       * Returns true if the MQTTClient is connected otherwise false.
       * @private
       * @param client to check the connection status on
       * @return true if the client is connected otherwise false
       */
      _getConnectionStatusOfConnection: function (client) {
        return client ? client.connected : false
      },

      /**
       * Event-Handler to handel the MQTTClient close event
       * @private
       */
      _handelClose: function () {
        this._setMqttConnected(this._getConnectionStatusOfConnection(this.client));
        this.fire("mqtt-connection-close");
      },

      /**
       * Event-Handler to handel the MQTTClient connect event
       * @private
       */
      _handelConnected: function () {
        this._setMqttConnected(this._getConnectionStatusOfConnection(this.client));
        this.fire("mqtt-connection-connected", {value: this.connected});
      },

      /**
       * Event-Handler to handel the MQTTClient error event
       * @private
       */
      _handelError: function () {
        this.fire("mqtt-connection-error");
      },

      /**
       * Event-Handler to handel the MQTTClient message event
       * @private
       * @param topic of the message
       * @param message the actual message
       * @param packet the MQTT-Packet
       */
      _handelMessage: function (topic, message, packet) {
        this.fire("mqtt-message", {'topic': topic, 'message': message, 'packet': packet});
        this.mqEmitter.emit({topic: topic, message: packet})
      },

      /**
       * Event-Handler to handel the MQTTClient offline event
       * @private
       */
      _handelOffline: function () {
        this._setMqttConnected(this._getConnectionStatusOfConnection(this.client));
        this.fire("mqtt-connection-offline", {value: this.connected});
      },

      /**
       * Event-Handler to handel the MQTTClient reconnect event
       * @private
       */
      _handelReconnect: function () {
        this.fire("mqtt-connection-reconnect");
      },

      /**
       * Event-Handler that is called when a <mqtt-subscriptions> is changed, for example when it recieves a mqtt-message
       * @private
       * @param element the <mqtt-subscriptions> elemment that has changed
       */
      _notifySubscriptionChanged: function (element) {
        this._propagateElementChanged(element, "subscriptions");
      },

      /**
       * Propagates the changes of any element e.g. <mqtt-subscriptions> to other observing elements e.g. <dom-repeat>
       * @private
       * @param element
       * @param arrElementPath
       */
      _propagateElementChanged: function (element, arrElementPath) {
        // the array property that hold the element that changed
        var arrName = arrElementPath;
        // get the index of the element
        var index = this[arrName].indexOf(element);
        // if the element exists within the array
        if(index >= 0) {
          // get the polymer collection key see:
          // https://github.com/Polymer/polymer/issues/2068
          var polymerPathId = Polymer.Collection.get(this.get(arrName)).getKey(this[arrName][index]);

          // check if the id is valid
          if(polymerPathId >= 0) {

            // get all defined properties of the custom element
            var propertyNames = Object.getOwnPropertyNames(this[arrName][index].properties);

            // iterate over all properties
            for (var i = 0; i < propertyNames.length; i++) {
              // build the path
              var path = [arrName, polymerPathId, propertyNames[i]].join(".");

              // check if the element is an array, then slice needs to get called to actually result in
              // Polymer_propertySetter
              // old !== value => true
              if(Array.isArray(this[arrName][index][propertyNames[i]])) {
                this.notifyPath(path, this[arrName][index][propertyNames[i]].slice());
              }
              // TOOD
              // if the prop is an object each property has to be checked recursively
              // for now use
              // this.set($path, {$prop1: $prop1Old, $prop2: prop2Changed};
              // to notify the path about changes
              // else if(typeof this[arrName][index][propertyNames[i]] === "object") {
              //
              // }
              else {
                this.notifyPath(path, this[arrName][index][propertyNames[i]]);
              }
            }
          }
        }
      },

      /**
       * Adds and registers and <mqtt-publish> element
       * @private
       * @param event that is fired from the <mqtt-publish> element on attach
       */
      _publishElementRegistered: function (event) {
        this._addElementAndRegister("publishers", event.target);
      },

      /**
       * Removes and <mqtt-publish> element
       * @private
       * @param event that is fires from the <mqtt-publish> element on detached
       */
      _publishElementUnregistered: function (event) {
        this._unregisterElement("publishers", event.detail.target);
      },

      /**
       * Removes an <mqtt-subscription> element from the MQEmitter
       * @private
       * @param sub the <mqtt-subscription> element that will be removed from the MQEmitter
       */
      _removeSubFromEmmitter: function (sub) {
        this.mqEmitter.removeListener(sub.topic, sub._handelMessageFunc);
      },

      /**
       * Sets the current connection state
       * @private
       * @param state the actual connection state false == disconnedted ; true == connected
       */
      _setMqttConnected: function (state) {
        if(typeof state === "boolean") {
          this._setConnected(state);
        }
      },

      /**
       * Sets this <mqtt-connection> elment as the parrent connection on the element
       * @private
       * @param element
       */
      _setParentConnectionOnElement: function (element) {
        element._parentConnection = this;
      },

      /**
       * When the mqtt-connection is disconnected all subscriptions have to be marked as unsubscribed
       * @private
       */
      _setUnsubscribedForAllSubscriptions: function () {
        for (var i = 0; i < this.subscriptions.length; i++) {
          this._setUnsubscribedToSubscription(this.subscriptions[i]);
        }
      },

      /**
       * Actually sets the <mqtt-subscription> element ot unsubscribed
       * @private
       * @param sub the <mqtt-subscription> element
       */
      _setUnsubscribedToSubscription: function (sub) {
        this._removeSubFromEmmitter(sub);
        sub._handelSubscriptionUnsubscribed();
      },

      /**
       * Subscribes all subscriptions
       * @private
       */
      _subscribeAllSubscriptions: function () {
        var sub;
        for (var i = 0; i < this.subscriptions.length; i++) {
          sub = this.subscriptions[i];
          // subscribe if the subscription is not subscribed
          if(!sub.subscribed) {
            this._createSubscription(sub);
          }
        }
      },

      /**
       * Adds and registers and <mqtt-subscription> element
       * @private
       * @param event that is fired from the <mqtt-subscription> on attach
       */
      _subscriptionElementRegistered: function (event) {
        var sub = event.target;
        this._addElementAndRegister("subscriptions", sub);

        if(this.client && this.client.connected) {
          this._createSubscription(sub);
        }
      },

      /**
       * Removes a <mqtt-subscription> element
       * @private
       * @param event that is fired from the <mqtt-subscription> element on detached
       */
      _subscriptionElementUnregistered: function (event) {
        var sub = event.detail.target;
        if(sub) {
          if(sub.topic && this.client) {
            this.client.unsubscribe(sub.topic, sub._handelSubscriptionUnsubscribed.bind(sub));
            this._removeSubFromEmmitter(sub);
          }
          this._unregisterElement("subscriptions", sub);
        }
      },

      /**
       * Removes an element from the array at path
       * @private
       * @param path of the array
       * @param element that should be removed from the array
       */
      _unregisterElement: function (path, element) {
        this.arrayDelete(path, element);
      },
    })
  })(window);
</script>