Skip to content

Commit

Permalink
fix(change-stream): fix change stream resuming with promises
Browse files Browse the repository at this point in the history
If a change stream resume occurred, and the user was using the
Promise-based version of .next, the promise would resolve before
waiting for the change stream to resume, resulting in bad behavior.

Fixes NODE-1493
  • Loading branch information
daprahamian committed Jun 6, 2018
1 parent 844c2c8 commit 3063f00
Showing 1 changed file with 21 additions and 19 deletions.
40 changes: 21 additions & 19 deletions lib/change_stream.js
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,13 @@ var createChangeStreamCursor = function(self) {
self.emit('error', error);
});

if (self.pipeDestinations) {
const cursorStream = changeStreamCursor.stream(self.streamOptions);
for (let pipeDestination in self.pipeDestinations) {
cursorStream.pipe(pipeDestination);
}
}

return changeStreamCursor;
};

Expand Down Expand Up @@ -303,27 +310,22 @@ var processNewChange = function(self, err, change, callback) {
// Handle resumable MongoNetworkErrors
if (isResumableError(err) && !self.attemptingResume) {
self.attemptingResume = true;
return self.cursor.close(function(closeErr) {
if (closeErr) {
if (callback) return callback(err, null);
return self.promiseLibrary.reject(err);
}

// Establish a new cursor
self.cursor = createChangeStreamCursor(self);

// Attempt to reconfigure piping
if (self.pipeDestinations) {
var cursorStream = self.cursor.stream(self.streamOptions);
for (var pipeDestination in self.pipeDestinations) {
cursorStream.pipe(pipeDestination);
if (callback) {
return self.cursor.close(function(closeErr) {
if (closeErr) {
return callback(err, null);
}
}

// Attempt the next() operation again
if (callback) return self.next(callback);
return self.next();
});
self.cursor = createChangeStreamCursor(self);

return self.next(callback);
});
}

return self.cursor
.close()
.then(() => (self.cursor = createChangeStreamCursor(self)))
.then(() => self.next());
}

if (typeof callback === 'function') return callback(err, null);
Expand Down

0 comments on commit 3063f00

Please sign in to comment.