fix(deps): update rxjs to v6

no issue
- update rxjs version to 6
- use rxjs correctly in yarn utils (we weren't using it correctly before)
- refactor tests to work better with observable usage

Author: acburdine

lib/tasks/yarn-install.js

@@ -61,15 +61,20 @@ module.exports = function yarnInstall(ui, zipFile) {
 
     tasks.push({
         title: 'Installing dependencies',
-        task: ctx => yarn(['install', '--no-emoji', '--no-progress'], {
-            cwd: ctx.installPath,
-            env: {NODE_ENV: 'production'},
-            observe: true
-        }).catch((error) => {
-            // Add error catcher so we can cleanup the install path if an error occurs
-            fs.removeSync(ctx.installPath);
-            return Promise.reject(error);
-        })
+        task: (ctx) => {
+            const observable = yarn(['install', '--no-emoji', '--no-progress'], {
+                cwd: ctx.installPath,
+                env: {NODE_ENV: 'production'},
+                observe: true
+            });
+
+            observable.subscribe({
+                // Add error catcher so we can cleanup the install path if an error occurs
+                error: () => fs.removeSync(ctx.installPath)
+            });
+
+            return observable;
+        }
     });
 
     return ui.listr(tasks, false);

lib/utils/yarn.js

@@ -1,8 +1,7 @@
 'use strict';
 const execa = require('execa');
-const Observable = require('rxjs').Observable;
-
-const errors = require('../errors');
+const {Observable} = require('rxjs');
+const {ProcessError} = require('../errors');
 
 /**
  * Runs a yarn command. Can return an Observer which allows
@@ -28,7 +27,7 @@ module.exports = function yarn(yarnArgs, options) {
         // some other properties, so we just pass
         // the entire error object in as options to
         // the ProcessError
-        return cp.catch(error => Promise.reject(new errors.ProcessError(error)));
+        return cp.catch(error => Promise.reject(new ProcessError(error)));
     }
 
     return new Observable((observer) => {
@@ -41,12 +40,11 @@ module.exports = function yarn(yarnArgs, options) {
         cp.then(() => {
             observer.complete();
         }).catch((error) => {
-            observer.error(error);
+            // execa augments the error object with
+            // some other properties, so we just pass
+            // the entire error object in as options to
+            // the ProcessError
+            observer.error(new ProcessError(error));
         });
-
-        // execa augments the error object with
-        // some other properties, so we just pass
-        // the entire error object in as options to
-        // the ProcessError
-    }).catch(error => Promise.reject(new errors.ProcessError(error)));
+    });
 };

package.json

@@ -76,7 +76,7 @@
     "prettyjson": "1.2.1",
     "read-last-lines": "1.6.0",
     "replace-in-file": "3.4.3",
-    "rxjs": "5.5.11",
+    "rxjs": "6.4.0",
     "semver": "5.6.0",
     "shasum": "1.0.2",
     "stat-mode": "0.3.0",

test/unit/tasks/yarn-install-spec.js

@@ -6,6 +6,7 @@ const Promise = require('bluebird');
 const {setupTestFolder, cleanupTestFolders} = require('../../utils/test-folder');
 const path = require('path');
 const fs = require('fs');
+const {Observable, isObservable} = require('rxjs');
 
 const modulePath = '../../../lib/tasks/yarn-install';
 const errors = require('../../../lib/errors');
@@ -16,7 +17,7 @@ describe('Unit: Tasks > yarn-install', function () {
     });
 
     it('base function calls subtasks and yarn util', function () {
-        const yarnStub = sinon.stub().resolves();
+        const yarnStub = sinon.stub().returns(new Observable(o => o.complete()));
         const yarnInstall = proxyquire(modulePath, {
             '../utils/yarn': yarnStub
         });
@@ -25,7 +26,10 @@ describe('Unit: Tasks > yarn-install', function () {
         const listrStub = sinon.stub().callsFake((tasks) => {
             expect(tasks).to.have.length(3);
 
-            return Promise.each(tasks, task => task.task(ctx));
+            return Promise.each(tasks, (task) => {
+                const result = task.task(ctx);
+                return isObservable(result) ? result.toPromise() : result;
+            });
         });
 
         const distTaskStub = sinon.stub(subTasks, 'dist').resolves();
@@ -67,7 +71,7 @@ describe('Unit: Tasks > yarn-install', function () {
     });
 
     it('catches errors from yarn and cleans up install folder', function () {
-        const yarnStub = sinon.stub().rejects(new Error('an error occurred'));
+        const yarnStub = sinon.stub().returns(new Observable(o => o.error(new Error('an error occurred'))));
         const yarnInstall = proxyquire(modulePath, {
             '../utils/yarn': yarnStub
         });
@@ -77,7 +81,10 @@ describe('Unit: Tasks > yarn-install', function () {
         const listrStub = sinon.stub().callsFake((tasks) => {
             expect(tasks).to.have.length(3);
 
-            return Promise.each(tasks, task => task.task(ctx));
+            return Promise.each(tasks, (task) => {
+                const result = task.task(ctx);
+                return isObservable(result) ? result.toPromise() : result;
+            });
         });
 
         const distTaskStub = sinon.stub(subTasks, 'dist').resolves();

test/unit/utils/yarn-spec.js

@@ -2,34 +2,31 @@
 const expect = require('chai').expect;
 const proxyquire = require('proxyquire');
 const sinon = require('sinon');
-const ProcessError = require('../../../lib/errors').ProcessError;
+const {isObservable} = require('rxjs');
+const {getReadableStream} = require('../../utils/stream');
+const {ProcessError} = require('../../../lib/errors');
 
 const modulePath = '../../../lib/utils/yarn';
 
-let yarn;
+const setup = proxies => proxyquire(modulePath, proxies);
 
 describe('Unit: yarn', function () {
     let currentEnv;
-    let execa;
 
     beforeEach(function () {
         currentEnv = process.env;
         process.env = {};
-
-        execa = sinon.stub().resolves();
-        yarn = proxyquire(modulePath, {
-            execa: execa
-        });
     });
 
     afterEach(function () {
         process.env = currentEnv;
     });
 
     it('spawns yarn process with no arguments correctly', function () {
-        const promise = yarn();
+        const execa = sinon.stub().resolves();
+        const yarn = setup({execa});
 
-        return promise.then(function () {
+        return yarn().then(function () {
             expect(execa.calledOnce).to.be.true;
             expect(execa.args[0]).to.be.ok;
             expect(execa.args[0]).to.have.lengthOf(3);
@@ -38,9 +35,10 @@ describe('Unit: yarn', function () {
     });
 
     it('spawns yarn process with correct arguments', function () {
-        const promise = yarn(['cache', 'clear']);
+        const execa = sinon.stub().resolves();
+        const yarn = setup({execa});
 
-        return promise.then(function () {
+        return yarn(['cache', 'clear']).then(function () {
             expect(execa.calledOnce).to.be.true;
             expect(execa.args[0]).to.be.ok;
             expect(execa.args[0]).to.have.lengthOf(3);
@@ -49,9 +47,10 @@ describe('Unit: yarn', function () {
     });
 
     it('correctly passes through options', function () {
-        const promise = yarn([], {cwd: 'test'});
+        const execa = sinon.stub().resolves();
+        const yarn = setup({execa});
 
-        return promise.then(function () {
+        return yarn([], {cwd: 'test'}).then(function () {
             expect(execa.calledOnce).to.be.true;
             expect(execa.args[0]).to.be.ok;
             expect(execa.args[0]).to.have.lengthOf(3);
@@ -61,11 +60,11 @@ describe('Unit: yarn', function () {
     });
 
     it('respects process.env overrides but doesn\'t mutate process.env', function () {
-        process.env.TESTENV = 'test';
-
-        const promise = yarn([], {env: {TESTENV: 'override'}});
+        const execa = sinon.stub().resolves();
+        const yarn = setup({execa});
 
-        return promise.then(() => {
+        process.env.TESTENV = 'test';
+        return yarn([], {env: {TESTENV: 'override'}}).then(() => {
             expect(execa.calledOnce).to.be.true;
             expect(execa.args[0][2]).to.be.an('object');
             expect(execa.args[0][2].env).to.be.an('object');
@@ -75,8 +74,8 @@ describe('Unit: yarn', function () {
     });
 
     it('fails gracefully when yarn fails', function () {
-        execa.rejects(new Error('YARN_TO_FAST'));
-        yarn = proxyquire(modulePath, {execa: execa});
+        const execa = sinon.stub().rejects(new Error('YARN_TO_FAST'));
+        const yarn = setup({execa});
 
         return yarn().then(() => {
             expect(false, 'Promise should have rejected').to.be.true;
@@ -88,105 +87,90 @@ describe('Unit: yarn', function () {
     });
 
     describe('can return observables', function () {
-        let stubs;
-        beforeEach(function () {
-            stubs = {
-                stdout: sinon.stub(),
-                observer: {
-                    next: sinon.stub(),
-                    complete: sinon.stub(),
-                    error: sinon.stub()
-                }
-            };
-
-            class TestClass {
-                constructor(fn) {
-                    stubs.proxy = {fn: fn};
-                    return (new Promise((resolve,reject) => {
-                        stubs.proxy.resolve = resolve;
-                        stubs.proxy.reject = reject;
-                    }));
-                }
-            }
-
-            const execaPromise = new Promise((resolve, reject) => {
-                stubs._execa = {
-                    resolve: resolve,
-                    reject: reject
-                };
+        it('ends properly', function () {
+            const execa = sinon.stub().callsFake(() => {
+                const promise = Promise.resolve();
+                promise.stdout = getReadableStream();
+                return promise;
             });
-            execaPromise.stdout = {
-                setEncoding: () => true,
-                on: stubs.stdout
+            const yarn = setup({execa});
+
+            const res = yarn([], {observe: true});
+            expect(isObservable(res)).to.be.true;
+
+            const subscriber = {
+                next: sinon.stub(),
+                error: sinon.stub(),
+                complete: sinon.stub()
             };
-            execa.returns(execaPromise);
 
-            yarn = proxyquire(modulePath, {
-                execa: execa,
-                rxjs: {Observable: TestClass}
+            res.subscribe(subscriber);
+
+            return res.toPromise().then(() => {
+                expect(execa.calledOnce).to.be.true;
+                expect(subscriber.next.called).to.be.false;
+                expect(subscriber.error.called).to.be.false;
+                expect(subscriber.complete.calledOnce).to.be.true;
             });
         });
 
-        it('ends properly', function () {
+        it('ends properly (error)', function () {
+            const execa = sinon.stub().callsFake(() => {
+                const promise = Promise.reject(new Error('test error'));
+                promise.stdout = getReadableStream();
+                return promise;
+            });
+            const yarn = setup({execa});
+
             const res = yarn([], {observe: true});
-            expect(stubs.proxy).to.be.an('Object');
+            expect(isObservable(res)).to.be.true;
 
-            stubs.proxy.fn(stubs.observer);
-            stubs._execa.resolve();
-            res.then(() => {
-                expect(stubs.observer.complete.calledOnce).to.be.true;
-            });
+            const subscriber = {
+                next: sinon.stub(),
+                error: sinon.stub(),
+                complete: sinon.stub()
+            };
 
-            stubs.proxy.resolve();
-            return res;
-        });
+            res.subscribe(subscriber);
 
-        it('ends properly (error)', function () {
-            const res = yarn([], {observe: true});
-            expect(stubs.proxy).to.be.an('Object');
-            stubs.proxy.fn(stubs.observer);
-            stubs._execa.reject('test');
-
-            res.then(() => {
-                expect(stubs.observer.complete.called).to.be.false;
-                expect(stubs.observer.error.calledOnce).to.be.true;
-                expect(stubs.observer.error.args[0][0]).to.equal('test');
+            return res.toPromise().catch((error) => {
+                expect(error.message).to.equal('test error');
+                expect(execa.calledOnce).to.be.true;
+                expect(subscriber.next.called).to.be.false;
+                expect(subscriber.error.calledOnce).to.be.true;
+                expect(subscriber.error.args[0][0]).to.be.an.instanceOf(ProcessError);
+                expect(subscriber.complete.called).to.be.false;
             });
-
-            stubs.proxy.resolve();
-            return res;
         });
 
-        it('proxies data through', function () {
-            yarn([], {observe: true});
-            expect(stubs.proxy).to.be.an('Object');
-            stubs.proxy.fn(stubs.observer);
-            expect(stubs.stdout.calledOnce).to.be.true;
-            expect(stubs.stdout.args[0][0]).to.equal('data');
-
-            const onFn = stubs.stdout.args[0][1];
-            onFn('\n\n\n');
-            onFn('test\n');
-            onFn('\nbest\n');
-            onFn('run');
-
-            const next = stubs.observer.next;
-            expect(next.callCount).to.equal(4);
-            expect(next.args[0][0]).to.equal('\n\n');
-            expect(next.args[1][0]).to.equal('test');
-            expect(next.args[2][0]).to.equal('\nbest');
-            expect(next.args[3][0]).to.equal('run');
-        });
+        it('passes data through', function () {
+            const execa = sinon.stub().callsFake(() => {
+                const promise = Promise.resolve();
+                promise.stdout = getReadableStream(function () {
+                    this.push('test message\n');
+                    this.push(null);
+                });
+                return promise;
+            });
+            const yarn = setup({execa});
 
-        it('cleans up observer error', function () {
             const res = yarn([], {observe: true});
-            expect(stubs.proxy).to.be.an('Object');
-            stubs.proxy.reject();
-            return res.then(() => {
-                expect(false, 'Promise should have rejected').to.be.true;
-            }).catch((e) => {
-                expect(e).to.be.ok;
-                expect(e).to.be.instanceOf(ProcessError);
+            expect(isObservable(res)).to.be.true;
+
+            const subscriber = {
+                next: sinon.stub(),
+                error: sinon.stub(),
+                complete: sinon.stub()
+            };
+
+            res.subscribe(subscriber);
+
+            return res.toPromise().then(() => {
+                expect(execa.calledOnce).to.be.true;
+                expect(subscriber.next.calledOnce).to.be.true;
+                expect(subscriber.next.calledWithExactly('test message')).to.be.true;
+                expect(subscriber.error.called).to.be.false;
+                expect(subscriber.complete.calledOnce).to.be.true;
             });
         });
     });