[Flight] Add Client Infrastructure (#17234)

* Change demo to server

* Expose client in package.json

* Reorganize tests

We don't want unit tests but instead test how both server and clients work
together. So this merges server/client test files.

* Fill in the client implementation a bit

* Use new client in fixture

* Add Promise/Uint8Array to lint rule

I'll probably end up deleting these deps later but they're here for now.

fixtures/flight-browser/index.html

@@ -18,9 +18,12 @@ <h1>Flight Example</h1>
     </div>
     <script src="../../build/dist/react.development.js"></script>
     <script src="../../build/dist/react-dom.development.js"></script>
+    <script src="../../build/dist/react-dom-unstable-flight-server.browser.development.js"></script>
     <script src="../../build/dist/react-dom-unstable-flight-client.development.js"></script>
     <script src="https://unpkg.com/babel-standalone@6/babel.js"></script>
     <script type="text/babel">
+      let Suspense = React.Suspense;
+
       function Text({children}) {
         return <span>{children}</span>;
       }
@@ -40,7 +43,7 @@ <h1>Flight Example</h1>
         }
       };
 
-      let stream = ReactFlightDOMClient.renderToReadableStream(model);
+      let stream = ReactFlightDOMServer.renderToReadableStream(model);
       let response = new Response(stream, {
         headers: {'Content-Type': 'text/html'},
       });
@@ -49,35 +52,35 @@ <h1>Flight Example</h1>
       async function display(responseToDisplay) {
         let blob = await responseToDisplay.blob();
         let url = URL.createObjectURL(blob);
-        let response = await fetch(url);
-        let body = await response.body;
 
-        let reader = body.getReader();
-        let charsReceived = 0;
-        let decoder = new TextDecoder();
+        let data = ReactFlightDOMClient.readFromFetch(
+          fetch(url)
+        );
+        // The client also supports XHR streaming.
+        // var xhr = new XMLHttpRequest();
+        // xhr.open('GET', url);
+        // let data = ReactFlightDOMClient.readFromXHR(xhr);
+        // xhr.send();
 
-        let json = '';
-        reader.read().then(function processChunk({ done, value }) {
-          if (done) {
-            renderResult(json);
-            return;
-          }
-          json += decoder.decode(value);
-          return reader.read().then(processChunk);
-        });
+        renderResult(data);
       }
 
-      function Shell({ model }) {
+      function Shell({ data }) {
+        let model = data.model;
         return <div>
           <h1>{model.title}</h1>
           <div dangerouslySetInnerHTML={model.content} />
         </div>;
       }
 
-      function renderResult(json) {
-        let model = JSON.parse(json);
+      function renderResult(data) {
         let container = document.getElementById('container');
-        ReactDOM.render(<Shell model={model} />, container);
+        ReactDOM.render(
+          <Suspense fallback="Loading...">
+            <Shell data={data} />
+          </Suspense>,
+          container
+        );
       }
     </script>
   </body>

packages/react-dom/package.json

@@ -39,6 +39,7 @@
     "unstable-fizz.js",
     "unstable-fizz.browser.js",
     "unstable-fizz.node.js",
+    "unstable-flight-client.js",
     "unstable-flight-server.js",
     "unstable-flight-server.browser.js",
     "unstable-flight-server.node.js",

packages/react-dom/src/client/flight/ReactFlightDOMClient.js

@@ -7,28 +7,80 @@
  * @flow
  */
 
-import type {ReactModel} from 'react-flight/src/ReactFlightClient';
+import type {ReactModelRoot} from 'react-flight/src/ReactFlightClient';
 
 import {
-  createRequest,
-  startWork,
-  startFlowing,
-} from 'react-flight/inline.dom-browser';
+  createResponse,
+  getModelRoot,
+  reportGlobalError,
+  processStringChunk,
+  processBinaryChunk,
+  complete,
+} from 'react-flight/inline.dom';
 
-function renderToReadableStream(model: ReactModel): ReadableStream {
-  let request;
-  return new ReadableStream({
-    start(controller) {
-      request = createRequest(model, controller);
-      startWork(request);
+function startReadingFromStream(response, stream: ReadableStream): void {
+  let reader = stream.getReader();
+  function progress({done, value}) {
+    if (done) {
+      complete(response);
+      return;
+    }
+    let buffer: Uint8Array = (value: any);
+    processBinaryChunk(response, buffer, 0);
+    return reader.read().then(progress, error);
+  }
+  function error(e) {
+    reportGlobalError(response, e);
+  }
+  reader.read().then(progress, error);
+}
+
+function readFromReadableStream<T>(stream: ReadableStream): ReactModelRoot<T> {
+  let response = createResponse(stream);
+  startReadingFromStream(response, stream);
+  return getModelRoot(response);
+}
+
+function readFromFetch<T>(
+  promiseForResponse: Promise<Response>,
+): ReactModelRoot<T> {
+  let response = createResponse(promiseForResponse);
+  promiseForResponse.then(
+    function(r) {
+      startReadingFromStream(response, (r.body: any));
     },
-    pull(controller) {
-      startFlowing(request, controller.desiredSize);
+    function(e) {
+      reportGlobalError(response, e);
     },
-    cancel(reason) {},
-  });
+  );
+  return getModelRoot(response);
+}
+
+function readFromXHR<T>(request: XMLHttpRequest): ReactModelRoot<T> {
+  let response = createResponse(request);
+  let processedLength = 0;
+  function progress(e: ProgressEvent): void {
+    let chunk = request.responseText;
+    processStringChunk(response, chunk, processedLength);
+    processedLength = chunk.length;
+  }
+  function load(e: ProgressEvent): void {
+    progress(e);
+    complete(response);
+  }
+  function error(e: ProgressEvent): void {
+    reportGlobalError(response, new TypeError('Network error'));
+  }
+  request.addEventListener('progress', progress);
+  request.addEventListener('load', load);
+  request.addEventListener('error', error);
+  request.addEventListener('abort', error);
+  request.addEventListener('timeout', error);
+  return getModelRoot(response);
 }
 
 export default {
-  renderToReadableStream,
+  readFromXHR,
+  readFromFetch,
+  readFromReadableStream,
 };

packages/react-dom/src/client/flight/ReactFlightDOMHostConfig.js

@@ -7,44 +7,28 @@
  * @flow
  */
 
-export type Destination = ReadableStreamController;
+export type Source = Promise<Response> | ReadableStream | XMLHttpRequest;
 
-export function scheduleWork(callback: () => void) {
-  callback();
-}
-
-export function flushBuffered(destination: Destination) {
-  // WHATWG Streams do not yet have a way to flush the underlying
-  // transform streams. https://github.com/whatwg/streams/issues/960
-}
-
-export function beginWriting(destination: Destination) {}
+export type StringDecoder = TextDecoder;
 
-export function writeChunk(destination: Destination, buffer: Uint8Array) {
-  destination.enqueue(buffer);
-}
-
-export function completeWriting(destination: Destination) {}
+export const supportsBinaryStreams = true;
 
-export function close(destination: Destination) {
-  destination.close();
+export function createStringDecoder(): StringDecoder {
+  return new TextDecoder();
 }
 
-const textEncoder = new TextEncoder();
-
-export function convertStringToBuffer(content: string): Uint8Array {
-  return textEncoder.encode(content);
-}
+const decoderOptions = {stream: true};
 
-export function formatChunkAsString(type: string, props: Object): string {
-  let str = '<' + type + '>';
-  if (typeof props.children === 'string') {
-    str += props.children;
-  }
-  str += '</' + type + '>';
-  return str;
+export function readPartialStringChunk(
+  decoder: StringDecoder,
+  buffer: Uint8Array,
+): string {
+  return decoder.decode(buffer, decoderOptions);
 }
 
-export function formatChunk(type: string, props: Object): Uint8Array {
-  return convertStringToBuffer(formatChunkAsString(type, props));
+export function readFinalStringChunk(
+  decoder: StringDecoder,
+  buffer: Uint8Array,
+): string {
+  return decoder.decode(buffer);
 }