fix: stabilized stream initialization

package.json

@@ -1,6 +1,6 @@
 {
   "name": "easy-network-stub",
-  "version": "7.0.3",
+  "version": "7.1.0",
   "repository": "https://github.com/LoaderB0T/easy-network-stub.git",
   "author": {
     "name": "Janik Schumacher (LoaderB0T)",

src/stream/http-stream.ts

@@ -10,16 +10,42 @@ export class HttpStreamResponse {
   private _httpServer?: http.Server;
   private _port?: number;
   private _res?: Res;
+  private _initializeState: 'none' | 'initializing' | 'initialized' = 'none';
+  private _clientConnected = false;
+  private readonly _initializedCallbacks: Array<() => void> = [];
+  private readonly _clientConnectedCallbacks: Array<() => void> = [];
   private readonly _kind: StreamKind;
 
   constructor(kind: StreamKind = 'eventStream') {
     this._kind = kind;
   }
 
+  public async waitForClientConnection() {
+    if (this._clientConnected) {
+      return;
+    }
+    await new Promise<void>(resolve => {
+      this._clientConnectedCallbacks.push(resolve);
+    });
+  }
+
   public async init() {
+    if (this._initializeState === 'initialized') {
+      return;
+    }
+    if (this._initializeState === 'initializing') {
+      return new Promise<void>(resolve => {
+        this._initializedCallbacks.push(resolve);
+      });
+    }
+    this._initializeState = 'initializing';
     const { server, port } = await this._createServer();
     this._httpServer = server;
     this._port = port;
+    setTimeout(() => {
+      this._initializeState = 'initialized';
+      this._initializedCallbacks.forEach(cb => cb());
+    }, 100);
   }
 
   private async _createServer() {
@@ -28,6 +54,10 @@ export class HttpStreamResponse {
     const server = http.createServer(async (req, res) => {
       this._sseStart(res);
       this._res = res;
+      setTimeout(() => {
+        this._clientConnected = true;
+        this._clientConnectedCallbacks.forEach(cb => cb());
+      }, 100);
     });
 
     // eslint-disable-next-line no-constant-condition

src/stream/stream-response-handler.ts

@@ -14,7 +14,9 @@ export class StreamResponseHandler extends CustomResponseHandler {
     req.reply({ statusCode: 302, headers: { location: this._stream.url } });
   }
 
-  public send<T>(fragment: T): void {
+  public async send<T>(fragment: T): Promise<void> {
+    await this._stream.init();
+    await this._stream.waitForClientConnection();
     this._stream.send(fragment);
   }