Update network streams (performance improvement).

Author: storycraft

src/config/index.ts renamed to src/config.ts

@@ -5,9 +5,9 @@
  */
 
 import { SessionConfig } from '../config';
-import { LocoSession, PacketResData, SessionFactory } from '../network/request-session';
-import { DefaultRes, DefaultReq, AsyncCommandResult } from '../request';
-import { LocoPacket } from '../packet';
+import { ConnectionSession, PacketResData, SessionFactory } from '../network/request-session';
+import { DefaultReq, AsyncCommandResult, DefaultRes } from '../request';
+import { BiStream } from '../stream';
 
 /**
  * Hook incoming datas
@@ -24,11 +24,6 @@ export interface SessionHook {
    */
   onRequest: (method: string, data: DefaultReq) => void;
 
-  /**
-   * Hook loco packet
-   */
-  onSendPacket: (packet: LocoPacket) => void;
-
   onClose(): () => void;
 
 }
@@ -41,60 +36,51 @@ export class HookedSessionFactory implements SessionFactory {
 
   }
 
-  async createSession(config: SessionConfig): AsyncCommandResult<LocoSession> {
-    const sessionRes = await this._factory.createSession(config);
+  async connect(config: SessionConfig): AsyncCommandResult<ConnectionSession> {
+    const sessionRes = await this._factory.connect(config);
     if (!sessionRes.success) return sessionRes;
 
-    return { status: sessionRes.status, success: true, result: new HookedLocoSession(sessionRes.result, this._hook) };
+    return { status: sessionRes.status, success: true, result: new InspectSession(sessionRes.result, this._hook) };
   }
 }
 
-/**
- * Hook loco session
- */
-export class HookedLocoSession implements LocoSession {
-  constructor(private _session: LocoSession, public hook: Partial<SessionHook> = {}) {
+export class InspectSession implements ConnectionSession {
+
+  constructor(
+    private _session: ConnectionSession,
+    private _hook: Partial<SessionHook> = {}
+  ) {
 
   }
 
+  get stream(): BiStream {
+    return this._session.stream;
+  }
+  
+  request<T = DefaultRes>(method: string, data: DefaultReq): Promise<DefaultRes & T> {
+    if (this._hook.onRequest) this._hook.onRequest(method, data);
+
+    return this._session.request(method, data);
+  }
+
   listen(): AsyncIterableIterator<PacketResData> {
-    const hook = this.hook;
     const iterator = this._session.listen();
 
     return {
-      [Symbol.asyncIterator]() {
+      [Symbol.asyncIterator](): AsyncIterableIterator<PacketResData> {
         return this;
       },
 
-      async next(): Promise<IteratorResult<PacketResData>> {
+      next: async (): Promise<IteratorResult<PacketResData>> => {
         const next = await iterator.next();
 
-        if (!next.done && hook.onData) {
-          const { method, data, push } = next.value;
-
-          hook.onData(method, data, push);
+        if (!next.done && this._hook.onData) {
+          this._hook.onData(next.value.method, next.value.data, next.value.push);
         }
 
         return next;
-      },
+      }
     };
   }
 
-  request<T = DefaultRes>(method: string, data: DefaultReq): Promise<DefaultRes & T> {
-    if (this.hook.onRequest) this.hook.onRequest(method, data);
-
-    return this._session.request(method, data);
-  }
-
-  sendPacket(packet: LocoPacket): Promise<LocoPacket> {
-    if (this.hook.onSendPacket) this.hook.onSendPacket(packet);
-
-    return this._session.sendPacket(packet);
-  }
-
-  close(): void {
-    if (this.hook.onClose) this.hook.onClose();
-
-    this._session.close();
-  }
 }

src/hook/session-factory-hook.ts

@@ -10,15 +10,18 @@ export interface StreamHook {
 
   /**
    * Hook data write
-   * @param data
+   *
+   * @param data Write data
    */
   onWrite(data: Uint8Array): void;
 
   /**
    * Hook data read
-   * @param buf
+   *
+   * @param buf Read buffer
+   * @param read Read size
    */
-  onRead(buf: Uint8Array): void;
+  onRead(buf: Uint8Array, read: number | null): void;
 
   onClose(): void;
 
@@ -41,7 +44,8 @@ export class HookedStream implements BiStream {
 
   async read(buffer: Uint8Array): Promise<number | null> {
     const read = await this._stream.read(buffer);
-    if (this.hook.onRead) this.hook.onRead(buffer);
+
+    if (this.hook.onRead) this.hook.onRead(buffer, read);
 
     return read;
   }

src/hook/stream-hook.ts

@@ -21,28 +21,24 @@ export class LocoPacketCodec {
     return this._stream;
   }
 
-  send(packet: LocoPacket): Promise<number> {
+  write(packet: LocoPacket): Promise<number> {
     const packetBuffer = new ArrayBuffer(22 + packet.data[1].byteLength);
     const packetArray = new Uint8Array(packetBuffer);
-    const namebuffer = new Uint8Array(11);
     const view = new DataView(packetBuffer);
 
     view.setUint32(0, packet.header.id, true);
     view.setUint16(4, packet.header.status & 0xffff, true);
-    view.setUint8(17, packet.data[0] & 0xff);
-    view.setUint32(18, packet.data[1].byteLength, true);
 
-    const nameLen = Math.min(packet.header.method.length, 11);
-    const nameList: number[] = [];
-    for (let i = 0; i < nameLen; i++) {
-      const code = packet.header.method.charCodeAt(i);
+    for (let i = 0; i < 11; i++) {
+      const code = packet.header.method.charCodeAt(i) || 0;
 
       if (code > 0xff) throw new Error('Invalid ASCII code at method string');
-      nameList.push(code);
+      packetArray[6 + i] = code;
     }
-    namebuffer.set(nameList, 0);
 
-    packetArray.set(namebuffer, 6);
+    view.setUint8(17, packet.data[0] & 0xff);
+    view.setUint32(18, packet.data[1].byteLength, true);
+
     packetArray.set(packet.data[1], 22);
 
     return this._stream.write(packetArray);
@@ -71,20 +67,4 @@ export class LocoPacketCodec {
       data: [dataType, data],
     };
   }
-
-  iterate(): AsyncIterableIterator<LocoPacket> {
-    return {
-      [Symbol.asyncIterator](): AsyncIterableIterator<LocoPacket> {
-        return this;
-      },
-
-      next: async(): Promise<IteratorResult<LocoPacket>> => {
-        if (this._stream.ended) return { done: true, value: null }
-
-        const read = await this.read();
-        if (!read) return { done: true, value: null };
-        return { done: false, value: read };
-      },
-    };
-  }
 }

src/network/loco-packet-codec.ts

@@ -45,42 +45,47 @@ export class LocoPacketDispatcher {
       this._packetMap.set(packet.header.id, [resolve, reject]);
     });
 
-    await this._codec.send(packet);
+    await this._codec.write(packet);
 
     return promise;
   }
 
   /**
-   * Listen and process incoming packets.
+   * Read one packet and process it.
+   */
+  async read(): Promise<PacketRes | undefined> {
+    const packet = await this._codec.read();
+    if (!packet) return;
+
+    if (this._packetMap.has(packet.header.id)) {
+      const resolver = this._packetMap.get(packet.header.id);
+      if (resolver) {
+        resolver[0](packet);
+        this._packetMap.delete(packet.header.id);
+      }
+      return { push: false, packet };
+    } else {
+      return { push: true, packet };
+    }
+  }
+
+  /**
+   * Listen and read incoming packets.
    *
    * @return {AsyncIterableIterator<PacketRes>}
    */
   listen(): AsyncIterableIterator<PacketRes> {
-    const packetMap = this._packetMap;
-    const iterator = this._codec.iterate();
-
     return {
       [Symbol.asyncIterator]() {
         return this;
       },
 
-      async next(): Promise<IteratorResult<PacketRes>> {
-        const next = await iterator.next();
-
-        if (next.done) return { done: true, value: null };
-
-        const packet = next.value;
+      next: async (): Promise<IteratorResult<PacketRes>> => {
+        const read = await this.read();
 
-        if (packetMap.has(packet.header.id)) {
-          const resolver = packetMap.get(packet.header.id);
-          if (resolver) {
-            resolver[0](packet);
-            packetMap.delete(packet.header.id);
-          }
-          return { done: false, value: { push: false, packet } };
-        } else {
-          return { done: false, value: { push: true, packet } };
-        }
+        if (!read) return { done: true, value: null };
+        
+        return { done: false, value: read };
       },
     };
   }

src/network/loco-packet-dispatcher.ts

@@ -48,7 +48,7 @@ export class LocoSecureLayer implements BiStream {
     buffer.set(data.subarray(0, readSize), 0);
 
     if (data.byteLength > buffer.byteLength) {
-      this._dataChunks.append(data.slice(buffer.byteLength));
+      this._dataChunks.append(data.subarray(buffer.byteLength));
     }
 
     return readSize;

src/network/loco-secure-layer.ts

@@ -43,6 +43,7 @@ export class PacketAssembler<T, R> {
    * Deconstruct LocoPacket into data.
    * This method can throw error if the type is not supported by codec.
    *
+   * @template R
    * @param {LocoPacket} packet
    * @return {R}
    */

src/network/packet-assembler.ts

@@ -6,7 +6,7 @@
 
 import { SessionConfig } from '../config';
 import { AsyncCommandResult, DefaultReq, DefaultRes } from '../request';
-import { BsonDataCodec, LocoPacket } from '../packet';
+import { BsonDataCodec } from '../packet';
 import { LocoPacketDispatcher } from './loco-packet-dispatcher';
 import { PacketAssembler } from './packet-assembler';
 import { BiStream } from '../stream';
@@ -23,37 +23,38 @@ export interface CommandSession {
 
 }
 
-export interface PacketResData {
+export interface ConnectionSession extends CommandSession {
 
-  method: string;
-  data: DefaultRes;
-  push: boolean;
-
-}
+  /**
+   * Connection stream
+   */
+  readonly stream: BiStream;
 
-export interface LocoSession extends CommandSession {
+  /**
+   * Listen incoming packets
+   */
+  listen(): AsyncIterableIterator<PacketResData>;
 
-  listen(): AsyncIterable<PacketResData> & AsyncIterator<PacketResData>;
+}
 
-  sendPacket(packet: LocoPacket): Promise<LocoPacket>;
+export interface PacketResData {
 
-  close(): void;
+  method: string;
+  data: DefaultRes;
+  push: boolean;
 
 }
 
 /**
- * Create LocoSession using configuration.
+ * Create connection using configuration.
  */
 export interface SessionFactory {
 
-  createSession(config: SessionConfig): AsyncCommandResult<LocoSession>;
+  connect(config: SessionConfig): AsyncCommandResult<ConnectionSession>;
 
 }
 
-/**
- * Holds current loco session.
- */
-export class DefaultLocoSession implements LocoSession {
+export class LocoSession implements ConnectionSession {
   private _assembler: PacketAssembler<DefaultReq, DefaultRes>;
   private _dispatcher: LocoPacketDispatcher;
 
@@ -62,12 +63,16 @@ export class DefaultLocoSession implements LocoSession {
     this._dispatcher = new LocoPacketDispatcher(stream);
   }
 
-  listen(): { [Symbol.asyncIterator](): AsyncIterator<PacketResData>, next(): Promise<IteratorResult<PacketResData>> } {
+  get stream(): BiStream {
+    return this._dispatcher.stream;
+  }
+
+  listen(): AsyncIterableIterator<PacketResData> {
     const iterator = this._dispatcher.listen();
     const assembler = this._assembler;
 
     return {
-      [Symbol.asyncIterator](): AsyncIterator<PacketResData> {
+      [Symbol.asyncIterator](): AsyncIterableIterator<PacketResData> {
         return this;
       },
 
@@ -78,23 +83,12 @@ export class DefaultLocoSession implements LocoSession {
         const { push, packet } = next.value;
 
         return { done: false, value: { push, method: packet.header.method, data: assembler.deconstruct(packet) } };
-      },
+      }
     };
   }
 
   async request<T = DefaultRes>(method: string, data: DefaultReq): Promise<DefaultRes & T> {
     const res = await this._dispatcher.sendPacket(this._assembler.construct(method, data));
     return this._assembler.deconstruct(res) as DefaultRes & T;
   }
-
-  sendPacket(packet: LocoPacket): Promise<LocoPacket> {
-    return this._dispatcher.sendPacket(packet);
-  }
-
-  /**
-   * Close session
-   */
-  close(): void {
-    this._dispatcher.stream.close();
-  }
 }