hasanaktasTR Posts

hasanaktasTR

•Created by hasanaktasTR on 10/14/2024 in #❓-help

Help for subscription with redis

My subscription is currently as in the function below. It works, but it doesn't make sense to check the response from redis once every second. How can I make the subscription more performant? I couldn't find an example using redis or rabbitmq.

const listSubscription= profileProcedure
    .input(
      z.object({
        matchId: z.string(),
      }),
    )
    .subscription(async function* ({ signal, input, ctx }) {
      console.log("connection opened");

      await ctx.eventSub.subscribe(input.matchId); //eventsub is a redis client
    
      const messageQueue: any[] = []; 

      const messageHandler = (channel: string, message: string) => {
        const post = JSON.parse(message);
        messageQueue.push(post); 
      };
      
      ctx.eventSub.on("message", messageHandler);

      
      try {
        while (!signal?.aborted) {
          if (messageQueue.length > 0) {
            yield messageQueue.shift();
          }
          await new Promise((resolve) => setTimeout(resolve, 1000)); 
        }
      } finally {
        console.log("connection closed");
        ctx.eventSub.off("message", messageHandler);
        ctx.eventSub.unsubscribe(input.matchId);
      }
    })

const listSubscription= profileProcedure
    .input(
      z.object({
        matchId: z.string(),
      }),
    )
    .subscription(async function* ({ signal, input, ctx }) {
      console.log("connection opened");

      await ctx.eventSub.subscribe(input.matchId); //eventsub is a redis client
    
      const messageQueue: any[] = []; 

      const messageHandler = (channel: string, message: string) => {
        const post = JSON.parse(message);
        messageQueue.push(post); 
      };
      
      ctx.eventSub.on("message", messageHandler);

      
      try {
        while (!signal?.aborted) {
          if (messageQueue.length > 0) {
            yield messageQueue.shift();
          }
          await new Promise((resolve) => setTimeout(resolve, 1000)); 
        }
      } finally {
        console.log("connection closed");
        ctx.eventSub.off("message", messageHandler);
        ctx.eventSub.unsubscribe(input.matchId);
      }
    })

12 replies