Merge branch 'mohd-akram-native-promise'

Author: cressie176

channel_api.js

@@ -1,11 +1,11 @@
 var raw_connect = require('./lib/connect').connect;
 var ChannelModel = require('./lib/channel_model').ChannelModel;
-var Promise = require('bluebird');
+var promisify = require('util').promisify;
 
 function connect(url, connOptions) {
-  return Promise.fromCallback(function(cb) {
+  return promisify(function(cb) {
     return raw_connect(url, connOptions, cb);
-  })
+  })()
   .then(function(conn) {
     return new ChannelModel(conn);
   });

examples/tutorials/receive_logs_direct.js

@@ -1,7 +1,6 @@
 #!/usr/bin/env node
 
 var amqp = require('amqplib');
-var all = require('bluebird').all;
 var basename = require('path').basename;
 
 var severities = process.argv.slice(2);
@@ -24,7 +23,7 @@ amqp.connect('amqp://localhost').then(function(conn) {
 
     ok = ok.then(function(qok) {
       var queue = qok.queue;
-      return all(severities.map(function(sev) {
+      return Promise.all(severities.map(function(sev) {
         ch.bindQueue(queue, ex, sev);
       })).then(function() { return queue; });
     });

examples/tutorials/receive_logs_topic.js

@@ -2,7 +2,6 @@
 
 var amqp = require('amqplib');
 var basename = require('path').basename;
-var all = require('bluebird').all;
 
 var keys = process.argv.slice(2);
 if (keys.length < 1) {
@@ -23,7 +22,7 @@ amqp.connect('amqp://localhost').then(function(conn) {
 
     ok = ok.then(function(qok) {
       var queue = qok.queue;
-      return all(keys.map(function(rk) {
+      return Promise.all(keys.map(function(rk) {
         ch.bindQueue(queue, ex, rk);
       })).then(function() { return queue; });
     });

examples/tutorials/rpc_client.js

@@ -2,7 +2,6 @@
 
 var amqp = require('amqplib');
 var basename = require('path').basename;
-var Promise = require('bluebird');
 var uuid = require('node-uuid');
 
 // I've departed from the form of the original RPC tutorial, which

lib/callback_model.js

@@ -5,7 +5,6 @@
 'use strict';
 
 var defs = require('./defs');
-var Promise = require('bluebird');
 var inherits = require('util').inherits;
 var EventEmitter = require('events').EventEmitter;
 var BaseChannel = require('./channel').BaseChannel;

lib/channel_model.js

@@ -5,7 +5,7 @@
 'use strict';
 
 const EventEmitter = require('events');
-const Promise = require('bluebird');
+const promisify = require('util').promisify;
 const defs = require('./defs');
 const {BaseChannel} = require('./channel');
 const {acceptMessage} = require('./channel');
@@ -23,7 +23,7 @@ class ChannelModel extends EventEmitter {
   }
 
   close() {
-    return Promise.fromCallback(this.connection.close.bind(this.connection));
+    return promisify(this.connection.close.bind(this.connection))();
   }
 
   async createChannel() {
@@ -53,27 +53,25 @@ class Channel extends BaseChannel {
   // response's fields; this is intended to be suitable for implementing
   // API procedures.
   async rpc(method, fields, expect) {
-    const f = await Promise.fromCallback(cb => {
+    const f = await promisify(cb => {
       return this._rpc(method, fields, expect, cb);
-    })
+    })();
 
     return f.fields;
   }
 
   // Do the remarkably simple channel open handshake
-  open() {
-    return Promise.try(this.allocate.bind(this)).then(
-      ch => {
-        return ch.rpc(defs.ChannelOpen, {outOfBand: ""},
-                      defs.ChannelOpenOk);
-      });
+  async open() {
+    const ch = await this.allocate.bind(this)();
+    return ch.rpc(defs.ChannelOpen, {outOfBand: ""},
+                  defs.ChannelOpenOk);
   }
 
   close() {
-    return Promise.fromCallback(cb => {
+    return promisify(cb => {
       return this.closeBecause("Goodbye", defs.constants.REPLY_SUCCESS,
                       cb);
-    });
+    })();
   }
 
   // === Public API, declaring queues and stuff ===
@@ -162,21 +160,21 @@ class Channel extends BaseChannel {
     // 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 => {
-      this._rpc(defs.BasicConsume, fields, defs.BasicConsumeOk, cb);
-    })
-    .then(ok => {
-      this.registerConsumer(ok.fields.consumerTag, callback);
-      return ok.fields;
+    return new Promise((resolve, reject) => {
+      this._rpc(defs.BasicConsume, fields, defs.BasicConsumeOk, (err, ok) => {
+        if (err) return reject(err);
+        this.registerConsumer(ok.fields.consumerTag, callback);
+        resolve(ok.fields);
+      });
     });
   }
 
   async cancel(consumerTag) {
-    const ok = await Promise.fromCallback(cb => {
+    const ok = await promisify(cb => {
       this._rpc(defs.BasicCancel, Args.cancel(consumerTag),
             defs.BasicCancelOk,
             cb);
-    })
+    })()
     .then(ok => {
       this.unregisterConsumer(consumerTag);
       return ok.fields;
@@ -185,25 +183,23 @@ class Channel extends BaseChannel {
 
   get(queue, options) {
     const fields = Args.get(queue, options);
-    return Promise.fromCallback(cb => {
-      return 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 => {
+    return new Promise((resolve, reject) => {
+      this.sendOrEnqueue(defs.BasicGet, fields, (err, f) => {
+        if (err) return reject(err);
+        if (f.id === defs.BasicGetEmpty) {
+          return resolve(false);
+        }
+        else if (f.id === defs.BasicGetOk) {
+          const fields = f.fields;
           this.handleMessage = acceptMessage(m => {
             m.fields = fields;
             resolve(m);
           });
-        });
-      }
-      else {
-        throw new Error(`Unexpected response to BasicGet: ${inspect(f)}`);
-      }
+        }
+        else {
+          reject(new Error(`Unexpected response to BasicGet: ${inspect(f)}`));
+        }
+      });
     });
   }
 
@@ -290,6 +286,13 @@ class ConfirmChannel extends Channel {
         awaiting.push(confirmed);
       }
     });
+    // Channel closed
+    if (!this.pending) {
+      var cb;
+      while (cb = this.unconfirmed.shift()) {
+        if (cb) cb(new Error('channel closed'));
+      }
+    }
     return Promise.all(awaiting);
   }
 }