Using Observables

Some parts of Nuclide use the Observable pattern (implemented by RxJS), to handle stream processing. An Introduction to Reactive Programming introduces some of the basic concepts of Observables.

Using Observables in the Service Framework

NOTE: First make sure that you read through Creating a Nuclide Service, which describes the basics of using the Service Framework.

The service framework has support for returning Observable streams from both functions and methods. This is useful in cases where you want to stream the results back from a command, but also retain the ability to cancel the command. Right now, the FindInProjectService uses this to stream the results of remote grep processes. This is what the definition file looks like:

FindInProjectService.def:

  export type search$FileResult = { ... };

  export function findInProjectSearch(directory: NuclideUri, regex: RegExp, subdirs: Array<string>):
    Observable<search$FileResult> {}

The implementation of the function follows this general structure:

FindInProjectService.js:

  export function findInProjectSearch(directory: NuclideUri, regex: RegExp, subdirs: Array<string>): Observable<search$FileResult> {
    return Observable.create(observer => {
      observer.onNext(...); // Emit an event.
      observer.onError(new Error()); // Throw an error.
      observer.onCompleted(); // Complete the stream.
      
      // You return a dispose function called in three situations.
      // 1. The stream completed.
      // 2. The stream threw an error.
      // 3. The subscriber to this stream ended the subscription.
      return () => {
        // Kill processes, stop watches, etc.
      };
    });
  }

Using the service now is very simple. First, we want send a search request to all of the remote directories that the user is connected to. We then merge the resulting streams, then redirect results to the UI.

  name=RemoteDirectorySearcher.js
  import {getServiceByNuclideUri} from 'nuclide-client';

  // Get the remote service that corresponds to each remote directory.
  var services = directories.map(dir => getServiceByNuclideUri('FindInProjectService', dir.getPath()));

  // Start the search in each directory, and merge the resulting streams.
  var searchStream = Observable.merge(directories.map((dir, index) =>
    services[index].findInProjectSearch(dir.getPath(), regex, options.subdirs)));

  // Subscribe to results.
  var subscription = searchStream.subscribe(next => {
    // Post search update to UI.
  }, error => {
    // Notify user of searching error.
  }, () => {
    // Notify user of search completion.
  });