fix(*): better timeout handling (#44)

tf2pickup-org · May 26, 2022 · f8c8590 · f8c8590
1 parent cd4238c
commit f8c8590
Show file tree

Hide file tree

Showing 6 changed files with 41 additions and 44 deletions.
diff --git a/src/commands/create-channel.ts b/src/commands/create-channel.ts
@@ -11,7 +11,7 @@ import {
   race,
   take,
   throwError,
-  timer,
+  timeout,
 } from 'rxjs';
 
 export const createChannel = async (
@@ -30,6 +30,11 @@ export const createChannel = async (
         ),
         take(1),
         map(channelState => channelState.channelId as number),
+        timeout({
+          first: CommandTimeout,
+          with: () =>
+            throwError(() => new CommandTimedOutError('createChannel')),
+        }),
       ),
       socket.packet.pipe(
         filterPacket(PermissionDenied),
@@ -38,11 +43,6 @@ export const createChannel = async (
           throwError(() => new PermissionDeniedError(permissionDenied)),
         ),
       ),
-      timer(CommandTimeout).pipe(
-        concatMap(() =>
-          throwError(() => new CommandTimedOutError('createChannel')),
-        ),
-      ),
     ),
   );
 

diff --git a/src/commands/fetch-channel-permissions.ts b/src/commands/fetch-channel-permissions.ts
@@ -3,32 +3,22 @@ import { CommandTimedOutError } from '@/errors';
 import { MumbleSocket } from '@/mumble-socket';
 import { filterPacket } from '@/rxjs-operators/filter-packet';
 import { PermissionQuery } from '@tf2pickup-org/mumble-protocol';
-import {
-  concatMap,
-  filter,
-  lastValueFrom,
-  race,
-  take,
-  throwError,
-  timer,
-} from 'rxjs';
+import { filter, lastValueFrom, take, throwError, timeout } from 'rxjs';
 
 export const fetchChannelPermissions = async (
   socket: MumbleSocket,
   channelId: number,
 ): Promise<PermissionQuery> => {
   const ret = lastValueFrom(
-    race(
-      socket.packet.pipe(
-        filterPacket(PermissionQuery),
-        filter(permissionQuery => permissionQuery.channelId === channelId),
-        take(1),
-      ),
-      timer(CommandTimeout).pipe(
-        concatMap(() =>
+    socket.packet.pipe(
+      filterPacket(PermissionQuery),
+      filter(permissionQuery => permissionQuery.channelId === channelId),
+      take(1),
+      timeout({
+        first: CommandTimeout,
+        with: () =>
           throwError(() => new CommandTimedOutError('fetchChannelPermissions')),
-        ),
-      ),
+      }),
     ),
   );
 

diff --git a/src/commands/link-channels.ts b/src/commands/link-channels.ts
@@ -11,7 +11,7 @@ import {
   race,
   take,
   throwError,
-  timer,
+  timeout,
 } from 'rxjs';
 
 export const linkChannels = async (
@@ -26,6 +26,11 @@ export const linkChannels = async (
         filter(channelSync => channelSync.channelId === channelId),
         take(1),
         map(() => void 0),
+        timeout({
+          first: CommandTimeout,
+          with: () =>
+            throwError(() => new CommandTimedOutError('linkChannels')),
+        }),
       ),
       socket.packet.pipe(
         filterPacket(PermissionDenied),
@@ -35,11 +40,6 @@ export const linkChannels = async (
           throwError(() => new PermissionDeniedError(permissionDenied)),
         ),
       ),
-      timer(CommandTimeout).pipe(
-        concatMap(() =>
-          throwError(() => new CommandTimedOutError('linkChannels')),
-        ),
-      ),
     ),
   );
   socket.send(

diff --git a/src/commands/move-user-to-channel.ts b/src/commands/move-user-to-channel.ts
@@ -1,4 +1,5 @@
-import { PermissionDeniedError } from '@/errors';
+import { CommandTimeout } from '@/config';
+import { CommandTimedOutError, PermissionDeniedError } from '@/errors';
 import { MumbleSocket } from '@/mumble-socket';
 import { filterPacket } from '@/rxjs-operators/filter-packet';
 import { PermissionDenied, UserState } from '@tf2pickup-org/mumble-protocol';
@@ -10,6 +11,7 @@ import {
   race,
   take,
   throwError,
+  timeout,
 } from 'rxjs';
 
 export const moveUserToChannel = async (
@@ -28,6 +30,11 @@ export const moveUserToChannel = async (
         ),
         take(1),
         map(() => void 0),
+        timeout({
+          first: CommandTimeout,
+          with: () =>
+            throwError(() => new CommandTimedOutError('moveUserToChannel')),
+        }),
       ),
       socket.packet.pipe(
         filterPacket(PermissionDenied),

diff --git a/src/commands/remove-channel.ts b/src/commands/remove-channel.ts
@@ -14,7 +14,7 @@ import {
   race,
   take,
   throwError,
-  timer,
+  timeout,
 } from 'rxjs';
 
 export const removeChannel = async (
@@ -28,6 +28,11 @@ export const removeChannel = async (
         filter(channelRemove => channelRemove.channelId === channelId),
         take(1),
         map(channelRemove => channelRemove.channelId),
+        timeout({
+          first: CommandTimeout,
+          with: () =>
+            throwError(() => new CommandTimedOutError('removeChannel')),
+        }),
       ),
       socket.packet.pipe(
         filterPacket(PermissionDenied),
@@ -37,11 +42,6 @@ export const removeChannel = async (
           throwError(() => new PermissionDeniedError(permissionDenied)),
         ),
       ),
-      timer(CommandTimeout).pipe(
-        concatMap(() =>
-          throwError(() => new CommandTimedOutError('removeChannel')),
-        ),
-      ),
     ),
   );
   socket.send(ChannelRemove, ChannelRemove.create({ channelId }));

diff --git a/src/commands/unlink-channels.ts b/src/commands/unlink-channels.ts
@@ -11,7 +11,7 @@ import {
   race,
   take,
   throwError,
-  timer,
+  timeout,
 } from 'rxjs';
 
 export const unlinkChannels = async (
@@ -26,6 +26,11 @@ export const unlinkChannels = async (
         filter(channelSync => channelSync.channelId === channelId),
         take(1),
         map(() => void 0),
+        timeout({
+          first: CommandTimeout,
+          with: () =>
+            throwError(() => new CommandTimedOutError('unlinkChannels')),
+        }),
       ),
       socket.packet.pipe(
         filterPacket(PermissionDenied),
@@ -35,11 +40,6 @@ export const unlinkChannels = async (
           throwError(() => new PermissionDeniedError(permissionDenied)),
         ),
       ),
-      timer(CommandTimeout).pipe(
-        concatMap(() =>
-          throwError(() => new CommandTimedOutError('linkChannels')),
-        ),
-      ),
     ),
   );
   socket.send(