Repository: trpc/trpc
File: packages/client/src/links/httpSubscriptionLink.ts (TypeScript)

Code before:

          eventSource = new EventSourceWrapper(url, eventSourceOptions);

          const onStarted = () => {
            observer.next({
              result: {
                type: 'started',
              },
              context: {
                // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
                eventSource: eventSource!.getEventSource(),
              },
            });
          };
          eventSource.addEventListener('open', onStarted, { once: true });

          const iterable = sseStreamConsumer<
            Partial<{
              id?: string;
              data: unknown;
            }>
          >({
            from: eventSource,
            deserialize: transformer.output.deserialize,
            tryHandleError: async (ev) => {
              if (
                'status' in ev &&
                typeof ev.status === 'number' &&
                typeof opts.shouldReinitialize === 'function' &&
                eventSource &&
                (await opts.shouldReinitialize(ev.status, ev))
              ) {
                eventSource.restart(
                  url,
                  await resultOf(opts.eventSourceOptions),
                );

                return true;
              }

              return false;
            },
          });

          for await (const chunk of iterable) {
            if (!chunk.ok) {
              // TODO: handle in https://github.com/trpc/trpc/issues/5871
              continue;
            }
            const chunkData = chunk.data;


Review comment (suggestion):
Important spec thing given the change in how this is run "once". When we do reinitialize, this callback will be called again with the new EventSource instance.

That felt right in-case anybody depends on holding that instance so they can update the reference and re-attach their own listeners

Code after:
            path,
            type,
            signal: null,
          });

          const eventSourceOptions = await resultOf(opts.eventSourceOptions);
          /* istanbul ignore if -- @preserve */
          if (unsubscribed) {
            // already unsubscribed - rare race condition
            return;
          }

          eventSource = new EventSourceWrapper(url, eventSourceOptions);

          const onStarted = () => {
            observer.next({
              result: {
                type: 'started',
              },
              context: {
                // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
                eventSource: eventSource!.getEventSource(),
              },
            });
          };
          eventSource.addEventListener('open', onStarted, { once: true });

          const iterable = sseStreamConsumer<
            Partial<{
              id?: string;
              data: unknown;
            }>
          >({
            from: eventSource,
            deserialize: transformer.output.deserialize,
            tryHandleError: async (ev) => {
              if (
                typeof opts.shouldRecreateOnError !== 'function' ||
                !eventSource
              ) {
                return false;
              }

              const recreateOnErrorOpts = createRecreateOnErrorOpts(ev);

              const shouldRestart = await opts.shouldRecreateOnError(
                recreateOnErrorOpts,
              );

              if (!shouldRestart) {
                return false;