fix

src/app.ts

@@ -40,8 +40,7 @@ app.middleware(YGOProCtosJoinGame, async (msg, client, next) => {
   await client.sendChat(`Your IP: ${client.ip}`);
   await client.sendChat(`Your physical IP: ${client.physicalIp()}`);
   await client.sendChat(`Your pass: ${msg.pass}`);
-  await client.die(
+  return client.die(
     'This server is for testing purposes only. Please use an official server to play the game.',
   );
-  return undefined;
 });

src/client.ts

-import {
-  filter,
-  firstValueFrom,
-  merge,
-  Observable,
-  Subject,
-  timeout,
-  TimeoutError,
-} from 'rxjs';
-import { take } from 'rxjs/operators';
+import { filter, merge, Observable, Subject } from 'rxjs';
+import { map, share, take, takeUntil } from 'rxjs/operators';
 import { Context } from './app';
 import {
   YGOProCtos,
@@ -18,7 +10,6 @@ import {
   YGOProStocErrorMsg,
 } from 'ygopro-msg-encode';
 import { YGOProProtoPipe } from './utility/ygopro-proto-pipe';
-import { ClassType } from 'nfkit';
 import { I18nService } from './services/i18n';
 import { Chnroute } from './services/chnroute';
 
@@ -33,36 +24,48 @@ export abstract class Client {
   isLocal = false;
 
   private logger = this.ctx.createLogger(this.constructor.name);
-  private receiveSubject?: Subject<YGOProCtosBase>;
   private disconnectSubject = new Subject<void>();
-  private manuallyDisconnected = false;
 
   constructor(protected ctx: Context) {}
 
+  receive$!: Observable<YGOProCtosBase>;
+  disconnect$!: Observable<void>;
+
   init() {
-    this.onDisconnect().subscribe(() => {
-      if (this.receiveSubject) {
-        this.receiveSubject.complete();
-        this.receiveSubject = undefined;
-      }
-    });
+    this.disconnect$ = merge(
+      this.disconnectSubject.asObservable(),
+      this._onDisconnect(),
+    ).pipe(take(1));
+    this.receive$ = this._receive().pipe(
+      YGOProProtoPipe(YGOProCtos, {
+        onError: (error) => {
+          this.logger.warn(
+            { ip: this.loggingIp() },
+            `Protocol decode error: ${error.message}`,
+          );
+        },
+      }),
+      filter((msg) => {
+        if (!msg) {
+          this.logger.warn(
+            { ip: this.loggingIp() },
+            `Received invalid message, skipping`,
+          );
+          return false;
+        }
+        return true;
+      }),
+      map((s) => s!),
+      takeUntil(this.disconnect$),
+      share(),
+    );
   }
 
-  async disconnect(): Promise<void> {
-    this.manuallyDisconnected = true;
+  disconnect() {
     this.disconnectSubject.next();
     this.disconnectSubject.complete();
-    await this._disconnect();
-  }
-
-  onDisconnect(): Observable<void> {
-    if (this.manuallyDisconnected) {
-      return this.disconnectSubject.asObservable();
-    }
-    return merge(
-      this.disconnectSubject.asObservable(),
-      this._onDisconnect(),
-    ).pipe(take(1));
+    this._disconnect().then();
+    return undefined;
   }
 
   async send(data: YGOProStocBase) {
@@ -97,49 +100,13 @@ export abstract class Client {
         code: 9,
       }),
     );
-    this.disconnect().then();
+    return this.disconnect();
   }
 
   loggingIp() {
     return this.ip || this.physicalIp() || 'unknown';
   }
 
-  receive(): Observable<YGOProCtosBase> {
-    // Create subject on first call and reuse it
-    if (!this.receiveSubject) {
-      this.receiveSubject = new Subject<YGOProCtosBase>();
-
-      this._receive()
-        .pipe(
-          YGOProProtoPipe(YGOProCtos, {
-            onError: (error) => {
-              this.logger.warn(
-                { ip: this.loggingIp() },
-                `Protocol decode error: ${error.message}`,
-              );
-            },
-          }),
-          filter((msg) => {
-            if (!msg) {
-              this.logger.warn(
-                { ip: this.loggingIp() },
-                `Received invalid message, skipping`,
-              );
-              return false;
-            }
-            return true;
-          }),
-        )
-        .subscribe({
-          next: (data) => this.receiveSubject?.next(data!),
-          error: (err) => this.receiveSubject?.error(err),
-          complete: () => this.receiveSubject?.complete(),
-        });
-    }
-
-    return this.receiveSubject.asObservable();
-  }
-
   name = '';
   vpass = '';
   name_vpass = '';

src/services/client-handler.ts

@@ -39,7 +39,7 @@ export class ClientHandler {
     client.init();
     try {
       client.init();
-      client.receive().subscribe(async (msg) => {
+      client.receive$.subscribe(async (msg) => {
         try {
           await this.ctx.dispatch(msg, client);
         } catch (e) {
@@ -49,7 +49,7 @@ export class ClientHandler {
         }
       });
     } catch {
-      client.disconnect().then();
+      client.disconnect();
     }
   }
 }

src/services/client-version-check.ts

@@ -30,7 +30,7 @@ export class ClientVersionCheck {
           code: YGOPRO_VERSION,
         }),
       );
-      await client.disconnect();
+      return client.disconnect();
     });
   }
 }

src/services/ip-resolver.ts

@@ -135,9 +135,7 @@ export class IpResolver {
         { ip: newIp, badCount, connectCount },
         'Rejecting bad IP',
       );
-      client.disconnect().catch((err) => {
-        this.logger.error({ err }, 'Error disconnecting client');
-      });
+      client.disconnect();
       return true;
     }
 

src/utility/ygopro-proto-pipe.ts

@@ -77,7 +77,7 @@ export const YGOProProtoPipe = <
             const total = 2 + len;
             if (total > maxFrameBytes) {
               opts.onError?.(new Error('Frame size exceeds maxFrameBytes'));
-              state.skipBytes += total;
+              state.skipBytes = (state.skipBytes ?? 0) + total;
               continue;
             }