feat: switch to native Promise

amqp-node · May 18, 2021 · 1e59be1 · 1e59be1
1 parent d87348d
commit 1e59be1
Show file tree

Hide file tree

Showing 10 changed files with 138 additions and 72 deletions.
diff --git a/.travis.yml b/.travis.yml
@@ -1,7 +1,5 @@
 language: node_js
 node_js:
-- "0.8"
-- "0.10"
 - "0.12"
 - "iojs-v1"
 - "iojs-v2"

diff --git a/channel_api.js b/channel_api.js
@@ -1,10 +1,15 @@
 var raw_connect = require('./lib/connect').connect;
 var ChannelModel = require('./lib/channel_model').ChannelModel;
-var Promise = require('bluebird');
 
 function connect(url, connOptions) {
-  return Promise.fromCallback(function(cb) {
-    return raw_connect(url, connOptions, cb);
+  return new Promise(function (resolve, reject) {
+    raw_connect(url, connOptions, function (err, result) {
+      if (err) {
+        reject(err);
+      } else {
+        resolve(result);
+      }
+    })
   })
   .then(function(conn) {
     return new ChannelModel(conn);

diff --git a/examples/tutorials/receive_logs_direct.js b/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; });
     });

diff --git a/examples/tutorials/receive_logs_topic.js b/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; });
     });

diff --git a/examples/tutorials/rpc_client.js b/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

diff --git a/lib/callback_model.js b/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;

diff --git a/lib/channel_model.js b/lib/channel_model.js
@@ -5,7 +5,7 @@
 'use strict';
 
 var defs = require('./defs');
-var Promise = require('bluebird');
+var Bluebird = require('bluebird');
 var inherits = require('util').inherits;
 var EventEmitter = require('events').EventEmitter;
 var BaseChannel = require('./channel').BaseChannel;
@@ -29,7 +29,17 @@ module.exports.ChannelModel = ChannelModel;
 var CM = ChannelModel.prototype;
 
 CM.close = function() {
-  return Promise.fromCallback(this.connection.close.bind(this.connection));
+  var close = this.connection.close.bind(this.connection);
+
+  return new Promise(function (resolve, reject) {
+    close(function (err, result) {
+      if (err) {
+        reject(err);
+      } else {
+        resolve(result);
+      }
+    });
+  });
 };
 
 // Channels
@@ -55,17 +65,31 @@ var C = Channel.prototype;
 // API procedures.
 C.rpc = function(method, fields, expect) {
   var self = this;
-  return Promise.fromCallback(function(cb) {
-    return self._rpc(method, fields, expect, cb);
-  })
-  .then(function(f) {
+
+  return new Promise(function (resolve, reject) {
+    self._rpc(method, fields, expect, function (err, result) {
+      if (err) {
+        reject(err);
+      } else {
+        resolve(result);
+      }
+    });
+  }).then(function (f) {
     return f.fields;
   });
 };
 
 // Do the remarkably simple channel open handshake
 C.open = function() {
-  return Promise.try(this.allocate.bind(this)).then(
+  var allocate = this.allocate.bind(this);
+
+  return new Promise(function (resolve, reject) {
+    try {
+      resolve(allocate());
+    } catch (e) {
+      reject(e);
+    }
+  }).then(
     function(ch) {
       return ch.rpc(defs.ChannelOpen, {outOfBand: ""},
                     defs.ChannelOpenOk);
@@ -74,10 +98,16 @@ C.open = function() {
 
 C.close = function() {
   var self = this;
-  return Promise.fromCallback(function(cb) {
+
+  return Bluebird.fromCallback(function(cb) {
     return self.closeBecause("Goodbye", defs.constants.REPLY_SUCCESS,
                     cb);
   });
+  // return new Promise(function(resolve, reject) {
+  //   return self.closeBecause("Goodbye", defs.constants.REPLY_SUCCESS,
+  //     (err, result) => err ? reject(err) : resolve(result)
+  //   );
+  // });
 };
 
 // === Public API, declaring queues and stuff ===
@@ -167,9 +197,18 @@ C.consume = function(queue, callback, options) {
   // NB we want the callback to be run synchronously, so that we've
   // registered the consumerTag before any messages can arrive.
   var fields = Args.consume(queue, options);
-  return Promise.fromCallback(function(cb) {
+  return Bluebird.fromCallback(function(cb) {
     self._rpc(defs.BasicConsume, fields, defs.BasicConsumeOk, cb);
   })
+  // return new Promise(function(resolve, reject) {
+  //     self._rpc(defs.BasicConsume, fields, defs.BasicConsumeOk, function (err, result) {
+  //     if (err) {
+  //       reject(err);
+  //     } else {
+  //       resolve(result);
+  //     }
+  //   });
+  // })
   .then(function(ok) {
     self.registerConsumer(ok.fields.consumerTag, callback);
     return ok.fields;
@@ -178,10 +217,16 @@ C.consume = function(queue, callback, options) {
 
 C.cancel = function(consumerTag) {
   var self = this;
-  return Promise.fromCallback(function(cb) {
+  return new Promise(function(resolve, reject) {
     self._rpc(defs.BasicCancel, Args.cancel(consumerTag),
           defs.BasicCancelOk,
-          cb);
+      function (err, result) {
+        if (err) {
+          reject(err);
+        } else {
+          resolve(result);
+        }
+      });
   })
   .then(function(ok) {
     self.unregisterConsumer(consumerTag);
@@ -192,9 +237,18 @@ C.cancel = function(consumerTag) {
 C.get = function(queue, options) {
   var self = this;
   var fields = Args.get(queue, options);
-  return Promise.fromCallback(function(cb) {
+  return Bluebird.fromCallback(function(cb) {
     return self.sendOrEnqueue(defs.BasicGet, fields, cb);
   })
+  // return new Promise(function(resolve, reject) {
+  //   return self.sendOrEnqueue(defs.BasicGet, fields, function (err, result) {
+  //     if (err) {
+  //       reject(err);
+  //     } else {
+  //       resolve(result);
+  //     }
+  //   });
+  // })
   .then(function(f) {
     if (f.id === defs.BasicGetEmpty) {
       return false;

diff --git a/test/channel.js b/test/channel.js
@@ -3,7 +3,6 @@
 'use strict';
 
 var assert = require('assert');
-var Promise = require('bluebird');
 var Channel = require('../lib/channel').Channel;
 var Connection = require('../lib/connection').Connection;
 var util = require('./util');
@@ -77,11 +76,19 @@ var DELIVER_FIELDS = {
 };
 
 function open(ch) {
-  return Promise.try(function() {
-    ch.allocate();
-    return Promise.fromCallback(function(cb) {
-      ch._rpc(defs.ChannelOpen, {outOfBand: ''}, defs.ChannelOpenOk, cb);
-    });
+  return new Promise(function (resolve, reject) {
+    try {
+      ch.allocate();
+      ch._rpc(defs.ChannelOpen, {outOfBand: ''}, defs.ChannelOpenOk, function (err, done) {
+        if (err) {
+          reject(err);
+        } else {
+          resolve(done);
+        }
+      });
+    } catch (e) {
+      reject(e);
+    }
   });
 }
 
@@ -286,7 +293,7 @@ test("RPC on closed channel", channelTest(
         failureCb(resolve, reject));
     });
 
-    Promise.join(close, fail1, fail2)
+    Promise.all([close, fail1, fail2])
       .then(succeed(done))
       .catch(fail(done));
   },