Enhance SCTP, SctpReconfig, StreamManager, and SCTPTimerManager classes with DTO methods for improved data handling and organization

Author: shinyoshiaki

.vscode/settings.json

@@ -83,5 +83,5 @@
   ],
   "typescript.tsdk": "node_modules/typescript/lib",
   "biome.lsp.bin": "node_modules/@biomejs/biome/bin/biome",
-  "prettier.enable": false
+  "prettier.enable": false,
 }

packages/sctp/src/reconfig.ts

@@ -18,6 +18,11 @@ import { tsnMinusOne, tsnPlusOne } from "./util";
 
 const log = debug("packages/sctp/src/reconfig.ts");
 
+interface Deps {
+  transmitter: SCTPTransmitter;
+  timerManager: SCTPTimerManager;
+}
+
 export class SctpReconfig {
   // # reconfiguration
 
@@ -27,12 +32,16 @@ export class SctpReconfig {
   reconfigResponseSeq = 0;
   reconfigRequest?: OutgoingSSNResetRequestParam;
   reconfigQueue: number[] = [];
+
+  private transmitter: SCTPTransmitter;
+  private timerManager: SCTPTimerManager;
+
   readonly onReconfigStreams = new Event<[number[]]>();
 
-  constructor(
-    private transmitter: SCTPTransmitter,
-    private timerManager: SCTPTimerManager,
-  ) {
+  constructor(deps: Deps) {
+    this.transmitter = deps.transmitter;
+    this.timerManager = deps.timerManager;
+
     this.reconfigRequestSeq = this.transmitter.localTsn;
 
     this.timerManager.onReconfigExpired.subscribe(async (reconfigFailures) => {
@@ -154,4 +163,35 @@ export class SctpReconfig {
     this.reconfigResponseSeq = add.requestSequence;
     await this.sendReconfigParam(res);
   }
+
+  toDto(): ReconfigDto {
+    return {
+      reconfigRequestSeq: this.reconfigRequestSeq,
+      reconfigResponseSeq: this.reconfigResponseSeq,
+      reconfigRequest: this.reconfigRequest?.bytes,
+      reconfigQueue: this.reconfigQueue,
+    };
+  }
+
+  static fromDto(dto: ReconfigDto, deps: Deps): SctpReconfig {
+    const reconfig = new SctpReconfig(deps);
+    reconfig.reconfigRequestSeq = dto.reconfigRequestSeq;
+    reconfig.reconfigResponseSeq = dto.reconfigResponseSeq;
+    reconfig.reconfigQueue = dto.reconfigQueue;
+
+    if (dto.reconfigRequest) {
+      reconfig.reconfigRequest = OutgoingSSNResetRequestParam.parse(
+        dto.reconfigRequest,
+      );
+    }
+
+    return reconfig;
+  }
+}
+
+export interface ReconfigDto {
+  reconfigRequestSeq: number;
+  reconfigResponseSeq: number;
+  reconfigRequest: Buffer<ArrayBuffer> | undefined;
+  reconfigQueue: number[];
 }

packages/sctp/src/sctp.ts

@@ -47,13 +47,14 @@ import {
   StreamAddOutgoingParam,
   type StreamParam,
 } from "./param";
-import { SctpReconfig } from "./reconfig";
-import { StreamManager } from "./stream";
-import { SCTPTimerManager } from "./timer";
+import { type ReconfigDto, SctpReconfig } from "./reconfig";
+import { StreamManager, type StreamManagerStateDTO } from "./stream";
+import { type SCTPTimerDto, SCTPTimerManager } from "./timer";
 import {
   type SCTPConnectionState,
   SCTPConnectionStates,
   SCTPTransmitter,
+  type SCTPTransmitterDto,
 } from "./transmitter";
 import type { Transport } from "./transport";
 import { tsnMinusOne, tsnPlusOne } from "./util";
@@ -63,27 +64,26 @@ const log = debug("werift/sctp/sctp");
 // SSN: Stream Sequence Number
 
 export class SCTP {
-  associationState = SCTP_STATE.CLOSED;
-  started = false;
-  isServer = true;
-
   private readonly hmacKey = randomBytes(16);
   private readonly localPartialReliability = true;
   private readonly localVerificationTag = random32();
 
+  associationState = SCTP_STATE.CLOSED;
+  started = false;
+  isServer = true;
   remoteExtensions: number[] = [];
   remotePartialReliability = true;
+  protected lastReceivedTsn?: number; // Transmission Sequence Number
+  protected sackDuplicates: number[] = [];
+  protected sackMisOrdered = new Set<number>();
+  protected sackNeeded = false;
 
-  private lastReceivedTsn?: number; // Transmission Sequence Number
-  private sackDuplicates: number[] = [];
-  private sackMisOrdered = new Set<number>();
-  private sackNeeded = false;
   private sackTimeout: NodeJS.Immediate | undefined;
 
-  readonly transmitter: SCTPTransmitter;
-  readonly timerManager: SCTPTimerManager;
-  readonly reconfig: SctpReconfig;
-  readonly stream = new StreamManager();
+  transmitter: SCTPTransmitter;
+  timerManager: SCTPTimerManager;
+  reconfig: SctpReconfig;
+  stream: StreamManager;
 
   readonly stateChanged: {
     [key in SCTPConnectionState]: Event<[]>;
@@ -98,16 +98,29 @@ export class SCTP {
     public transport: Transport,
     public port = 5000,
   ) {
-    this.transport.onData = (buf) => {
-      this.handleData(buf);
-    };
     this.timerManager = new SCTPTimerManager({
       sendChunk: async (chunk) => {
         await this.transmitter.sendChunk(chunk);
       },
     });
-    this.transmitter = new SCTPTransmitter(transport, this.timerManager, port);
-    this.reconfig = new SctpReconfig(this.transmitter, this.timerManager);
+    this.transmitter = new SCTPTransmitter({
+      transport,
+      timerManager: this.timerManager,
+      port,
+    });
+    this.reconfig = new SctpReconfig({
+      transmitter: this.transmitter,
+      timerManager: this.timerManager,
+    });
+    this.stream = new StreamManager();
+
+    this.setup();
+  }
+
+  protected setup() {
+    this.transport.onData = (buf) => {
+      this.handleData(buf);
+    };
 
     this.timerManager.onT1Expired.subscribe(() => {
       this.setState(SCTP_STATE.CLOSED);
@@ -614,6 +627,73 @@ export class SCTP {
       this.reconfig.transmitReconfigRequest(this.associationState);
     }
   }
+
+  toDto(): SCTPDto {
+    return {
+      associationState: this.associationState,
+      started: this.started,
+      isServer: this.isServer,
+      remoteExtensions: this.remoteExtensions,
+      remotePartialReliability: this.remotePartialReliability,
+      port: this.port,
+      lastReceivedTsn: this.lastReceivedTsn,
+      sackDuplicates: this.sackDuplicates,
+      sackMisOrdered: Array.from(this.sackMisOrdered),
+      sackNeeded: this.sackNeeded,
+      transmitter: this.transmitter.toDto(),
+      timerManager: this.timerManager.toDto(),
+      reconfig: this.reconfig.toDto(),
+      stream: this.stream.toDto(),
+    };
+  }
+
+  fromDto(dto: SCTPDto, transport: Transport) {
+    const sctp = new SCTP(transport);
+    sctp.associationState = dto.associationState;
+    sctp.started = dto.started;
+    sctp.isServer = dto.isServer;
+    sctp.remoteExtensions = dto.remoteExtensions;
+    sctp.remotePartialReliability = dto.remotePartialReliability;
+    sctp.port = dto.port;
+    sctp.lastReceivedTsn = dto.lastReceivedTsn;
+    sctp.sackDuplicates = dto.sackDuplicates;
+    sctp.sackMisOrdered = new Set(dto.sackMisOrdered);
+    sctp.sackNeeded = dto.sackNeeded;
+
+    sctp.timerManager = SCTPTimerManager.fromDto(dto.timerManager, {
+      sendChunk: async (chunk) => {
+        await sctp.transmitter.sendChunk(chunk);
+      },
+    });
+    sctp.transmitter = SCTPTransmitter.fromDto(dto.transmitter, {
+      timerManager: sctp.timerManager,
+      transport,
+    });
+    sctp.reconfig = SctpReconfig.fromDto(dto.reconfig, {
+      timerManager: sctp.timerManager,
+      transmitter: sctp.transmitter,
+    });
+    sctp.stream = StreamManager.fromDto(dto.stream);
+    sctp.setup();
+    return sctp;
+  }
+}
+
+export interface SCTPDto {
+  associationState: SCTP_STATE;
+  started: boolean;
+  isServer: boolean;
+  remoteExtensions: number[];
+  remotePartialReliability: boolean;
+  port: number;
+  lastReceivedTsn?: number;
+  sackDuplicates: number[];
+  sackMisOrdered: number[];
+  sackNeeded: boolean;
+  transmitter: SCTPTransmitterDto;
+  timerManager: SCTPTimerDto;
+  reconfig: ReconfigDto;
+  stream: StreamManagerStateDTO;
 }
 
 export class RTCSctpCapabilities {

packages/sctp/src/stream.ts

@@ -1,8 +1,9 @@
-import type {
-  DataChunk,
-  ForwardTsnChunk,
-  InitAckChunk,
-  InitChunk,
+import {
+  type DataChunk,
+  type ForwardTsnChunk,
+  type InitAckChunk,
+  type InitChunk,
+  parseChunk,
 } from "./chunk";
 import {
   MAX_STREAMS,
@@ -113,6 +114,44 @@ export class StreamManager {
       this.receive(...message);
     }
   }
+
+  toDto(): StreamManagerStateDTO {
+    return {
+      advertisedRwnd: this.advertisedRwnd,
+      inboundStreams: Object.fromEntries(
+        Object.entries(this.inboundStreams).map(([k, v]) => [k, v.toDto()]),
+      ),
+      _inboundStreamsCount: this._inboundStreamsCount,
+      outboundStreamSeq: this.outboundStreamSeq,
+      _outboundStreamsCount: this._outboundStreamsCount,
+    };
+  }
+
+  static fromDto(dto: StreamManagerStateDTO): StreamManager {
+    const streamManager = new StreamManager();
+    streamManager.advertisedRwnd = dto.advertisedRwnd;
+    streamManager._inboundStreamsCount = dto._inboundStreamsCount;
+    streamManager.outboundStreamSeq = dto.outboundStreamSeq;
+    streamManager._outboundStreamsCount = dto._outboundStreamsCount;
+
+    for (const [k, v] of Object.entries(dto.inboundStreams)) {
+      streamManager.inboundStreams[Number(k)] = InboundStream.fromDto(v);
+    }
+
+    return streamManager;
+  }
+}
+
+export interface StreamManagerStateDTO {
+  advertisedRwnd: number;
+  inboundStreams: {
+    [k: string]: InboundStreamDTO;
+  };
+  _inboundStreamsCount: number;
+  outboundStreamSeq: {
+    [streamId: number]: number;
+  };
+  _outboundStreamsCount: number;
 }
 
 export class InboundStream {
@@ -223,4 +262,25 @@ export class InboundStream {
     this.reassembly = this.reassembly.slice(pos + 1);
     return size;
   }
+
+  toDto(): InboundStreamDTO {
+    return {
+      reassembly: this.reassembly.map((chunk) => chunk.bytes),
+      streamSequenceNumber: this.streamSequenceNumber,
+    };
+  }
+
+  static fromDto(dto: InboundStreamDTO): InboundStream {
+    const stream = new InboundStream();
+    stream.reassembly = dto.reassembly.map(
+      (chunk) => parseChunk(chunk).chunk as DataChunk,
+    );
+    stream.streamSequenceNumber = dto.streamSequenceNumber;
+    return stream;
+  }
+}
+
+interface InboundStreamDTO {
+  reassembly: Buffer<ArrayBuffer>[];
+  streamSequenceNumber: number;
 }

packages/sctp/src/timer.ts

@@ -333,7 +333,7 @@ export class SCTPTimerManager {
   }
 }
 
-interface SCTPTimerDto {
+export interface SCTPTimerDto {
   t1Chunk: Buffer | undefined;
   t1Failures: number;
   t2Chunk: Buffer | undefined;