Support token refresh

Author: simolus3

packages/common/src/client/sync/stream/AbstractRemote.ts

@@ -269,23 +269,21 @@ export abstract class AbstractRemote {
    */
   async socketStream(options: SocketSyncStreamOptions): Promise<DataStream<StreamingSyncLine>> {
     const bson = await this.getBSON();
-    return await this.socketStreamInternal(options, bson);
-  }
-
-  /**
-   * Connects to the sync/stream websocket endpoint without decoding BSON in JavaScript.
-   */
-  async socketStreamRaw(options: SocketSyncStreamOptions): Promise<DataStream<Buffer<ArrayBuffer>>> {
-    return this.socketStreamInternal(options);
+    return await this.socketStreamRaw(options, (data) => bson.deserialize(data), bson);
   }
 
   /**
    * Returns a data stream of sync line data.
    *
-   * @param bson A BSON encoder and decoder. When set, the data stream will emit parsed instances of
-   * {@link StreamingSyncLine}. Otherwise, unparsed buffers will be emitted instead.
+   * @param map Maps received payload frames to the typed event value.
+   * @param bson A BSON encoder and decoder. When set, the data stream will be requested with a BSON payload
+   * (required for compatibility with older sync services).
    */
-  private async socketStreamInternal(options: SocketSyncStreamOptions, bson?: typeof BSON): Promise<DataStream> {
+  async socketStreamRaw<T>(
+    options: SocketSyncStreamOptions,
+    map: (buffer: Buffer) => T,
+    bson?: typeof BSON
+  ): Promise<DataStream> {
     const { path, fetchStrategy = FetchStrategy.Buffered } = options;
     const mimeType = bson == null ? 'application/json' : 'application/bson';
 
@@ -415,7 +413,7 @@ export abstract class AbstractRemote {
               return;
             }
 
-            stream.enqueueData(bson != null ? bson.deserialize(data) : data);
+            stream.enqueueData(map(data));
           },
           onComplete: () => {
             stream.close();
@@ -455,43 +453,18 @@ export abstract class AbstractRemote {
   }
 
   /**
-   * Connects to the sync/stream http endpoint
+   * Connects to the sync/stream http endpoint, parsing lines as JSON.
    */
   async postStream(options: SyncStreamOptions): Promise<DataStream<StreamingSyncLine>> {
-    const jsonS = await this.postStreamRaw(options);
-
-    const stream = new DataStream({
-      logger: this.logger
-    });
-
-    const r = jsonS.getReader();
-
-    const l = stream.registerListener({
-      lowWater: async () => {
-        try {
-          const { done, value } = await r.read();
-          // Exit if we're done
-          if (done) {
-            stream.close();
-            l?.();
-            return;
-          }
-          stream.enqueueData(JSON.parse(value));
-        } catch (ex) {
-          stream.close();
-          throw ex;
-        }
-      },
-      closed: () => {
-        r.cancel();
-        l?.();
-      }
+    return await this.postStreamRaw(options, (line) => {
+      return JSON.parse(line) as StreamingSyncLine;
     });
-
-    return stream;
   }
 
-  async postStreamRaw(options: SyncStreamOptions): Promise<ReadableStream<string>> {
+  /**
+   * Connects to the sync/stream http endpoint, mapping and emitting each received string line.
+   */
+  async postStreamRaw<T>(options: SyncStreamOptions, mapLine: (line: string) => T): Promise<DataStream<T>> {
     const { data, path, headers, abortSignal } = options;
 
     const request = await this.buildRequest(path);
@@ -566,40 +539,52 @@ export abstract class AbstractRemote {
     const decoder = new TextDecoder();
     let buffer = '';
 
-    const outputStream = new ReadableStream<string>({
-      pull: async (controller) => {
-        let didCompleteLine = false;
+    const stream = new DataStream<T>({
+      logger: this.logger
+    });
 
-        while (!didCompleteLine) {
-          const { done, value } = await reader.read();
-          if (done) {
-            const remaining = buffer.trim();
-            if (remaining.length != 0) {
-              controller.enqueue(remaining);
-            }
+    const l = stream.registerListener({
+      lowWater: async () => {
+        try {
+          let didCompleteLine = false;
+          while (!didCompleteLine) {
+            const { done, value } = await reader.read();
+            if (done) {
+              const remaining = buffer.trim();
+              if (remaining.length != 0) {
+                stream.enqueueData(mapLine(remaining));
+              }
 
-            controller.close();
-            await closeReader();
-            return;
-          }
+              stream.close();
+              await closeReader();
+              return;
+            }
 
-          const data = decoder.decode(value, { stream: true });
-          buffer += data;
+            const data = decoder.decode(value, { stream: true });
+            buffer += data;
 
-          const lines = buffer.split('\n');
-          for (var i = 0; i < lines.length - 1; i++) {
-            var l = lines[i].trim();
-            if (l.length > 0) {
-              controller.enqueue(l);
-              didCompleteLine = true;
+            const lines = buffer.split('\n');
+            for (var i = 0; i < lines.length - 1; i++) {
+              var l = lines[i].trim();
+              if (l.length > 0) {
+                stream.enqueueData(mapLine(l));
+                didCompleteLine = true;
+              }
             }
-          }
 
-          buffer = lines[lines.length - 1];
+            buffer = lines[lines.length - 1];
+          }
+        } catch (ex) {
+          stream.close();
+          throw ex;
         }
+      },
+      closed: () => {
+        closeReader();
+        l?.();
       }
     });
 
-    return outputStream;
+    return stream;
   }
 }

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -837,6 +837,7 @@ The next upload iteration will be delayed.`);
 
     const abortController = new AbortController();
     signal.addEventListener('abort', () => abortController.abort());
+    let controlInvocations: DataStream<[string, ArrayBuffer | string | undefined]> | null = null;
 
     async function connect(instr: EstablishSyncStream) {
       const syncOptions: SyncStreamOptions = {
@@ -846,26 +847,29 @@ The next upload iteration will be delayed.`);
       };
 
       if (resolvedOptions.connectionMethod == SyncStreamConnectionMethod.HTTP) {
-        const lines = await remote.postStreamRaw(syncOptions);
-        for await (const syncLine of lines as any) {
-          await control('line_text', syncLine);
-        }
+        controlInvocations = await remote.postStreamRaw(syncOptions, (line) => ['line_text', line]);
       } else {
-        const stream = await remote.socketStreamRaw({ ...syncOptions, fetchStrategy: resolvedOptions.fetchStrategy });
-
-        try {
-          while (!stream.closed) {
-            const line = await stream.read();
-            if (line == null) {
-              return;
-            }
+        controlInvocations = await remote.socketStreamRaw(
+          {
+            ...syncOptions,
+            fetchStrategy: resolvedOptions.fetchStrategy
+          },
+          // TODO: Can we avoid the copy here?
+          (buffer) => ['line_binary', new Uint8Array(buffer).buffer]
+        );
+      }
 
-            const copy = new Uint8Array(line);
-            await control('line_binary', copy.buffer);
+      try {
+        while (!controlInvocations.closed) {
+          const line = await controlInvocations.read();
+          if (line == null) {
+            return;
           }
-        } finally {
-          await stream.close();
+
+          await control(line[0], line[1]);
         }
+      } finally {
+        await controlInvocations.close();
       }
     }
 
@@ -926,7 +930,17 @@ The next upload iteration will be delayed.`);
         if (instruction.FetchCredentials.did_expire) {
           remote.invalidateCredentials();
         } else {
-          // TODO: Async prefetch
+          remote.invalidateCredentials();
+
+          // Restart iteration after the credentials have been refreshed.
+          remote.fetchCredentials().then(
+            (_) => {
+              controlInvocations?.enqueueData(['refreshed_token', undefined]);
+            },
+            (err) => {
+              syncImplementation.logger.warn('Could not prefetch credentials', err);
+            }
+          );
         }
       } else if ('CloseSyncStream' in instruction) {
         abortController.abort();
@@ -949,7 +963,7 @@ The next upload iteration will be delayed.`);
 
     try {
       this.notifyCompletedUploads = () => {
-        control('completed_upload');
+        controlInvocations?.enqueueData(['completed_upload', undefined]);
       };
 
       await control('start', JSON.stringify(resolvedOptions.params));