fix(api): don't proxy EventEmitter

boneskull · boneskull · commit 7863d26b29be · 2017-10-12T20:39:49.000-07:00
BREAKING CHANGE: This significant API change proxies `subscribe` and
`publish` instead of `on`, `once`, and `emit`.

There is no longer an equivalent for `once`.

This was done to keep compatibility with the "interface" of an
`EventEmitter`, among other things.

Also it will require Node.js v7 or newer due to use of features
unsupported in earlier versions.

Furthermore, it fixes connection issues arising from misuse of the
`pify` module.
diff --git a/.travis.yml b/.travis.yml
@@ -4,7 +4,6 @@ notifications:
 node_js:
   - '8'
   - '7'
-  - '6.5'
 after_success:
   - nyc report --reporter=text-lcov | node_modules/.bin/coveralls
   - npm run semantic-release
diff --git a/README.md b/README.md
@@ -59,7 +59,7 @@ What's better is that `EventEmitter`s are standardized.  They are easy to consum
 
 ### Promise Support
 
-[async-mqtt](https://npm.im/async-mqtt) does the same thing.  I wanted it.
+[async-mqtt](https://npm.im/async-mqtt) does the same thing here--more or less.
 
 ### WONTFIX: Message Formats
 
@@ -95,7 +95,7 @@ client.on('radscript/4ever/format/+', (buf, {topic})=> {
 
 ## Install
 
-**Node.js v6.5 or greater required**.
+**Node.js v7.0.0 or greater required**.
 
 ```bash
 $ npm install mqttletoad
@@ -104,43 +104,44 @@ $ npm install mqttletoad
 ## Usage
 
 ```js
-async function connect() {
-  const toad = require('mqttletoad');
-  const client = await toad.connect('wxs://test.mosquitto.org')
-  // a "real" `MqttClient` event.  these are whitelisted
+const toad = require('mqttletoad');
+
+const myfunc = async () => {
+  const client = await toad.connect('wss://test.mosquitto.org');
+  
   client.on('disconnect', () => {
-    console.warn('disconnected; reconnecting...');
+    console.warn('client disconnected');
   })
-    .on('winken/+/nod', (buf, packet) => {
-      // an MQTT event
-      console.log(`topic: "${packet.topic}", message: "${String(buf)}"`);
-      }, {qos: 1})
-    .on('suback', ({topic}) => {
-      // another "real" (non-MQTT) event
-      console.log(`subscribed to ${topic}`);
-    });
+  .on('offline', () => {
+    console.warn('client offline; reconnecting...');
+  });
+  
+  // yes, `buf` is still a Buffer.
+  const suback = await client.subscribe('winken/+/nod', (buf, packet) => {
+    console.log(`topic: "${packet.topic}", message: "${String(buf)}"`);
+  }, {qos: 1});
+  
+  console.log(`subscribed to ${suback.topic} w/ QoS ${suback.qos}`);
   
-  // note this breaks the `EventEmitter#emit` contract.  this might change!
-  await client.emit('phi/slamma/jamma', 'go local sports team', {qos: 2});
+  await client.publish('winken/blinken/nod', 'foo');
 }
 ```
 
 ## API
 
-Basically it's [async-mqtt](https://npm.im/async-mqtt) except:
+Basically it's [async-mqtt](https://npm.im/async-mqtt) (which is [mqtt](https://npm.im/mqtt)) except:
 
-- Use `on(topic, [opts], handler)` to subscribe
-- Use `removeListener(topic, handler)` to unsubscribe (only if there are no more active listeners on this topic)
-- Use `emit(topic, message, [opts])` to publish
-- Listen for event `suback` if you want "subscribe" confirmation 
-- Listen for event `unsuback` if you want "unsubscribe" confirmation
-- Surprises!
+- Use `client.subscribe(topic, [opts], listener)` to *register a listener* for the topic. 
+  - `opts` are the standard options `mqtt.Client#subscribe()` supports
+  - While `mqtt.Client#subscribe()` supports an `Array` of topics, our `topic` is singular, and must be a string.
+  - Standard MQTT topic wildcards are supported, and listeners are executed first in order of specificity; i.e. `foo/bar` will take precedence over `foo/+` and `foo/+` will take precedence over `foo/#`.
+- Use `client.unsubscribe(topic, listener)` to remove the listener for the topic.
+  - This will not necessarily *unsubscribe* from the topic (at the broker level), because there may be other listeners, but it *will* remove the listener.
+- `client.end()` won't throw a fit if already disconnected
 
 ## Roadmap
 
-- [ ] Use ES modules so it can work in the browser (the whole point of this was to use it in the browser...)
-- [ ] Reconsider `emit()` `Proxy`
-- [x] Probably use [prettier](https://npm.im/prettier) 
+- [ ] Something something Rollup?
  
 ## Maintainers
 
diff --git a/lib/index.js b/lib/index.js
@@ -5,188 +5,115 @@ const pify = require('pify');
 const {EventEmitter2} = require('eventemitter2');
 
 const eventify = topic => topic.replace(/#/g, '**').replace(/\+/g, '*');
-const mqttify = event => event.replace(/\*\*/g, '#').replace(/\*/g, '+');
-
-/**
- * If the topic/event is not internal, then trigger a MQTT subscription.
- * @param {Set} clientEvents - Set of events to ignore
- * @param {string} methodName - Method name to monkeypatch (`on` or `once`)
- */
-const eventProxy = (clientEvents, methodName) => (
-  target,
-  client,
-  [topic, ...args]
-) => {
-  // let original events pass thru
-  if (clientEvents.has(topic)) {
-    return target.apply(client, [topic, ...args]);
-  }
-  let [opts, listener] = args;
-  if (typeof opts === 'function') {
-    listener = opts;
-    opts = {};
-  }
-  if (!listener) {
-    throw new TypeError('Invalid parameters');
-  }
-  const ee2Topic = eventify(topic);
-  client._topicEmitter[methodName](ee2Topic, listener);
-
-  // TODO: find a way to not subscribe to already-subscribed topics
-  // (note that options such as QoS may differ, and in that case, we need to
-  // re-subscribe)
-  client
-    .subscribe(topic, opts)
-    .catch(err => {
-      client._topicEmitter.removeListener(ee2Topic, listener);
-      client.emit('error', err);
-    })
-    .then(suback => {
-      client.emit('suback', suback);
-    });
-
-  return client;
-};
-
-const unsub = (client, topic) => {
-  return client
-    .unsubscribe(topic)
-    .then(() => {
-      client.emit('unsuback', {topic});
-    })
-    .catch(err => {
-      client.emit('error', err);
-    });
-};
-
-/*
- */
 
 /**
  * Monkeypatches a `MqttClient` instance.
  * Promisifies `end`, `subscribe`, `publish`, and `unsubscribe`.
  * Adds special behavior around `on`, `once`, `removeListener`, `emit`, etc.
  * @param {MqttClient} client - MqttClient (does not mutate)
+ * @param {Object} connack - Connection acknowledgment object
+ * @param {boolean} connack.sessionPresent - If true, not clean session
  * @returns {MqttClient} Patched client
  */
-const toadpatch = client => {
-  client = pify(client, {
-    include: ['end', 'subscribe', 'publish', 'unsubscribe']
+const toadpatch = (client, connack) => {
+  const end = pify(client.end);
+  const subscribe = pify(client.subscribe);
+  const publish = pify(client.publish);
+  const unsubscribe = pify(client.unsubscribe);
+
+  Object.defineProperty(client, 'sessionPresent', {
+    value: Boolean(connack.sessionPresent)
   });
 
-  /**
-   * Events that MqttClient actually uses internally
-   * @type {Set}
-   */
-  const clientEvents = new Set(
-    client
-      .eventNames()
-      .concat([
-        'suback',
-        'unsuback',
-        'packetsend',
-        'packetreceive',
-        'message',
-        'offline',
-        'close',
-        'reconnect'
-      ])
-  );
-
   /**
    * Adapter between MQTT topics (supporting wildcards) and the client events.
    * @type {EventEmitter2}
    * @private
    */
-  client._topicEmitter = new EventEmitter2({
+  client.toad = new EventEmitter2({
     wildcard: true,
     delimiter: '/'
-  }).on('removeListener', event => {
-    // if we're removing a listener for any reason, check to see if an
-    // "unsubscribe" needs to happen (and do it)
-    if (!client._topicEmitter.listenerCount(event)) {
-      unsub(client, mqttify(event));
-    }
   });
 
   /**
-   * This will subscribe to MQTT topics if the topic/event is *not* an internal
-   * `MqttClient` or `mqttletoad` event.  `EventEmitter#on`
+   * Subscribe to a topic with a specific listener.
    * @public
-   * @param {string} topic - MQTT topic (or event)
-   * @param {Buffer|string|ArrayBuffer|*} [message] - Event data or MQTT
-   *   message.  If the latter, must be one of `Buffer`, `string`, or
-   *   `ArrayBuffer`.
-   * @param {Object} [opts] - Any options for MQTT subscription (like `qos`),
-   *   or more data for event
-   * @param {*} [...args] - More data for event
-   * @returns {MqttClient}
+   * @function
+   * @param {string} topic - MQTT topic
+   * @param {Buffer|string} message - MQTT messqage
+   * @param {Function} listener - Listener function; called with `topic`,
+   *   `message`, and raw `packet`
+   * @param {Object} [opts] - Any options for MQTT subscription
+   * @param {number} [opts.qos=0] - QoS
+   * @returns Promise<{{topic, qos}}> Object w/ topic subscribed to and QoS
+   *   granted by broker
    */
-  client.on = new Proxy(client.on, {
-    apply: eventProxy(clientEvents, 'on')
+  client.subscribe = new Proxy(subscribe, {
+    async apply(target, client, [topic, listener, opts = {}]) {
+      if (typeof topic !== 'string' || typeof listener !== 'function') {
+        throw new TypeError('Invalid parameters');
+      }
+
+      const {toad} = client;
+      const event = eventify(topic);
+      toad.on(event, listener);
+
+      // TODO: find a way to not subscribe to already-subscribed topics
+      // TODO: note that a different QoS requires a new subscription
+      // TODO: even if the topic is identical!
+      try {
+        const result = await target.apply(client, [topic, opts]);
+        return result.shift();
+      } catch (err) {
+        toad.removeListener(event, listener);
+        throw err;
+      }
+    }
   });
 
   /**
-   * This will subscribe to MQTT topics if the topic/event is *not* an internal
-   * `MqttClient` or `mqttletoad` event.  Works like `EventEmitter#once`
+   * Topic must match exactly.
+   * Only unsubscribes at broker level if no more listeners are registered for
+   * the topic.
    * @public
-   * @param {string} topic - MQTT topic (or event)
-   * @param {Buffer|string|ArrayBuffer|*} [message] - Event data or MQTT
-   *   message.  If the latter, must be one of `Buffer`, `string`, or
-   *   `ArrayBuffer`.
-   * @param {Object} [opts] - Any options for MQTT subscription (like `qos`),
-   *   or more data for event
-   * @param {*} [...args] - More data for event
-   * @returns {MqttClient}
+   * @function
+   * @param {string} topic - MQTT topic
+   * @param {Function} listener - Listener function to remove
+   * @returns {Promise<void>}
    */
-  client.once = new Proxy(client.once, {
-    apply: eventProxy(clientEvents, 'once')
+  client.unsubscribe = new Proxy(unsubscribe, {
+    async apply(target, client, [topic, listener]) {
+      const {toad} = client;
+      const event = eventify(topic);
+      toad.removeListener(event, listener);
+      if (!toad.listenerCount(event)) {
+        return target.apply(client, topic);
+      }
+    }
   });
 
   /**
-   * If the topic/event is *not* internal, remove it from the internal EE2
-   * instance, which may cause an unsubscribe to happen.  Works like
-   * `EventEmitter#removeListener`.
+   * Disconnects client (if connected)
+   * @function
    * @public
-   * @param {string} topic - MQTT topic (or event)
-   * @param {Function} listener - Listener function to remove
-   * @returns {MqttClient}
+   * @returns {Promise<void>}
    */
-  client.removeListener = new Proxy(client.removeListener, {
-    apply(target, client, [topic, listener]) {
-      // let original events pass thru
-      if (clientEvents.has(topic)) {
-        return target.apply(client, [topic, listener]);
+  client.end = new Proxy(end, {
+    async apply(target, client, ...args) {
+      if (client.connected) {
+        return target.apply(client, ...args);
       }
-      // remove it from internal EE2; if none remain for the topic, it
-      // will be unsubscribed from.
-      client._topicEmitter.removeListener(eventify(topic), listener);
-      return client;
     }
   });
 
   /**
-   * Publishes a MQTT message if *not* an internal event.
-   * This breaks the contract where `emit` should return the number of
-   * listeners. We don't know the number of listeners, and the publishing
-   * process is async anyway (though, arguably, not with QoS 0, but...zalgo).
-   * @public
-   * @param {string} topic - MQTT topic (or event)
-   * @param {Buffer|string|ArrayBuffer} [message] - Message if MQTT; otherwise
-   *   optional if internal
-   * @param {Object} [opts] - MQTT options or more data for internal event
-   * @param {*} [...args] - More data for internal event (ignored by MQTT)
-   * @returns {Promise<string>|number} Topic, once publish happens (depends on
-   *   QoS), or just the number of listeners if internal (yes, this is bad)
+   * Publishes a message to a topic
+   * @function
+   * @returns {Promise<void>}
    */
-  client.emit = new Proxy(client.emit, {
-    apply(target, client, [topic, ...args]) {
-      // let original events pass thru
-      if (clientEvents.has(topic)) {
-        return target.apply(client, [topic, ...args]);
-      }
-      const [message, opts] = args;
-      return client.publish(topic, message, opts).then(() => topic);
+  client.publish = new Proxy(publish, {
+    async apply(target, client, ...args) {
+      return target.apply(client, ...args);
     }
   });
 
@@ -195,7 +122,7 @@ const toadpatch = client => {
    * where the real listeners for subscriptions are stored.
    */
   client.on('message', (topic, message, packet) => {
-    client._topicEmitter.emit(eventify(topic), message, packet);
+    client.toad.emit(eventify(topic), message, packet);
   });
 
   return client;
@@ -211,8 +138,8 @@ const toadpatch = client => {
 exports.connect = (...args) => {
   const client = MQTT.connect(...args);
   return new Promise((resolve, reject) => {
-    client.on('error', reject).on('connect', () => {
-      resolve(toadpatch(client));
+    client.on('error', reject).on('connect', connack => {
+      resolve(toadpatch(client, connack));
     });
   });
 };
diff --git a/package.json b/package.json
@@ -58,7 +58,7 @@
     "unexpected": "^10.36.0"
   },
   "engines": {
-    "node": ">=6.5"
+    "node": ">=7.0"
   },
   "bugs": {
     "url": "https://github.com/boneskull/mqttltoad/issues"
diff --git a/test/harness/index.js b/test/harness/index.js
diff --git a/test/mqttletoad.spec.js b/test/mqttletoad.spec.js
