Theo's Typesafe Cult•3mo ago

SSE httpSubscriptionLink (tRCP 11 - next 14) help please :)

Hi everyone, I have turborepo with a nextjs/prisma/tRCP and I am trying to setup sse subs (based myself on tRPC doc example). Atm I have 2 subs and a Redis client based generator

export async function* createRedisSubscription({
  redis,
  signal,
  channel,
}: {
  redis: Redis;
  signal?: AbortSignal;
  channel: string;
}) {
  const subscriber = redis.duplicate(); 
  await subscriber.subscribe(channel);
  console.log(`Subscribed to channel: ${channel}`);

  try {
    const handleMessage = (channelId: string, message: string) => {
      if (channelId === channel) {
        console.log(`Received message on channel ${channel}:`, message);
        return JSON.parse(message);
      }
      console.log(`Ignored message from channel ${channelId}`);
      return null;
    };

    while (!signal?.aborted) {
      const message = await new Promise((resolve) => {
        const messageHandler = (channelId: string, message: string) => {
          const parsedMessage = handleMessage(channelId, message);
          if (parsedMessage) {
            resolve(parsedMessage);
            subscriber.off("message", messageHandler);
          }
        };
        subscriber.on("message", messageHandler);
      });

      
      if (message) {
        yield message;
      }
      await new Promise((resolve) => setTimeout(resolve, 100)); 
    }
  } finally {
    console.log(`Cleaning up subscription for channel: ${channel}`);
    await subscriber.unsubscribe(channel);
    await subscriber.quit();
  }
}

export async function* createRedisSubscription({
  redis,
  signal,
  channel,
}: {
  redis: Redis;
  signal?: AbortSignal;
  channel: string;
}) {
  const subscriber = redis.duplicate(); 
  await subscriber.subscribe(channel);
  console.log(`Subscribed to channel: ${channel}`);

  try {
    const handleMessage = (channelId: string, message: string) => {
      if (channelId === channel) {
        console.log(`Received message on channel ${channel}:`, message);
        return JSON.parse(message);
      }
      console.log(`Ignored message from channel ${channelId}`);
      return null;
    };

    while (!signal?.aborted) {
      const message = await new Promise((resolve) => {
        const messageHandler = (channelId: string, message: string) => {
          const parsedMessage = handleMessage(channelId, message);
          if (parsedMessage) {
            resolve(parsedMessage);
            subscriber.off("message", messageHandler);
          }
        };
        subscriber.on("message", messageHandler);
      });

      
      if (message) {
        yield message;
      }
      await new Promise((resolve) => setTimeout(resolve, 100)); 
    }
  } finally {
    console.log(`Cleaning up subscription for channel: ${channel}`);
    await subscriber.unsubscribe(channel);
    await subscriber.quit();
  }
}

The publish is triggered programmatically by a mutation. In isolation or if used in a single route the subs seems to be working, but if I split the subscription to be used from multiple routes eg app/page - app/stations/[hostname]/page and I open these pages in multiple tabs on reload requests get stuck on pending. Can anybody please try to help me to understand where the problem can come from?

0 Replies

No replies yetBe the first to reply to this messageJoin

Gaming

Programming

SSE httpSubscriptionLink (tRCP 11 - next 14) help please :)