fix: sync device logs

Author: noomorph

src/logs/log-buffer.ts

@@ -5,6 +5,7 @@ import type { Emitter, AndroidEntry, IosEntry } from 'logkitten';
 import { Level, logkitten } from 'logkitten';
 import type { DetoxAllure2AdapterDeviceLogsOptions } from '../types';
 import type { DeviceWrapper } from '../utils';
+import { Deferred } from '../utils';
 
 import { PIDEntryCollection } from './pid-entry-collection';
 
@@ -32,6 +33,7 @@ export class LogBuffer implements StepLogRecorder {
   private readonly _appEntries = new PIDEntryCollection();
   private readonly _detoxEntries = new PIDEntryCollection();
   private _options: DetoxAllure2AdapterDeviceLogsOptions;
+  private readonly _deferreds = new Set<Deferred<number>>();
 
   constructor(readonly _config: LogBufferOptions) {
     const deviceId = this._config.device.id;
@@ -102,8 +104,32 @@ export class LogBuffer implements StepLogRecorder {
     }
   };
 
+  private _updateDeferreds(ts: number) {
+    for (const deferred of this._deferreds) {
+      deferred.update(ts);
+    }
+  }
+
+  /**
+   * Waits until a log entry with timestamp > reference is seen, or times out.
+   * @param reference timestamp to compare against
+   * @param timeoutMs timeout in ms (default 1000)
+   */
+  public ensureIntegrity(reference: number, timeoutMs = 1000): Promise<void> {
+    const deferred = new Deferred<number>({
+      timeoutMs,
+      predicate: (ts) => ts > reference,
+      cleanup: () => {
+        this._deferreds.delete(deferred);
+      },
+    });
+    this._deferreds.add(deferred);
+    return deferred.promise.then(() => {}); // resolve to void
+  }
+
   private readonly _onEntry = (entry: AnyEntry) => {
     this._appEntries.push(entry);
+    this._updateDeferreds(entry.ts);
   };
 
   private _iosFilter(entry: IosEntry): boolean {

src/utils/deferred.ts

@@ -0,0 +1,50 @@
+export class Deferred<T> {
+  public promise: Promise<T>;
+  private _resolve!: (value: T) => void;
+  private _reject!: (reason?: any) => void;
+  private _settled = false;
+  private _timeout: NodeJS.Timeout;
+  private _cleanup: () => void;
+  private _predicate: (value: T) => boolean;
+
+  constructor({
+    timeoutMs,
+    predicate,
+    cleanup,
+  }: {
+    timeoutMs: number;
+    predicate: (value: T) => boolean;
+    cleanup: () => void;
+  }) {
+    this._predicate = predicate;
+    this._cleanup = cleanup;
+    this.promise = new Promise<T>((resolve, reject) => {
+      this._resolve = (value) => {
+        if (!this._settled) {
+          this._settled = true;
+          clearTimeout(this._timeout);
+          this._cleanup();
+          resolve(value);
+        }
+      };
+      this._reject = (reason) => {
+        if (!this._settled) {
+          this._settled = true;
+          clearTimeout(this._timeout);
+          this._cleanup();
+          reject(reason);
+        }
+      };
+    });
+    this._timeout = setTimeout(() => {
+      this._reject(new Error('Deferred timed out'));
+    }, timeoutMs);
+  }
+
+  update(value: T) {
+    if (this._settled) return;
+    if (this._predicate(value)) {
+      this._resolve(value);
+    }
+  }
+}

src/utils/index.ts

@@ -1,2 +1,3 @@
+export * from './deferred';
 export * from './device-wrapper';
 export * from './worker-wrapper';