STITCH-2499 Add support for 'watch' in React Native

packages/core/testutils/package.json

@@ -52,6 +52,7 @@
   },
   "devDependencies": {
     "cross-fetch": "^2.2.3",
+    "eventsource": "^1.0.7",
     "prettier": "^1.13.5",
     "tslint": "^5.10.0",
     "tslint-config-prettier": "^1.13.0",

packages/core/testutils/src/JestFetchStreamTransport.ts

@@ -15,6 +15,7 @@
  */
 
 import fetch from "cross-fetch";
+import EventSource from "eventsource";
 import {
     BasicRequest,
     ContentTypes,

packages/react-native/core/package.json

@@ -47,7 +47,8 @@
   "license": "Apache-2.0",
   "dependencies": {
     "bson": "4.0.2",
-    "mongodb-stitch-core-sdk": "^4.8.0"
+    "mongodb-stitch-core-sdk": "^4.8.0",
+    "rn-eventsource": "^1.0.0"
   },
   "devDependencies": {
     "prettier": "^1.13.5",

packages/react-native/core/src/core/Stitch.ts

@@ -19,7 +19,7 @@ import {
 } from "mongodb-stitch-core-sdk";
 
 import RNAsyncStorage from "./internal/common/RNAsyncStorage";
-import RNFetchTransport from "./internal/net/RNFetchTransport";
+import RNFetchStreamTransport from "./internal/net/RNFetchStreamTransport";
 import StitchAppClientImpl from "./internal/StitchAppClientImpl";
 import StitchAppClient from "./StitchAppClient";
 
@@ -139,7 +139,7 @@ export default class Stitch {
       builder.withStorage(new RNAsyncStorage(clientAppId));
     }
     if (builder.transport === undefined) {
-      builder.withTransport(new RNFetchTransport());
+      builder.withTransport(new RNFetchStreamTransport());
     }
     if (builder.baseUrl === undefined || builder.baseUrl === "") {
       builder.withBaseUrl(DEFAULT_BASE_URL);

packages/react-native/core/src/core/internal/common/RNAsyncStorage.ts

@@ -15,7 +15,7 @@
  */
 
 import { Storage } from "mongodb-stitch-core-sdk";
-import { AsyncStorage } from "react-native"
+import { AsyncStorage } from "react-native";
 
 const stitchPrefixKey = "__stitch.client";
 

packages/react-native/core/src/core/internal/net/EventSourceEventStream.ts

@@ -0,0 +1,88 @@
+/**
+ * Copyright 2018-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import {
+  BaseEventStream,
+  Event,
+  StitchClientError,
+  StitchClientErrorCode, 
+  StitchEvent
+} from "mongodb-stitch-core-sdk";
+import EventSource from "rn-eventsource";
+
+/** @hidden */
+export default class EventSourceEventStream extends BaseEventStream<EventSourceEventStream> {
+  private evtSrc: EventSource;
+  private readonly onOpenError: (error: Error) => void;
+  private openedOnce: boolean;
+
+  constructor(
+    evtSrc: EventSource,
+    onOpen: (stream: EventSourceEventStream) => void,
+    onOpenError: (error: Error) => void,
+    reconnecter?: () => Promise<EventSourceEventStream>
+  ) {
+    super(reconnecter);
+    this.evtSrc = evtSrc;
+    this.onOpenError = onOpenError;
+    this.openedOnce = false;
+
+    this.evtSrc.onopen = e => {
+      onOpen(this);
+      this.openedOnce = true;
+    };
+    this.reset();
+  }
+
+  public open(): void {
+    if (this.closed) {
+      throw new StitchClientError(StitchClientErrorCode.StreamClosed);
+    }
+  }
+
+  public afterClose(): void {
+    this.evtSrc.close();
+  }
+
+  protected onReconnect(next: EventSourceEventStream) {
+    this.evtSrc = next.evtSrc;
+    this.reset();
+    this.events = next.events.concat(this.events);
+  }
+
+  private reset(): void {
+    this.evtSrc.onmessage = e => {
+      this.events.push(new Event(Event.MESSAGE_EVENT, e.data));
+      this.poll();
+    };
+    this.evtSrc.onerror = e => {
+      if (e.data !== undefined) {
+        this.lastErr = e.data;
+        this.events.push(new Event(StitchEvent.ERROR_EVENT_NAME, this.lastErr!));
+        this.close();
+        this.poll();
+        return;
+      }
+      if (!this.openedOnce) {
+        this.close();
+        this.onOpenError(new Error("event source failed to open and will not reconnect; check network log for more details"));
+        return;
+      }
+      this.evtSrc.close();
+      this.reconnect();
+    }
+  }
+}

packages/react-native/core/src/core/internal/net/RNFetchTransport.ts → packages/react-native/core/src/core/internal/net/RNFetchStreamTransport.ts

@@ -18,15 +18,17 @@ import {
     BasicRequest,
     ContentTypes,
     EventStream,
+    handleRequestError,
     Headers,
     Response,
-    StitchClientError, 
-    StitchClientErrorCode, 
-    Transport
+    Transport,
   } from "mongodb-stitch-core-sdk";
-
+
+import EventSource from "rn-eventsource";
+import EventSourceEventStream from "./EventSourceEventStream";
+
   /** @hidden */
-  export default class RNFetchTransport implements Transport {
+  export default class RNFetchStreamTransport implements Transport {
     public roundTrip(request: BasicRequest): Promise<Response> {
       const responsePromise = fetch(request.url, {
         body: request.body,
@@ -51,6 +53,37 @@ import {
     }
 
     public stream(request: BasicRequest, open = true, retryRequest?: () => Promise<EventStream>): Promise<EventStream> {
-      throw new StitchClientError(StitchClientErrorCode.StreamingNotSupported);
+      const reqHeaders = { ...request.headers };
+      reqHeaders[Headers.ACCEPT] = ContentTypes.TEXT_EVENT_STREAM;
+      reqHeaders[Headers.CONTENT_TYPE] = ContentTypes.TEXT_EVENT_STREAM;
+
+      // Verify we can start a request with current params and potentially
+      // Force ourselves to refresh a token.
+      return fetch(request.url + "&stitch_validate=true", {
+        body: request.body,
+        headers: reqHeaders,
+        method: request.method,
+        mode: 'cors'
+      }).then(response => {
+        const respHeaders: { [key: string]: string } = {};
+        response.headers.forEach((value, key) => {
+          respHeaders[key] = value;
+        });
+        if (response.status < 200 || response.status >= 300) {
+          return response.text()
+          .then(body => handleRequestError(new Response(respHeaders, response.status, body)));
+        }
+
+        return new Promise<EventStream>((resolve, reject) => 
+          new EventSourceEventStream(
+            new EventSource(request.url),
+            stream => resolve(stream),
+            error => reject(error),
+            retryRequest ? 
+              () => retryRequest().then(es => es as EventSourceEventStream)
+              : undefined
+            )
+        );
+      });
     }
   }

packages/react-native/core/src/index.ts

@@ -59,7 +59,7 @@ import StitchAuth from "./core/auth/StitchAuth";
 import StitchAuthListener from "./core/auth/StitchAuthListener";
 import StitchUser from "./core/auth/StitchUser";
 import RNAsyncStorage from "./core/internal/common/RNAsyncStorage";
-import RNFetchTransport from "./core/internal/net/RNFetchTransport";
+import RNFetchStreamTransport from "./core/internal/net/RNFetchStreamTransport";
 import Stitch from "./core/Stitch";
 import StitchAppClient from "./core/StitchAppClient";
 import NamedServiceClientFactory from "./services/internal/NamedServiceClientFactory";
@@ -82,7 +82,7 @@ export {
   GoogleCredential,
   MemoryStorage,
   NamedServiceClientFactory,
-  RNFetchTransport,
+  RNFetchStreamTransport,
   RNAsyncStorage,
   ServerApiKeyAuthProvider,
   ServerApiKeyCredential,