Function runRaw

packages/platform/src/Socket.ts:408–489

(handler: (_: string | Uint8Array) => Effect.Effect<_, E, R> | void, opts?: {
      readonly onOpen?: Effect.Effect<void> | undefined
    })

    const closeCodeIsError = options?.closeCodeIsError ?? defaultCloseCodeIsError

    const runRaw = <_, E, R>(handler: (_: string | Uint8Array) => Effect.Effect<_, E, R> | void, opts?: {
      readonly onOpen?: Effect.Effect<void> | undefined
    }) =>
      Effect.scopedWith(Effect.fnUntraced(function*(scope) {
        const fiberSet = yield* FiberSet.make<any, E | SocketError>().pipe(
          Scope.extend(scope)
        )
        const ws = yield* Scope.extend(acquire, scope)
        const run = yield* Effect.provideService(FiberSet.runtime(fiberSet)<R>(), WebSocket, ws)
        let open = false

        function onMessage(event: MessageEvent) {
          if (event.data instanceof Blob) {
            return Effect.promise(() => event.data.arrayBuffer() as Promise<ArrayBuffer>).pipe(
              Effect.andThen((buffer) => handler(new Uint8Array(buffer))),
              run
            )
          }
          const result = handler(event.data)
          if (Effect.isEffect(result)) {
            run(result)
          }
        }
        function onError(cause: Event) {
          ws.removeEventListener("message", onMessage)
          ws.removeEventListener("close", onClose)
          Deferred.unsafeDone(
            fiberSet.deferred,
            Effect.fail(new SocketGenericError({ reason: open ? "Read" : "Open", cause }))
          )
        }
        function onClose(event: globalThis.CloseEvent) {
          ws.removeEventListener("message", onMessage)
          ws.removeEventListener("error", onError)
          Deferred.unsafeDone(
            fiberSet.deferred,
            Effect.fail(
              new SocketCloseError({
                reason: "Close",
                code: event.code,
                closeReason: event.reason
              })
            )
          )
        }

        ws.addEventListener("close", onClose, { once: true })
        ws.addEventListener("error", onError, { once: true })
        ws.addEventListener("message", onMessage)

        if (ws.readyState !== 1) {
          const openDeferred = Deferred.unsafeMake<void>(fiber.id())
          ws.addEventListener("open", () => {
            open = true
            Deferred.unsafeDone(openDeferred, Effect.void)
          }, { once: true })
          yield* Deferred.await(openDeferred).pipe(
            Effect.timeoutFail({