Simplify RN node executor

Summary:This greatly simplifies the RN node executor.

Previously, messages from the RN app would be translated by `ChildManager` into calls of `Child` methods, which would then create its own message format to send to the executor. Responsibilities for handling these messages were divided between all three. After this change, the RN message format is maintained all the way to executor. Not only does this remove unnecessary transformations, it also makes things a lot easier to understand for people familiar with RN already.

In addition, I've moved the script loading into the executor. This results in a clearer division of responsibilities, with RN sending and receiving messages to the executor and Nuclide only being there to facilitate the transfer. With that in mind, I was able to replace `Child` with a function (`executeRnRequests`) that accepts a stream of RN messages and outputs a stream of results. Since this sums up all of Nuclide's responsibilities here, I'd like to remove `ChildManager` entirely (it's now just an OO interface to `executeRnRequests`) but it's currently being used by the `ExecutorServer`, so I'll hold off until we're able to remove that in favor of the new `DebuggerProxyClient`.

This refactor also allowed me to switch from calling `process._debugProcess` on a running process to starting it with `--debug-brk` which fixed the issue with the debugger not skipping the loader breakpoint when using node 5.x (since "restartframe" is no longer called?see nodejs/node#5221).

public

Reviewed By: ssorallen

Differential Revision: D3073832

fb-gh-sync-id: 3604c2d19c5ab336b854766ff10c5a77695bade5
shipit-source-id: 3604c2d19c5ab336b854766ff10c5a77695bade5

pkg/nuclide-react-native-node-executor/lib/ChildManager.js

@@ -9,17 +9,16 @@
  * the root directory of this source tree.
  */
 
-import invariant from 'assert';
-import http from 'http';
-import url from 'url';
-
-import Child from './Child';
 import type {
   RnRequest,
+  ExecutorResponse,
   ServerReplyCallback,
 } from './types';
 import type {EventEmitter} from 'events';
 
+import {executeRnRequests} from './executeRnRequests';
+import {Observable, Subject} from 'rx';
+
 let logger;
 function getLogger() {
   if (!logger) {
@@ -30,95 +29,58 @@ function getLogger() {
 
 export default class ChildManager {
 
-  _child: ?Child;
   _onReply: ServerReplyCallback;
   _emitter: EventEmitter;
 
+  _executorSubscription: ?IDisposable;
+  _executorResponses: Observable<ExecutorResponse>;
+  _rnRequests: Subject<RnRequest>;
+
   constructor(onReply: ServerReplyCallback, emitter: EventEmitter) {
     this._onReply = onReply;
     this._emitter = emitter;
+    this._rnRequests = new Subject();
+    this._executorResponses = executeRnRequests(this._rnRequests);
   }
 
   _createChild(): void {
-    if (this._child == null) {
-      this._child = new Child(this._onReply, this._emitter);
+    if (this._executorSubscription != null) {
+      return;
     }
+
+    this._executorSubscription = this._executorResponses.subscribe(response => {
+      switch (response.kind) {
+        case 'result':
+          this._onReply(response.replyId, response.result);
+          return;
+        case 'error':
+          getLogger().error(response.message);
+          return;
+        case 'pid':
+          this._emitter.emit('eval_application_script', response.pid);
+          return;
+      }
+    });
   }
 
-  async killChild(): Promise<void> {
-    if (!this._child) {
+  killChild(): void {
+    if (!this._executorSubscription) {
       return;
     }
-    await this._child.kill();
-    this._child = null;
+    this._executorSubscription.dispose();
+    this._executorSubscription = null;
   }
 
-  handleMessage(message: RnRequest): void {
-    if (message.replyID) {
+  handleMessage(request: RnRequest): void {
+    if (request.replyID) {
       // getting cross-talk from another executor (probably Chrome)
       return;
     }
 
-    switch (message.method) {
-      case 'prepareJSRuntime':
-        return this._prepareJSRuntime(message);
-      case 'executeApplicationScript':
-        return this._executeApplicationScript(message);
-      default:
-        return this._executeJSCall(message);
-    }
-  }
-
-  _prepareJSRuntime(message: RnRequest): void {
+    // Make sure we have a worker to run the JS.
     this._createChild();
-    this._onReply(message.id);
-  }
-
-  _executeApplicationScript(message: RnRequest): void {
-    (async () => {
-      if (!this._child) {
-        // Warn Child not initialized;
-        return;
-      }
-
-      const {id: messageId, url: messageUrl, inject} = message;
-      invariant(messageId != null);
-      invariant(messageUrl != null);
-      invariant(inject != null);
-
-      const parsedUrl = url.parse(messageUrl, /* parseQueryString */ true);
-      invariant(parsedUrl.query);
-      parsedUrl.query.inlineSourceMap = true;
-      delete parsedUrl.search;
-      // $FlowIssue url.format() does not accept what url.parse() returns.
-      const scriptUrl = url.format(parsedUrl);
-      const script = await getScriptContents(scriptUrl);
-      invariant(this._child);
-      this._child.executeApplicationScript(script, inject, messageId);
-    })();
-  }
 
-  _executeJSCall(message: RnRequest): void {
-    if (!this._child) {
-      // Warn Child not initialized;
-      return;
-    }
-    this._child.execCall(message, message.id);
+    this._rnRequests.onNext(request);
   }
-}
 
-function getScriptContents(src): Promise<string> {
-  return new Promise((resolve, reject) => {
-    http.get(src, res => {
-      res.setEncoding('utf8');
-      let buff = '';
-      res.on('data', chunk => buff += chunk);
-      res.on('end', () => {
-        resolve(buff);
-      });
-    }).on('error', err => {
-      getLogger().error('Failed to get script from packager.');
-      reject(err);
-    });
-  });
 }

pkg/nuclide-react-native-node-executor/lib/executeRnRequests.js

@@ -0,0 +1,80 @@
+'use babel';
+/* @flow */
+
+/*
+ * Copyright (c) 2015-present, Facebook, Inc.
+ * All rights reserved.
+ *
+ * This source code is licensed under the license found in the LICENSE file in
+ * the root directory of this source tree.
+ */
+
+import type {ExecutorResponse, RnRequest} from './types';
+
+import featureConfig from '../../nuclide-feature-config';
+import {
+  createProcessStream,
+  forkWithExecEnvironment,
+  getOutputStream,
+} from '../../nuclide-commons/lib/process';
+import {getLogger} from '../../nuclide-logging';
+import {CompositeDisposable} from 'atom';
+import path from 'path';
+import {Observable} from 'rx';
+
+const logger = getLogger();
+
+export function executeRnRequests(rnRequests: Observable<RnRequest>): Observable<ExecutorResponse> {
+  const workerProcess = createProcessStream(() => (
+    // TODO: The node location/path needs to be more configurable. We need to figure out a way to
+    //   handle this across the board.
+    forkWithExecEnvironment(
+      path.join(__dirname, 'executor.js'),
+      [],
+      {
+        execArgv: ['--debug-brk'],
+        execPath: featureConfig.get('nuclide-react-native.pathToNode'),
+        silent: true,
+      },
+    )
+  ));
+
+  return Observable.merge(
+    workerProcess.map(process => ({
+      kind: 'pid',
+      pid: process.pid,
+    })),
+
+    // The messages we're receiving from the worker process.
+    (
+      workerProcess.flatMap(
+        process => Observable.fromEvent(process, 'message')
+      ): Observable<ExecutorResponse>
+    ),
+
+    Observable.create(() => (
+      new CompositeDisposable(
+        // Send the incoming requests to the worker process for evaluation.
+        rnRequests
+          .withLatestFrom(workerProcess, (r, p) => ([r, p]))
+          .subscribe(([request, process]) => { process.send(request); }),
+
+        // Pipe output from forked process. This just makes things easier to debug for us.
+        workerProcess
+          .flatMapLatest(process => getOutputStream(process))
+          .subscribe(message => {
+            switch (message.kind) {
+              case 'error':
+                logger.error(message.error.message);
+                return;
+              case 'stderr':
+              case 'stdout':
+                logger.info(message.data.toString());
+                return;
+            }
+          }),
+      )
+    )),
+
+  ).share();
+}