Merge 2e79a21 into 507621c

lib/datasource.js

@@ -130,6 +130,7 @@ function DataSource(name, settings, modelBuilder) {
   this.models = this.modelBuilder.models;
   this.definitions = this.modelBuilder.definitions;
   this.juggler = juggler;
+  this._queuedInvocations = 0;
 
   // operation metadata
   // Initialize it before calling setup as the connector might register operations
@@ -200,6 +201,18 @@ DataSource.prototype._setupConnector = function() {
       this.connector.dataSource = this;
     }
     const dataSource = this;
+
+    // Set max listeners to a default value
+    // Override this default value with a datasource setting
+    // 'maxOfflineRequests' from an application's datasources.json
+    // solves bug https://github.com/strongloop/loopback-next/issues/2198
+    let maxOfflineRequests = 16;
+    if (this.settings && this.settings.maxOfflineRequests) {
+      if (this.settings.maxOfflineRequests > maxOfflineRequests)
+        maxOfflineRequests = this.settings.maxOfflineRequests;
+    }
+    dataSource.setMaxListeners(maxOfflineRequests);
+
     this.connector.log = function(query, start) {
       dataSource.log(query, start);
     };
@@ -471,18 +484,22 @@ DataSource.prototype.setup = function(dsName, settings) {
       if (this.connected) {
         debug('DataSource %s is now connected to %s', this.name, this.connector.name);
         this.emit('connected');
-      } else {
+      } else if (err) {
         // The connection fails, let's report it and hope it will be recovered in the next call
-        if (err) {
-          // Reset the connecting to `false`
-          this.connecting = false;
+        // Reset the connecting to `false`
+        this.connecting = false;
+        if (this._queuedInvocations) {
+          // Another operation is already waiting for connect() result,
+          // let them handle the connection error.
+          debug('Connection fails: %s\nIt will be retried for the next request.', err);
+        } else {
           g.error('Connection fails: %s\nIt will be retried for the next request.', err);
           this.emit('error', err);
-        } else {
-          // Either lazyConnect or connector initialize() defers the connection
-          debug('DataSource %s will be connected to connector %s', this.name,
-            this.connector.name);
         }
+      } else {
+        // Either lazyConnect or connector initialize() defers the connection
+        debug('DataSource %s will be connected to connector %s', this.name,
+          this.connector.name);
       }
     }.bind(this);
 
@@ -1047,6 +1064,14 @@ DataSource.prototype.automigrate = function(models, cb) {
     }
   }
 
+  const args = [models, cb];
+  args.callee = this.automigrate;
+  const queued = this.ready(this, args);
+  if (queued) {
+    // waiting to connect
+    return cb.promise;
+  }
+
   this.connector.automigrate(models, cb);
   return cb.promise;
 };
@@ -1108,6 +1133,14 @@ DataSource.prototype.autoupdate = function(models, cb) {
     }
   }
 
+  const args = [models, cb];
+  args.callee = this.autoupdate;
+  const queued = this.ready(this, args);
+  if (queued) {
+    // waiting to connect
+    return cb.promise;
+  }
+
   this.connector.autoupdate(models, cb);
   return cb.promise;
 };
@@ -2508,12 +2541,15 @@ function(obj, args) {
     return false;
   }
 
+  this._queuedInvocations++;
+
   const method = args.callee;
   // Set up a callback after the connection is established to continue the method call
 
   let onConnected = null, onError = null, timeoutHandle = null;
   onConnected = function() {
     debug('Datasource %s is now connected - executing method %s', self.name, method.name);
+    this._queuedInvocations--;
     // Remove the error handler
     self.removeListener('error', onError);
     if (timeoutHandle) {
@@ -2536,6 +2572,7 @@ function(obj, args) {
   };
   onError = function(err) {
     debug('Datasource %s fails to connect - aborting method %s', self.name, method.name);
+    this._queuedInvocations--;
     // Remove the connected listener
     self.removeListener('connected', onConnected);
     if (timeoutHandle) {
@@ -2557,6 +2594,7 @@ function(obj, args) {
   timeoutHandle = setTimeout(function() {
     debug('Datasource %s fails to connect due to timeout - aborting method %s',
       self.name, method.name);
+    this._queuedInvocations--;
     self.connecting = false;
     self.removeListener('error', onError);
     self.removeListener('connected', onConnected);
@@ -2569,7 +2607,11 @@ function(obj, args) {
 
   if (!this.connecting) {
     debug('Connecting datasource %s to connector %s', this.name, this.connector.name);
-    this.connect();
+    // When no callback is provided to `connect()`, it returns a Promise.
+    // We are not handling that promise and thus UnhandledPromiseRejection
+    // warning is triggered when the connection could not be established.
+    // We are avoiding this problem by providing a no-op callback.
+    this.connect(() => {});
   }
   return true;
 };

test/datasource.test.js

@@ -352,4 +352,80 @@ describe('DataSource', function() {
         .should.not.containEql('TestModel');
     });
   });
+
+  describe('automigrate', () => {
+    it('reports connection errors (immediate connect)', async () => {
+      const dataSource = new DataSource({
+        connector: givenConnectorFailingOnConnect(),
+      });
+      dataSource.define('MyModel');
+      await dataSource.automigrate().should.be.rejectedWith(/test failure/);
+    });
+
+    it('reports connection errors (lazy connect)', () => {
+      const dataSource = new DataSource({
+        connector: givenConnectorFailingOnConnect(),
+        lazyConnect: true,
+      });
+      dataSource.define('MyModel');
+      return dataSource.automigrate().should.be.rejectedWith(/test failure/);
+    });
+
+    function givenConnectorFailingOnConnect() {
+      return givenMockConnector({
+        connect: function(cb) {
+          process.nextTick(() => cb(new Error('test failure')));
+        },
+        automigrate: function(models, cb) {
+          cb(new Error('automigrate should not have been called'));
+        },
+      });
+    }
+  });
+
+  describe('autoupdate', () => {
+    it('reports connection errors (immediate connect)', async () => {
+      const dataSource = new DataSource({
+        connector: givenConnectorFailingOnConnect(),
+      });
+      dataSource.define('MyModel');
+      await dataSource.autoupdate().should.be.rejectedWith(/test failure/);
+    });
+
+    it('reports connection errors (lazy connect)', () => {
+      const dataSource = new DataSource({
+        connector: givenConnectorFailingOnConnect(),
+        lazyConnect: true,
+      });
+      dataSource.define('MyModel');
+      return dataSource.autoupdate().should.be.rejectedWith(/test failure/);
+    });
+
+    function givenConnectorFailingOnConnect() {
+      return givenMockConnector({
+        connect: function(cb) {
+          process.nextTick(() => cb(new Error('test failure')));
+        },
+        autoupdate: function(models, cb) {
+          cb(new Error('autoupdate should not have been called'));
+        },
+      });
+    }
+  });
 });
+
+function givenMockConnector(props) {
+  const connector = {
+    name: 'loopback-connector-mock',
+    initialize: function(ds, cb) {
+      ds.connector = connector;
+      if (ds.settings.lazyConnect) {
+        cb(null, false);
+      } else {
+        connector.connect(cb);
+      }
+    },
+    ...props,
+  };
+  return connector;
+}