Method subscribe

packages/jstack-shared/src/socket.ts:136–217

(room: string)

  }

  private async subscribe(room: string): Promise<void> {
    return new Promise(async (resolve, reject) => {
      try {
        const controller = new AbortController()
        this.controllers.set(room, controller)

        // initialize heartbeat
        this.lastPingTimes.set(room, Date.now())

        const stream = await fetch(`${this.redisUrl}/subscribe/${room}`, {
          headers: {
            Authorization: `Bearer ${this.redisToken}`,
            accept: "text/event-stream",
          },
          signal: controller.signal,
        })

        const reader = stream.body?.getReader()
        const decoder = new TextDecoder()
        let buffer = ""

        while (reader) {
          const { done, value } = await reader.read()

          // continue execution once connection is established
          // otherwise subscription below would be blocking
          resolve()

          if (done) break

          const chunk = decoder.decode(value)
          buffer += chunk

          const messages = buffer.split("\n")
          buffer = messages.pop() || ""

          for (const message of messages) {
            logger.info("Received message:", message)
            if (message.startsWith("data: ")) {
              const data = message.slice(6)
              try {
                // extract payload from message format: message,room,payload
                // skip first two commas to get the start of the payload
                const firstCommaIndex = data.indexOf(",")
                const secondCommaIndex = data.indexOf(",", firstCommaIndex + 1)

                if (firstCommaIndex === -1 || secondCommaIndex === -1) {
                  logger.warn("Invalid message format - missing commas")
                  continue
                }

                const payloadStr = data.slice(secondCommaIndex + 1)

                if (!payloadStr) {
                  logger.warn("Missing payload in message")
                  continue
                }