Upgrade to RxJS 5

Summary:This change upgrades us to the RxJS 5 beta. It includes:

* Update Flow definitions to reflect new APIs
* Add `DisposableSubscription` and `CompositeSubscription` for dealing with new Subscription API (See also ReactiveX/rxjs#1583)
* Add `bufferUntil` to replace some `buffer()` usages (See ReactiveX/rxjs#1610)
* Remove rx-dom
* Update package.json
* Update all code to use new APIs

Reviewed By: peterhal

Differential Revision: D3131543

fb-gh-sync-id: 437c7b90d6e1fd8e463a57eb341146bbc4e9bbd3
fbshipit-source-id: 437c7b90d6e1fd8e463a57eb341146bbc4e9bbd3

.flowconfig

@@ -6,6 +6,7 @@
 .*/node_modules/babel-types/.*
 # ignore module src to prefer declaration
 .*/node_modules/rx/.*
+.*/node_modules/@reactivex/rxjs/.*
 # ignore module src to prefer declaration
 .*/node_modules/react/react.js
 # Annotated with `@flow` but has errors

npm-shrinkwrap.production.json

@@ -2,6 +2,11 @@
   "name": "nuclide",
   "version": "0.130.0",
   "dependencies": {
+    "@reactivex/rxjs": {
+      "version": "5.0.0-beta.6",
+      "from": "@reactivex/[email protected]",
+      "resolved": "https://registry.npmjs.org/@reactivex/rxjs/-/rxjs-5.0.0-beta.6.tgz"
+    },
     "acorn": {
       "version": "1.2.2",
       "from": "acorn@>=1.0.3 <2.0.0",
@@ -1261,16 +1266,6 @@
       "from": "[email protected]",
       "resolved": "https://registry.npmjs.org/rimraf/-/rimraf-2.5.2.tgz"
     },
-    "rx": {
-      "version": "3.1.1",
-      "from": "[email protected]",
-      "resolved": "https://registry.npmjs.org/rx/-/rx-3.1.1.tgz"
-    },
-    "rx-dom": {
-      "version": "7.0.3",
-      "from": "[email protected]",
-      "resolved": "https://registry.npmjs.org/rx-dom/-/rx-dom-7.0.3.tgz"
-    },
     "sax": {
       "version": "1.2.1",
       "from": "sax@>=0.6.0",

package.json

@@ -36,6 +36,7 @@
     "test": "npm run lint-all && npm run flow"
   },
   "dependencies": {
+    "@reactivex/rxjs": "5.0.0-beta.6",
     "atom-package-deps": "4.0.1",
     "babel-core": "5.8.38",
     "babel-runtime": "5.8.38",
@@ -68,8 +69,6 @@
     "react-for-atom": "0.14.7-0",
     "request": "2.70.0",
     "rimraf": "2.5.2",
-    "rx": "3.1.1",
-    "rx-dom": "7.0.3",
     "season": "5.3.0",
     "semver": "5.1.0",
     "shallowequal": "0.2.2",

pkg/nuclide-adb-logcat/lib/Activation.js

@@ -15,7 +15,7 @@ import {createProcessStream} from './createProcessStream';
 import createMessageStream from './createMessageStream';
 import {LogTailer} from '../../nuclide-console/lib/LogTailer';
 import {CompositeDisposable, Disposable} from 'atom';
-import Rx from 'rx';
+import Rx from '@reactivex/rxjs';
 
 const NOENT_ERROR_DESCRIPTION = `**Troubleshooting Tips**
 1. Make sure that adb is installed
@@ -42,21 +42,23 @@ class Activation {
               0,
             )
           ))
-          .tapOnError(err => {
-            if (isNoEntError(err)) {
+          .do({
+            error(err) {
+              if (isNoEntError(err)) {
+                atom.notifications.addError(
+                  "adb wasn't found on your path!",
+                  {
+                    dismissable: true,
+                    description: NOENT_ERROR_DESCRIPTION,
+                  },
+                );
+                return;
+              }
               atom.notifications.addError(
-                "adb wasn't found on your path!",
-                {
-                  dismissable: true,
-                  description: NOENT_ERROR_DESCRIPTION,
-                },
+                'adb logcat has crashed 3 times.'
+                + ' You can manually restart it using the "Nuclide Adb Logcat: Start" command.'
               );
-              return;
-            }
-            atom.notifications.addError(
-              'adb logcat has crashed 3 times.'
-              + ' You can manually restart it using the "Nuclide Adb Logcat: Start" command.'
-            );
+            },
           })
       )
     );

pkg/nuclide-adb-logcat/lib/createMessageStream.js

@@ -11,10 +11,10 @@
 
 import type {Message} from '../../nuclide-console/lib/types';
 
+import {CompositeSubscription} from '../../nuclide-commons';
 import createMessage from './createMessage';
 import parseLogcatMetadata from './parseLogcatMetadata';
-import {CompositeDisposable} from 'atom';
-import Rx from 'rx';
+import Rx from '@reactivex/rxjs';
 
 export default function createMessageStream(
   line$: Rx.Observable<string>,
@@ -36,7 +36,7 @@ export default function createMessageStream(
         buffer.pop();
       }
 
-      observer.onNext({
+      observer.next({
         metadata: prevMetadata,
         message: buffer.join('\n'),
       });
@@ -46,8 +46,7 @@ export default function createMessageStream(
 
     const sharedLine$ = line$.share();
 
-    return new CompositeDisposable(
-
+    return new CompositeSubscription(
       // Buffer incoming lines.
       sharedLine$.subscribe(
         // onNext
@@ -71,21 +70,20 @@ export default function createMessageStream(
         // onError
         error => {
           flush();
-          observer.onError(error);
+          observer.error(error);
         },
 
         // onCompleted
         () => {
           flush();
-          observer.onCompleted();
+          observer.complete();
         },
       ),
 
       // We know *for certain* that we have a complete entry once we see the metadata for the next
       // one. But what if the next one takes a long time to happen? After a certain point, we need
       // to just assume we have the complete entry and move on.
-      sharedLine$.debounce(200).subscribe(flush),
-
+      sharedLine$.debounceTime(200).subscribe(flush),
     );
 
   })

pkg/nuclide-adb-logcat/lib/createProcessStream.js

@@ -11,7 +11,7 @@
 
 import {observeProcess, safeSpawn} from '../../nuclide-commons';
 import featureConfig from '../../nuclide-feature-config';
-import Rx from 'rx';
+import Rx from '@reactivex/rxjs';
 
 export function createProcessStream(): Rx.Observable<string> {
   return observeProcess(spawnAdbLogcat)

pkg/nuclide-adb-logcat/spec/createMessageStream-spec.js

@@ -10,7 +10,7 @@
  */
 
 import createMessageStream from '../lib/createMessageStream';
-import Rx from 'rx';
+import Rx from '@reactivex/rxjs';
 
 describe('createMessageStream', () => {
 

pkg/nuclide-analytics/lib/main.js

@@ -9,9 +9,10 @@
  * the root directory of this source tree.
  */
 
-import type Rx from 'rx';
+import type Rx from '@reactivex/rxjs';
 
 import invariant from 'assert';
+import {DisposableSubscription} from '../../nuclide-commons';
 import {track as rawTrack} from './track';
 
 export type TrackingEvent = {
@@ -45,7 +46,7 @@ function trackEvent(event: TrackingEvent): Promise<mixed> {
  * Track each event in a stream of TrackingEvents.
  */
 function trackEvents(events: Rx.Observable<TrackingEvent>): IDisposable {
-  return events.forEach(trackEvent);
+  return new DisposableSubscription(events.subscribe(trackEvent));
 }
 
 /**

pkg/nuclide-arcanist-base/lib/ArcanistBaseService.js

@@ -13,7 +13,7 @@ import type {NuclideUri} from '../../nuclide-remote-uri';
 import type {RevisionFileChanges} from '../../nuclide-hg-repository-base/lib/HgService';
 
 import invariant from 'assert';
-import {Observable} from 'rx';
+import {Observable} from '@reactivex/rxjs';
 import path from 'path';
 import {
   fsPromise,

pkg/nuclide-arcanist-client/lib/main.js

@@ -9,7 +9,7 @@
  * the root directory of this source tree.
  */
 
-import type {Observable} from 'rx';
+import type {Observable} from '@reactivex/rxjs';
 import type {NuclideUri} from '../../nuclide-remote-uri';
 import typeof * as ArcanistBaseService from '../../nuclide-arcanist-base';
 

pkg/nuclide-atom-helpers/lib/go-to-location.js

@@ -9,8 +9,8 @@
  * the root directory of this source tree.
  */
 
-import {Subject} from 'rx';
-import type {Observable} from 'rx';
+import {Subject} from '@reactivex/rxjs';
+import type {Observable} from '@reactivex/rxjs';
 
 // Opens the given file at the line/column.
 // By default will center the opened text editor.
@@ -49,7 +49,7 @@ function goToLocationInEditor(
     editor.scrollToBufferPosition([line, column], {center: true});
   }
 
-  goToLocationSubject.onNext(editor);
+  goToLocationSubject.next(editor);
 }
 
 

pkg/nuclide-atom-helpers/lib/text-editor.js

@@ -13,7 +13,7 @@ import type {NuclideUri} from '../../nuclide-remote-uri';
 
 import invariant from 'assert';
 import {TextBuffer, TextEditor} from 'atom';
-import {Observable} from 'rx';
+import {Observable} from '@reactivex/rxjs';
 
 // TODO(most): move to remote-connection/lib/RemoteTextBuffer.js
 import NuclideTextBuffer from '../../nuclide-remote-projects/lib/NuclideTextBuffer';
@@ -152,7 +152,7 @@ export function getCursorPositions(editor: atom$TextEditor): Observable<atom$Poi
   const cursor = editor.getCursors()[0];
   invariant(cursor != null);
   return Observable.merge(
-    Observable.just(cursor.getBufferPosition()),
+    Observable.of(cursor.getBufferPosition()),
     observableFromSubscribeFunction(cursor.onDidChangePosition.bind(cursor))
       .map(event => event.newBufferPosition),
   );

pkg/nuclide-buck-base/lib/BuckProject.js

@@ -47,7 +47,7 @@ type BuckCommandAndOptions = {
     queueName: string;
   };
 };
-import type {Observable} from 'rx';
+import type {Observable} from '@reactivex/rxjs';
 
 /**
  * As defined in com.facebook.buck.cli.Command, some of Buck's subcommands are

pkg/nuclide-buck-toolbar/lib/BuckToolbarStore.js

@@ -397,7 +397,7 @@ class BuckToolbarStore {
 
       return {
         kill() {
-          subscription.dispose();
+          subscription.unsubscribe();
           disposable.dispose();
         },
       };

pkg/nuclide-busy-signal-interfaces/lib/types.js

@@ -9,7 +9,7 @@
  * the root directory of this source tree.
  */
 
-import type {Observable} from 'rx';
+import type {Observable} from '@reactivex/rxjs';
 
 export type BusySignalMessage = BusySignalMessageBusy | BusySignalMessageDone;
 

pkg/nuclide-busy-signal-provider-base/lib/BusySignalProviderBase.js

@@ -11,11 +11,11 @@
 
 import type {BusySignalMessage} from '../../nuclide-busy-signal-interfaces';
 import type {NuclideUri} from '../../nuclide-remote-uri';
-import type {Observable} from 'rx';
+import type {Observable} from '@reactivex/rxjs';
 
 import {Disposable, CompositeDisposable} from 'atom';
 
-import {Subject} from 'rx';
+import {Subject} from '@reactivex/rxjs';
 import invariant from 'assert';
 
 import {promises} from '../../nuclide-commons';
@@ -72,9 +72,9 @@ export class BusySignalProviderBase {
 
   _displayMessage(message: string): IDisposable {
     const {busy, done} = this._nextMessagePair(message);
-    this._messages.onNext(busy);
+    this._messages.next(busy);
     return new Disposable(() => {
-      this._messages.onNext(done);
+      this._messages.next(done);
     });
   }
 

pkg/nuclide-busy-signal/lib/MessageStore.js

@@ -15,7 +15,7 @@ import type {
   BusySignalMessageBusy,
 } from '../../nuclide-busy-signal-interfaces';
 
-import {Observable, BehaviorSubject} from 'rx';
+import {Observable, BehaviorSubject} from '@reactivex/rxjs';
 import {Disposable} from 'atom';
 import invariant from 'assert';
 
@@ -33,7 +33,7 @@ export class MessageStore {
     const subscription =
       provider.messages.subscribe(message => this._processUpdate(provider, message));
     return new Disposable(() => {
-      subscription.dispose();
+      subscription.unsubscribe();
       this._currentMessages.delete(provider);
       this._publishMessages();
     });
@@ -65,6 +65,6 @@ export class MessageStore {
         messages.push(message);
       }
     }
-    this._messageStream.onNext(messages);
+    this._messageStream.next(messages);
   }
 }

pkg/nuclide-busy-signal/lib/StatusBarTile.js

@@ -11,7 +11,7 @@
 
 /* eslint-env browser */
 
-import type {Observable} from 'rx';
+import type {Observable} from '@reactivex/rxjs';
 
 import type {BusySignalMessageBusy} from '../../nuclide-busy-signal-interfaces';