start using async where it may fit

Author: jimmywarting

lib/channel_model.js

@@ -26,19 +26,17 @@ class ChannelModel extends EventEmitter {
     return Promise.fromCallback(this.connection.close.bind(this.connection));
   }
 
-  createChannel() {
-    const c = new Channel(this.connection);
-    return c.open().then(openOk => { return c; });
+  async createChannel() {
+    const channel = new Channel(this.connection);
+    await channel.open();
+    return channel;
   }
 
-  createConfirmChannel() {
-    const c = new ConfirmChannel(this.connection);
-    return c.open()
-      .then(openOk => {
-        return c.rpc(defs.ConfirmSelect, {nowait: false},
-                     defs.ConfirmSelectOk)
-      })
-      .then(() => { return c; });
+  async createConfirmChannel() {
+    const channel = new ConfirmChannel(this.connection);
+    await channel.open();
+    await channel.rpc(defs.ConfirmSelect, {nowait: false}, defs.ConfirmSelectOk);
+    return channel;
   }
 }
 
@@ -54,13 +52,12 @@ class Channel extends BaseChannel {
   // An RPC that returns a 'proper' promise, which resolves to just the
   // response's fields; this is intended to be suitable for implementing
   // API procedures.
-  rpc(method, fields, expect) {
-    return Promise.fromCallback(cb => {
+  async rpc(method, fields, expect) {
+    const f = await Promise.fromCallback(cb => {
       return this._rpc(method, fields, expect, cb);
     })
-    .then(f => {
-      return f.fields;
-    });
+
+    return f.fields;
   }
 
   // Do the remarkably simple channel open handshake
@@ -161,53 +158,50 @@ class Channel extends BaseChannel {
     return this.publish('', queue, content, options);
   }
 
-  consume(queue, callback, options) {
+  async consume(queue, callback, options) {
     // NB we want the callback to be run synchronously, so that we've
     // registered the consumerTag before any messages can arrive.
     const fields = Args.consume(queue, options);
-    return Promise.fromCallback(cb => {
+    const ok = await Promise.fromCallback(cb => {
       this._rpc(defs.BasicConsume, fields, defs.BasicConsumeOk, cb);
     })
-    .then(ok => {
-      this.registerConsumer(ok.fields.consumerTag, callback);
-      return ok.fields;
-    });
+
+    this.registerConsumer(ok.fields.consumerTag, callback);
+    return ok.fields;
   }
 
-  cancel(consumerTag) {
-    return Promise.fromCallback(cb => {
+  async cancel(consumerTag) {
+    const ok = await Promise.fromCallback(cb => {
       this._rpc(defs.BasicCancel, Args.cancel(consumerTag),
             defs.BasicCancelOk,
             cb);
     })
-    .then(ok => {
-      this.unregisterConsumer(consumerTag);
-      return ok.fields;
-    });
+
+    this.unregisterConsumer(consumerTag);
+    return ok.fields;
   }
 
-  get(queue, options) {
+  async get(queue, options) {
     const fields = Args.get(queue, options);
-    return Promise.fromCallback(cb => {
-      return this.sendOrEnqueue(defs.BasicGet, fields, cb);
+    const f = await Promise.fromCallback(cb => {
+      this.sendOrEnqueue(defs.BasicGet, fields, cb);
     })
-    .then(f => {
-      if (f.id === defs.BasicGetEmpty) {
-        return false;
-      }
-      else if (f.id === defs.BasicGetOk) {
-        const fields = f.fields;
-        return new Promise(resolve => {
-          this.handleMessage = acceptMessage(m => {
-            m.fields = fields;
-            resolve(m);
-          });
+
+    if (f.id === defs.BasicGetEmpty) {
+      return false;
+    }
+    else if (f.id === defs.BasicGetOk) {
+      const fields = f.fields;
+      return new Promise(resolve => {
+        this.handleMessage = acceptMessage(m => {
+          m.fields = fields;
+          resolve(m);
         });
-      }
-      else {
-        throw new Error(`Unexpected response to BasicGet: ${inspect(f)}`);
-      }
-    });
+      });
+    }
+    else {
+      throw new Error(`Unexpected response to BasicGet: ${inspect(f)}`);
+    }
   }
 
   ack(message, allUpTo) {
@@ -246,18 +240,20 @@ class Channel extends BaseChannel {
                     Args.recover(),
                     defs.BasicRecoverOk);
   }
+
+  qos(count, global) {
+    return this.rpc(defs.BasicQos,
+                    Args.prefetch(count, global),
+                    defs.BasicQosOk);
+  }
 }
 
 // There are more options in AMQP than exposed here; RabbitMQ only
 // implements prefetch based on message count, and only for individual
 // channels or consumers. RabbitMQ v3.3.0 and after treat prefetch
 // (without `global` set) as per-consumer (for consumers following),
 // and prefetch with `global` set as per-channel.
-Channel.prototype.prefetch = Channel.prototype.qos = function(count, global) {
-  return this.rpc(defs.BasicQos,
-                  Args.prefetch(count, global),
-                  defs.BasicQosOk);
-};
+Channel.prototype.prefetch = Channel.prototype.qos
 
 // Confirm channel. This is a channel with confirms 'switched on',
 // meaning sent messages will provoke a responding 'ack' or 'nack'
@@ -280,8 +276,7 @@ class ConfirmChannel extends Channel {
     const awaiting = [];
     const unconfirmed = this.unconfirmed;
     unconfirmed.forEach((val, index) => {
-      if (val === null); // already confirmed
-      else {
+      if (val !== null) {
         const confirmed = new Promise((resolve, reject) => {
           unconfirmed[index] = err => {
             if (val) val(err);