feat(client+core): input/outputs

Author: blakebyrnes

client/connections/ConnectionToCore.ts

@@ -52,7 +52,7 @@ export default abstract class ConnectionToCore extends TypedEventEmitter<{
   constructor(options?: IConnectionToCoreOptions) {
     super();
     this.options = options ?? { isPersistent: true };
-    this.commandQueue = new CoreCommandQueue(null, this);
+    this.commandQueue = new CoreCommandQueue(null, this, null);
     this.coreSessions = new CoreSessions(
       this.options.maxConcurrency,
       this.options.agentTimeoutMillis,
@@ -85,13 +85,15 @@ export default abstract class ConnectionToCore extends TypedEventEmitter<{
     if (!this.connectPromise) {
       this.connectPromise = new Resolvable();
       try {
+        const startTime = new Date();
         const connectError = await this.createConnection();
         if (connectError) throw connectError;
         if (this.isDisconnecting) throw new DisconnectedFromCoreError(this.resolvedHost);
         // can be resolved if canceled by a disconnect
         if (this.connectPromise.isResolved) return;
 
         const connectResult = await this.internalSendRequestAndWait({
+          startDate: startTime,
           command: 'Core.connect',
           args: [this.connectOptions],
         });
@@ -122,11 +124,13 @@ export default abstract class ConnectionToCore extends TypedEventEmitter<{
 
   public async disconnect(fatalError?: Error): Promise<void> {
     // user triggered disconnect sends a disconnect to Core
+    const startTime = new Date();
     await this.internalDisconnect(fatalError, async () => {
       try {
         await this.internalSendRequestAndWait(
           {
             command: 'Core.disconnect',
+            startDate: startTime,
             args: [fatalError],
           },
           2e3,
@@ -144,7 +148,7 @@ export default abstract class ConnectionToCore extends TypedEventEmitter<{
   ///////  PIPE FUNCTIONS  /////////////////////////////////////////////////////////////////////////////////////////////
 
   public async sendRequest(
-    payload: Omit<ICoreRequestPayload, 'messageId'>,
+    payload: Omit<ICoreRequestPayload, 'messageId' | 'sendDate'>,
   ): Promise<ICoreResponsePayload> {
     const result = await this.connect();
     if (result) throw result;
@@ -243,7 +247,7 @@ export default abstract class ConnectionToCore extends TypedEventEmitter<{
   }
 
   protected async internalSendRequestAndWait(
-    payload: Omit<ICoreRequestPayload, 'messageId'>,
+    payload: Omit<ICoreRequestPayload, 'messageId' | 'sendDate'>,
     timeoutMs?: number,
   ): Promise<ICoreResponsePayload> {
     const { promise, id, resolve } = this.createPendingResult();
@@ -257,6 +261,7 @@ export default abstract class ConnectionToCore extends TypedEventEmitter<{
     try {
       await this.internalSendRequest({
         messageId: id,
+        sendDate: new Date(),
         ...payload,
       });
     } catch (error) {
@@ -300,7 +305,7 @@ export default abstract class ConnectionToCore extends TypedEventEmitter<{
       }
       this.rejectPendingRequest(pending, responseError);
     } else {
-      pending.resolve({ data: message.data, commandId: message.commandId });
+      pending.resolve({ data: message.data });
     }
   }
 

client/index.mjs

@@ -16,6 +16,7 @@ const {
   Tab,
   XPathResult,
   LocationStatus,
+  Observable,
   LocationTrigger,
 } = cjsImport;
 
@@ -33,6 +34,7 @@ export {
   Node,
   FrameEnvironment,
   Tab,
+  Observable,
   XPathResult,
   LocationStatus,
   LocationTrigger,

client/index.ts

@@ -15,10 +15,15 @@ import type Tab from './lib/Tab';
 import RemoteConnectionToCore from './connections/RemoteConnectionToCore';
 import ConnectionToCore from './connections/ConnectionToCore';
 import ConnectionFactory from './connections/ConnectionFactory';
+import { Observable } from './lib/ObjectObserver';
+import { readCommandLineArgs } from './lib/Input';
 
-export default new Agent();
+const input = readCommandLineArgs();
+
+export default new Agent({ input });
 
 export {
+  Observable,
   Handler,
   Agent,
   RemoteConnectionToCore,

client/interfaces/IAgentCreateOptions.ts

@@ -7,4 +7,5 @@ export default interface IAgentCreateOptions
   name?: string;
   showReplay?: boolean;
   connectionToCore?: IConnectionToCoreOptions | ConnectionToCore;
+  input?: { command?: string } & any;
 }

client/interfaces/ICommandCounter.ts

@@ -0,0 +1,4 @@
+export default interface ICommandCounter {
+  nextCommandId: number;
+  lastCommandId: number;
+}

client/lib/Agent.ts

@@ -52,6 +52,7 @@ import FrameEnvironment, { getCoreFrameEnvironment } from './FrameEnvironment';
 import getClientExtenderPluginClass from './getClientExtenderPluginClass';
 import FrozenTab from './FrozenTab';
 import FileChooser from './FileChooser';
+import Output, { createObservableOutput } from './Output';
 
 export const DefaultOptions = {
   defaultBlockedResourceTypes: [BlockedResourceType.None],
@@ -73,6 +74,7 @@ const propertyKeys: (keyof Agent)[] = [
   'sessionId',
   'meta',
   'tabs',
+  'output',
   'frameEnvironments',
   'mainFrameEnvironment',
   'coreHost',
@@ -86,6 +88,10 @@ const propertyKeys: (keyof Agent)[] = [
 export default class Agent extends AwaitedEventTarget<{ close: void }> implements IAgent {
   protected static options: IAgentDefaults = { ...DefaultOptions };
 
+  public readonly input: { command?: string } & any;
+
+  #output: Output;
+
   constructor(options: IAgentCreateOptions = {}) {
     super(() => {
       return {
@@ -97,6 +103,7 @@ export default class Agent extends AwaitedEventTarget<{ close: void }> implement
     options.blockedResourceTypes =
       options.blockedResourceTypes || Agent.options.defaultBlockedResourceTypes;
     options.userProfile = options.userProfile || Agent.options.defaultUserProfile;
+    this.input = options.input;
 
     const sessionName = scriptInstance.generateSessionName(options.name);
     delete options.name;
@@ -115,6 +122,24 @@ export default class Agent extends AwaitedEventTarget<{ close: void }> implement
     });
   }
 
+  public get output(): any | any[] {
+    if (!this.#output) {
+      const coreSession = getState(this)
+        .connection.getCoreSessionOrReject()
+        .catch(() => null);
+      this.#output = createObservableOutput(coreSession);
+    }
+    return this.#output;
+  }
+
+  public set output(value: any | any[]) {
+    const output = this.output;
+    for (const key of Object.keys(output)) {
+      delete output[key];
+    }
+    Object.assign(this.output, value);
+  }
+
   public get activeTab(): Tab {
     return getState(this).connection.activeTab;
   }
@@ -462,7 +487,8 @@ class SessionConnection {
     }
     this.hasConnected = true;
     const { plugins } = getState(this.agent);
-    const { showReplay, connectionToCore, ...options } = getState(this.agent).options;
+    const { showReplay, connectionToCore, ...options } = getState(this.agent)
+      .options as IAgentCreateOptions;
 
     const connection = ConnectionFactory.createConnection(
       connectionToCore ?? { isPersistent: false },

client/lib/CoreCommandQueue.ts

@@ -1,28 +1,42 @@
 import ISessionMeta from '@secret-agent/interfaces/ISessionMeta';
 import { CanceledPromiseError } from '@secret-agent/commons/interfaces/IPendingWaitEvent';
 import Queue from '@secret-agent/commons/Queue';
+import ICoreRequestPayload from '@secret-agent/interfaces/ICoreRequestPayload';
 import ConnectionToCore from '../connections/ConnectionToCore';
 import { convertJsPathArgs } from './SetupAwaitedHandler';
+import ICommandCounter from '../interfaces/ICommandCounter';
 
 export default class CoreCommandQueue {
-  public lastCommandId = 0;
+  public get lastCommandId(): number {
+    return this.commandCounter?.lastCommandId;
+  }
+
+  public get nextCommandId(): number {
+    return this.commandCounter?.nextCommandId;
+  }
 
   private readonly internalState: {
     queue: Queue;
-    batchSendCommands: { [command: string]: any[][] };
+    commandsToRecord: ICoreRequestPayload['recordCommands'];
   };
 
+  private readonly commandCounter?: ICommandCounter;
   private readonly sessionMarker: string = '';
+  private readonly meta: ISessionMeta;
+  private readonly connection: ConnectionToCore;
+  private flushOnTimeout: NodeJS.Timeout;
 
   private get internalQueue(): Queue {
     return this.internalState.queue;
   }
 
   constructor(
-    private readonly meta: (ISessionMeta & { sessionName: string }) | null,
-    private readonly connection: ConnectionToCore,
+    meta: (ISessionMeta & { sessionName: string }) | null,
+    connection: ConnectionToCore,
+    commandCounter: ICommandCounter,
     internalState?: CoreCommandQueue['internalState'],
   ) {
+    this.connection = connection;
     if (meta) {
       const markers = [
         ''.padEnd(50, '-'),
@@ -31,38 +45,47 @@ export default class CoreCommandQueue {
         ''.padEnd(50, '-'),
       ].join('\n');
       this.sessionMarker = `\n\n${markers}`;
+      this.meta = { sessionId: meta.sessionId, tabId: meta.tabId, frameId: meta.frameId };
     }
+    this.commandCounter = commandCounter;
 
-    if (internalState) {
-      this.internalState = internalState;
-    } else {
-      this.internalState = {
-        queue: new Queue('CORE COMMANDS', 1),
-        batchSendCommands: {},
-      };
-    }
+    this.internalState = internalState ?? {
+      queue: new Queue('CORE COMMANDS', 1),
+      commandsToRecord: [],
+    };
   }
 
-  public queueBatchedCommand(command: string, ...args: any[]): void {
-    this.internalState.batchSendCommands[command] ??= [];
-    this.internalState.batchSendCommands[command].push(args);
-    if (this.internalState.batchSendCommands[command].length > 1000) this.flush().catch(() => null);
-    this.internalQueue.enqueue(this.flush.bind(this));
+  public record(command: { command: string; args: any[]; commandId?: number }): void {
+    this.internalState.commandsToRecord.push({
+      ...command,
+      startDate: new Date(),
+    });
+    if (this.internalState.commandsToRecord.length > 1000) {
+      this.flush().catch(() => null);
+    } else if (!this.flushOnTimeout) {
+      this.flushOnTimeout = setTimeout(() => this.flush(), 1e3).unref();
+    }
   }
 
   public async flush(): Promise<void> {
-    for (const [command, args] of Object.entries(this.internalState.batchSendCommands)) {
-      delete this.internalState.batchSendCommands[command];
-      if (!args) continue;
-      await this.connection.sendRequest({
-        meta: this.meta,
-        command,
-        args,
-      });
-    }
+    clearTimeout(this.flushOnTimeout);
+    this.flushOnTimeout = null;
+    if (!this.internalState.commandsToRecord.length) return;
+    const recordCommands = [...this.internalState.commandsToRecord];
+    this.internalState.commandsToRecord.length = 0;
+
+    await this.connection.sendRequest({
+      meta: this.meta,
+      command: 'Session.flush',
+      startDate: new Date(),
+      args: [],
+      recordCommands,
+    });
   }
 
   public run<T>(command: string, ...args: any[]): Promise<T> {
+    clearTimeout(this.flushOnTimeout);
+    this.flushOnTimeout = null;
     if (this.connection.isDisconnecting) {
       return Promise.resolve(null);
     }
@@ -71,17 +94,24 @@ export default class CoreCommandQueue {
         convertJsPathArgs(arg);
       }
     }
+    const startTime = new Date();
+    const commandId = this.nextCommandId;
     return this.internalQueue
       .run<T>(async () => {
+        const recordCommands = [...this.internalState.commandsToRecord];
+        this.internalState.commandsToRecord.length = 0;
+
         const response = await this.connection.sendRequest({
           meta: this.meta,
           command,
           args,
+          startDate: startTime,
+          commandId,
+          recordCommands,
         });
 
         let data: T = null;
         if (response) {
-          this.lastCommandId = response.commandId;
           data = response.data;
         }
         return data;
@@ -101,6 +131,6 @@ export default class CoreCommandQueue {
   }
 
   public createSharedQueue(meta: ISessionMeta & { sessionName: string }): CoreCommandQueue {
-    return new CoreCommandQueue(meta, this.connection, this.internalState);
+    return new CoreCommandQueue(meta, this.connection, this.commandCounter, this.internalState);
   }
 }

client/lib/CoreEventHeap.ts

@@ -44,6 +44,7 @@ export default class CoreEventHeap {
 
     const subscriptionPromise = this.connection.sendRequest({
       meta: this.meta,
+      startDate: new Date(),
       command: 'Session.addEventListener',
       args: [jsPath, type, options],
     });
@@ -82,6 +83,7 @@ export default class CoreEventHeap {
     this.connection
       .sendRequest({
         meta: this.meta,
+        startDate: new Date(),
         command: 'Session.removeEventListener',
         args: [listenerId],
       })

client/lib/CoreFrameEnvironment.ts

@@ -44,13 +44,12 @@ export default class CoreFrameEnvironment {
     return await this.commandQueue.run('FrameEnvironment.execJsPath', jsPath);
   }
 
-  public recordDetachedJsPath(index: number, startTime: Date, endTime: Date): void {
-    this.commandQueue.queueBatchedCommand(
-      'FrameEnvironment.recordDetachedJsPaths',
-      index,
-      startTime.getTime(),
-      endTime.getTime(),
-    );
+  public recordDetachedJsPath(index: number, startDate: Date, endDate: Date): void {
+    this.commandQueue.record({
+      commandId: this.commandQueue.nextCommandId,
+      command: 'FrameEnvironment.recordDetachedJsPath',
+      args: [index, startDate.getTime(), endDate.getTime()],
+    });
   }
 
   public async getJsValue<T>(expression: string): Promise<T> {