STITCH-2576 Close streams when user context switches (#247)

Author: adamchel

Diff for: packages/browser/core/src/core/internal/StitchAppClientImpl.ts

@@ -15,8 +15,12 @@
  */
 
 import {
+  AuthEventKind,
+  AuthRebindEvent,
   CoreStitchAppClient,
+  CoreStitchServiceClient,
   CoreStitchServiceClientImpl,
+  RebindEvent,
   StitchAppClientConfiguration,
   StitchAppClientInfo,
   StitchAppRequestClient
@@ -28,17 +32,22 @@ import StitchServiceClientImpl from "../../services/internal/StitchServiceClient
 import StitchServiceClient from "../../services/StitchServiceClient";
 import StitchAuthImpl from "../auth/internal/StitchAuthImpl";
 import StitchBrowserAppRoutes from "../auth/internal/StitchBrowserAppRoutes";
+import StitchAuth from "../auth/StitchAuth";
+import StitchAuthListener from "../auth/StitchAuthListener";
+import StitchUser from "../auth/StitchUser";
 import StitchAppClient from "../StitchAppClient";
 
-
 /** @hidden */
-export default class StitchAppClientImpl implements StitchAppClient {
+export default class StitchAppClientImpl implements StitchAppClient, 
+  StitchAuthListener {
   public readonly auth: StitchAuthImpl;
 
   private readonly coreClient: CoreStitchAppClient;
   private readonly info: StitchAppClientInfo;
   private readonly routes: StitchBrowserAppRoutes;
 
+  private serviceClients: CoreStitchServiceClient[];
+
   public constructor(
     clientAppId: string,
     config: StitchAppClientConfiguration
@@ -64,42 +73,75 @@ export default class StitchAppClientImpl implements StitchAppClient {
       this.info
     );
     this.coreClient = new CoreStitchAppClient(this.auth, this.routes);
+    this.serviceClients = [];
+    this.auth.addSynchronousAuthListener(this);
   }
 
   public getServiceClient<T>(
     factory: ServiceClientFactory<T> | NamedServiceClientFactory<T>,
     serviceName?: string
   ): T {
     if (isServiceClientFactory(factory)) {
-      return factory.getClient(
-        new CoreStitchServiceClientImpl(this.auth, this.routes.serviceRoutes, ""),
-        this.info
+      const serviceClient = new CoreStitchServiceClientImpl(
+        this.auth, this.routes.serviceRoutes, ""
       );
+      this.bindServiceClient(serviceClient);
+      return factory.getClient(serviceClient, this.info);
     } else {
+      const serviceClient = new CoreStitchServiceClientImpl(
+        this.auth,
+        this.routes.serviceRoutes,
+        serviceName!
+      );
+      this.bindServiceClient(serviceClient);
       return factory.getNamedClient(
-        new CoreStitchServiceClientImpl(
-          this.auth,
-          this.routes.serviceRoutes,
-          serviceName!
-        ),
+        serviceClient,
         this.info
       );
     }
   }
 
   public getGeneralServiceClient(serviceName: string): StitchServiceClient {
+    const serviceClient = new CoreStitchServiceClientImpl(
+      this.auth,
+      this.routes.serviceRoutes,
+      serviceName
+    );
+    this.bindServiceClient(serviceClient);
     return new StitchServiceClientImpl(
-      new CoreStitchServiceClientImpl(
-        this.auth,
-        this.routes.serviceRoutes,
-        serviceName
-      )
+      serviceClient
     );
   }
 
   public callFunction(name: string, args: any[]): Promise<any> {
     return this.coreClient.callFunction(name, args);
   }
+
+  // note: this is the only rebind event we care about for JS. if we add 
+  // services in the future, or update existing services in such a way that 
+  // they'll need to rebind on other types of events, those handlers should be
+  // added to this file.
+  public onActiveUserChanged(
+    _: StitchAuth, 
+    currentActiveUser: StitchUser | undefined, 
+    previousActiveUser: StitchUser | undefined
+  ) {
+    this.onRebindEvent(new AuthRebindEvent({
+      currentActiveUser,
+      kind: AuthEventKind.ActiveUserChanged,
+      previousActiveUser
+    }))
+  }
+
+  private bindServiceClient(coreStitchServiceClient: CoreStitchServiceClient) {
+    this.serviceClients.push(coreStitchServiceClient);
+  }
+
+  private onRebindEvent(rebindEvent: RebindEvent) {
+    this.serviceClients.forEach(serviceClient => {
+      serviceClient.onRebindEvent(rebindEvent);
+    })
+  }
 }
 
 function isServiceClientFactory<T>(

Diff for: packages/core/sdk/__tests__/StreamTestUtils.ts

@@ -0,0 +1,78 @@
+/**
+ * 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 { instance, mock, when } from "ts-mockito";
+import { BaseEventStream, Decoder, Event, EventStream, Stream } from "../src";
+
+// prevent Jest from thinking this file is an empty test suite
+it('should pass', () => {})
+
+class TestStream <T> extends Stream<T> {
+  public closeCalled: number
+  
+  constructor(eventStream: EventStream, decoder: Decoder<T>) {
+    super(eventStream, decoder)
+    this.closeCalled = 0;
+  }
+
+  public close() {
+    this.closeCalled += 1;
+    super.close();
+  }
+}
+
+/**
+ * Utility methods for working with streams in tests by creating a real 
+ * {@link Stream} object with a mocked {@link EventStream} underneath.
+ */
+class StreamTestUtils {
+  public static createStream<T>(
+    decoder: Decoder<T> | undefined,
+    ...content: string[]
+  ): Stream<T> {
+    return this.doCreateStream(decoder, true, content);
+  }
+
+  public static createClosedStream<T>(
+    decoder: Decoder<T> | undefined,
+    ...content: string[]
+  ): Stream<T> {
+    return this.doCreateStream(decoder, false, content);
+  }
+
+  private static doCreateStream<T>(
+    decoder: Decoder<T> | undefined, 
+    open: boolean, 
+    streamEvents: string[],
+  ): Stream<T> {
+    const eventStream = mock(BaseEventStream);
+
+    when(eventStream.isOpen()).thenReturn(open);
+
+    if (open) {
+      when(eventStream.nextEvent()).thenResolve(
+        ...streamEvents.map(s => new Event(Event.MESSAGE_EVENT, s))
+      );
+    }
+
+    return new TestStream(instance(eventStream), decoder)
+  }
+}
+
+export {
+  StreamTestUtils,
+  TestStream
+}