.opencode/node_modules/effect/dist/unstable/socket/Socket.js

import * as Channel from "../../Channel.js";
import * as Context from "../../Context.js";
import * as Deferred from "../../Deferred.js";
import * as Effect from "../../Effect.js";
import * as Exit from "../../Exit.js";
import * as FiberSet from "../../FiberSet.js";
import { constVoid, dual, flow } from "../../Function.js";
import * as Latch from "../../Latch.js";
import * as Layer from "../../Layer.js";
import * as Predicate from "../../Predicate.js";
import * as Pull from "../../Pull.js";
import * as Queue from "../../Queue.js";
import * as Result from "../../Result.js";
import * as Schema from "../../Schema.js";
import * as Scope from "../../Scope.js";
/**
 * Runtime type identifier attached to `Socket` services.
 *
 * @category type IDs
 * @since 4.0.0
 */
export const TypeId = "~effect/socket/Socket";
/**
 * Returns `true` when a value is a `Socket`.
 *
 * @category guards
 * @since 4.0.0
 */
export const isSocket = u => Predicate.hasProperty(u, TypeId);
/**
 * Service tag for bidirectional socket transports.
 *
 * **When to use**
 *
 * Use to access or provide the socket implementation used by programs that
 * read and write frames through the Effect environment.
 *
 * @category services
 * @since 4.0.0
 */
export const Socket = /*#__PURE__*/Context.Service("effect/socket/Socket");
/**
 * Constructs a `Socket` from a raw read loop and scoped writer, deriving binary
 * and string read loops when they are not provided.
 *
 * @category constructors
 * @since 4.0.0
 */
export const make = options => Socket.of({
  [TypeId]: TypeId,
  runRaw: options.runRaw,
  run: options.run ?? ((handler, opts) => options.runRaw(data => typeof data === "string" ? handler(encoder.encode(data)) : data instanceof Uint8Array ? handler(data) : handler(new Uint8Array(data)), opts)),
  runString: options.runString ?? (options.run ? (handler, opts) => options.run(data => handler(decoder.decode(data)), opts) : (handler, opts) => options.runRaw(data => typeof data === "string" ? handler(data) : data instanceof Uint8Array ? handler(decoder.decode(data)) : handler(decoder.decode(new Uint8Array(data))), opts)),
  writer: options.writer
});
const encoder = /*#__PURE__*/new TextEncoder();
const decoder = /*#__PURE__*/new TextDecoder();
const CloseEventTypeId = "~effect/socket/Socket/CloseEvent";
/**
 * Represents a socket close event value carrying a close code and optional
 * reason.
 *
 * @category models
 * @since 4.0.0
 */
export class CloseEvent {
  /**
   * Marks this value as a socket close event for runtime guards.
   *
   * @since 4.0.0
   */
  [CloseEventTypeId];
  code;
  reason;
  constructor(code = 1000, reason) {
    this[CloseEventTypeId] = CloseEventTypeId;
    this.code = code;
    this.reason = reason;
  }
  /**
   * Formats the close code and optional reason for display.
   *
   * @since 4.0.0
   */
  toString() {
    return this.reason ? `${this.code}: ${this.reason}` : `${this.code}`;
  }
}
/**
 * Returns `true` when a value is a `CloseEvent`.
 *
 * @category refinements
 * @since 4.0.0
 */
export const isCloseEvent = u => Predicate.hasProperty(u, CloseEventTypeId);
/**
 * Runtime type identifier attached to `SocketError` values.
 *
 * @category type IDs
 * @since 4.0.0
 */
export const SocketErrorTypeId = "~effect/socket/Socket/SocketError";
/**
 * Returns `true` when a value is a `SocketError`.
 *
 * @category refinements
 * @since 4.0.0
 */
export const isSocketError = u => Predicate.hasProperty(u, SocketErrorTypeId);
/**
 * Typed error for failures that occur while reading from a socket.
 *
 * @category errors
 * @since 4.0.0
 */
export class SocketReadError extends /*#__PURE__*/Schema.ErrorClass("effect/socket/Socket/SocketReadError")({
  _tag: /*#__PURE__*/Schema.tag("SocketReadError"),
  cause: /*#__PURE__*/Schema.Defect()
}) {
  /**
   * Default message used for socket read failures.
   *
   * @since 4.0.0
   */
  message = `An error occurred during Read`;
}
/**
 * Typed error for failures that occur while writing to a socket.
 *
 * @category errors
 * @since 4.0.0
 */
export class SocketWriteError extends /*#__PURE__*/Schema.ErrorClass("effect/socket/Socket/SocketWriteError")({
  _tag: /*#__PURE__*/Schema.tag("SocketWriteError"),
  cause: /*#__PURE__*/Schema.Defect()
}) {
  /**
   * Default message used for socket write failures.
   *
   * @since 4.0.0
   */
  message = `An error occurred during Write`;
}
/**
 * Typed error for failures that occur while opening a socket, including
 * unknown open failures and open timeouts.
 *
 * @category errors
 * @since 4.0.0
 */
export class SocketOpenError extends /*#__PURE__*/Schema.ErrorClass("effect/socket/Socket/SocketOpenError")({
  _tag: /*#__PURE__*/Schema.tag("SocketOpenError"),
  kind: /*#__PURE__*/Schema.Literals(["Unknown", "Timeout"]),
  cause: /*#__PURE__*/Schema.Defect()
}) {
  /**
   * Formats timeout and unknown open failures for display.
   *
   * @since 4.0.0
   */
  get message() {
    return this.kind === "Timeout" ? `timeout waiting for "open"` : `An error occurred during Open`;
  }
}
/**
 * Typed error for a socket close event, carrying the close code and optional
 * close reason.
 *
 * @category errors
 * @since 4.0.0
 */
export class SocketCloseError extends /*#__PURE__*/Schema.ErrorClass("effect/socket/Socket/SocketCloseError")({
  _tag: /*#__PURE__*/Schema.tag("SocketCloseError"),
  code: Schema.Number,
  closeReason: /*#__PURE__*/Schema.optional(Schema.String)
}) {
  /**
   * Separates clean socket close errors from errors that should remain failures.
   *
   * @since 4.0.0
   */
  static filterClean(isClean) {
    return function (u) {
      return SocketError.is(u) && u.reason._tag === "SocketCloseError" && isClean(u.reason.code) ? Result.succeed(u.reason) : Result.fail(u);
    };
  }
  get message() {
    if (this.closeReason) {
      return `${this.code}: ${this.closeReason}`;
    }
    return `${this.code}`;
  }
}
/**
 * Schema for all socket-specific error reasons.
 *
 * @category errors
 * @since 4.0.0
 */
export const SocketErrorReason = /*#__PURE__*/Schema.Union([SocketReadError, SocketWriteError, SocketOpenError, SocketCloseError]);
/**
 * Tagged error that wraps socket read, write, open, and close failures while
 * preserving the underlying reason.
 *
 * @category errors
 * @since 4.0.0
 */
export class SocketError extends /*#__PURE__*/Schema.TaggedErrorClass(SocketErrorTypeId)("SocketError", {
  _tag: /*#__PURE__*/Schema.tag("SocketError"),
  reason: SocketErrorReason
}) {
  // @effect-diagnostics-next-line overriddenSchemaConstructor:off
  constructor(props) {
    if ("cause" in props.reason) {
      super({
        ...props,
        cause: props.reason.cause
      });
    } else {
      super(props);
    }
  }
  /**
   * Marks this value as a socket error wrapper for runtime guards.
   *
   * @since 4.0.0
   */
  [SocketErrorTypeId] = SocketErrorTypeId;
  /**
   * Returns `true` when the value is a `SocketError`.
   *
   * @since 4.0.0
   */
  static is(u) {
    return isSocketError(u);
  }
  message = this.reason.message;
}
/**
 * Converts a `Socket` into a bidirectional `Channel`, mapping incoming string
 * or binary frames and writing outgoing frame batches to the socket.
 *
 * @category combinators
 * @since 4.0.0
 */
export const toChannelMap = (self, f) => Channel.fromTransform(Effect.fnUntraced(function* (upstream, scope) {
  const queue = yield* Queue.make();
  const writeScope = yield* Scope.fork(scope);
  const write = yield* Scope.provide(self.writer, writeScope);
  let chunk;
  let index = 0;
  const writeChunk = Effect.whileLoop({
    while: () => index < chunk.length,
    body: () => write(chunk[index++]),
    step: constVoid
  });
  yield* upstream.pipe(Effect.flatMap(arr => {
    if (arr.length === 1) return write(arr[0]);
    chunk = arr;
    index = 0;
    return writeChunk;
  }), Effect.forever({
    disableYield: true
  }), Effect.catchCauseFilter(Pull.filterNoDone, cause => Queue.failCause(queue, cause)), Effect.ensuring(Scope.close(writeScope, Exit.void)), Effect.forkIn(scope));
  yield* self.runRaw(data => {
    Queue.offerUnsafe(queue, f(data));
  }).pipe(Queue.into(queue), Effect.forkIn(scope));
  // @effect-diagnostics-next-line returnEffectInGen:off
  return Queue.takeAll(queue);
}));
/**
 * Converts a `Socket` into a binary `Channel`, encoding incoming string frames
 * as UTF-8 bytes.
 *
 * @category combinators
 * @since 4.0.0
 */
export const toChannel = self => {
  const encoder = new TextEncoder();
  return toChannelMap(self, data => typeof data === "string" ? encoder.encode(data) : data);
};
/**
 * Converts a `Socket` into a string `Channel`, decoding binary frames with the
 * optional text encoding.
 *
 * @category combinators
 * @since 4.0.0
 */
export const toChannelString = /*#__PURE__*/dual(args => isSocket(args[0]), (self, encoding) => {
  const decoder = new TextDecoder(encoding);
  return toChannelMap(self, data => typeof data === "string" ? data : decoder.decode(data));
});
/**
 * Creates a `Socket` to binary `Channel` adapter with a fixed upstream error
 * type.
 *
 * @category combinators
 * @since 4.0.0
 */
export const toChannelWith = () => self => toChannel(self);
/**
 * Creates a binary socket `Channel` from the `Socket` service in the
 * environment.
 *
 * @category constructors
 * @since 4.0.0
 */
export const makeChannel = () => Channel.unwrap(Effect.map(Socket, toChannelWith()));
/**
 * Default close-code classifier that treats every socket close code as an
 * error.
 *
 * @category predicates
 * @since 4.0.0
 */
export const defaultCloseCodeIsError = _code => true;
/**
 * Context service for the active `WebSocket` instance available while a
 * WebSocket-backed socket run is handling events.
 *
 * @category services
 * @since 4.0.0
 */
export class WebSocket extends /*#__PURE__*/Context.Service()("~effect/socket/Socket/WebSocket") {}
/**
 * Context service for constructing `WebSocket` instances from a URL and
 * optional protocols.
 *
 * @category services
 * @since 4.0.0
 */
export class WebSocketConstructor extends /*#__PURE__*/Context.Service()("@effect/platform/Socket/WebSocketConstructor") {}
/**
 * Layer that provides `WebSocketConstructor` using `globalThis.WebSocket`.
 *
 * @category layers
 * @since 4.0.0
 */
export const layerWebSocketConstructorGlobal = /*#__PURE__*/Layer.succeed(WebSocketConstructor)((url, protocols) => new globalThis.WebSocket(url, protocols));
/**
 * Creates a `Socket` backed by a `WebSocketConstructor`, acquiring the
 * WebSocket for each run and using the close-code classifier to decide which
 * closes fail the run.
 *
 * @category constructors
 * @since 4.0.0
 */
export const makeWebSocket = (url, options) => WebSocketConstructor.use(makeWs => fromWebSocket(Effect.acquireRelease((typeof url === "string" ? Effect.succeed(url) : url).pipe(Effect.map(url => makeWs(url, options?.protocols))), ws => Effect.sync(() => ws.close(1000))), options));
/**
 * Builds a `Socket` from a scoped WebSocket acquisition effect, waiting for the
 * socket to open, dispatching message handlers in fibers, and translating
 * open, read, and close events into `SocketError` values.
 *
 * @category constructors
 * @since 4.0.0
 */
export const fromWebSocket = (acquire, options) => Effect.withFiber(fiber => {
  let currentWS;
  const latch = Latch.makeUnsafe(false);
  const acquireContext = fiber.context;
  const closeCodeIsError = options?.closeCodeIsError ?? defaultCloseCodeIsError;
  const runRaw = (handler, opts) => Effect.scopedWith(Effect.fnUntraced(function* (scope) {
    const fiberSet = yield* FiberSet.make().pipe(Scope.provide(scope));
    const ws = yield* Scope.provide(acquire, scope);
    const run = yield* Effect.provideService(FiberSet.runtime(fiberSet)(), WebSocket, ws);
    let open = false;
    function onMessage(event) {
      if (event.data instanceof Blob) {
        const effect = Effect.flatMap(Effect.promise(() => event.data.arrayBuffer()), buffer => {
          const result = handler(new Uint8Array(buffer));
          return Effect.isEffect(result) ? result : Effect.void;
        });
        return run(effect);
      }
      const result = handler(event.data);
      if (Effect.isEffect(result)) {
        run(result);
      }
    }
    function onError(cause) {
      ws.removeEventListener("message", onMessage);
      ws.removeEventListener("close", onClose);
      Deferred.doneUnsafe(fiberSet.deferred, Effect.fail(new SocketError({
        reason: open ? new SocketReadError({
          cause
        }) : new SocketOpenError({
          kind: "Unknown",
          cause
        })
      })));
    }
    function onClose(event) {
      const code = typeof event.code === "number" ? event.code : 1001;
      ws.removeEventListener("message", onMessage);
      ws.removeEventListener("error", onError);
      Deferred.doneUnsafe(fiberSet.deferred, Effect.fail(new SocketError({
        reason: new SocketCloseError({
          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.makeUnsafe();
      ws.addEventListener("open", () => {
        open = true;
        Deferred.doneUnsafe(openDeferred, Effect.void);
      }, {
        once: true
      });
      yield* Deferred.await(openDeferred).pipe(Effect.timeoutOrElse({
        duration: options?.openTimeout ?? 10000,
        orElse: () => Effect.fail(new SocketError({
          reason: new SocketOpenError({
            kind: "Timeout",
            cause: new Error("timeout waiting for \"open\"")
          })
        }))
      }), Effect.raceFirst(FiberSet.join(fiberSet)));
    }
    open = true;
    currentWS = ws;
    latch.openUnsafe();
    if (opts?.onOpen) yield* opts.onOpen;
    return yield* Effect.catchFilter(FiberSet.join(fiberSet), SocketCloseError.filterClean(_ => !closeCodeIsError(_)), () => Effect.void);
  })).pipe(Effect.updateContext(input => Context.merge(acquireContext, input)), Effect.ensuring(Effect.sync(() => {
    latch.closeUnsafe();
    currentWS = undefined;
  })));
  const write = chunk => latch.whenOpen(Effect.sync(() => {
    const ws = currentWS;
    if (isCloseEvent(chunk)) {
      ws.close(chunk.code, chunk.reason);
    } else {
      ws.send(chunk);
    }
  }));
  const writer = Effect.succeed(write);
  return Effect.succeed(make({
    runRaw,
    writer
  }));
});
/**
 * Creates a binary `Channel` backed by a WebSocket URL, requiring a
 * `WebSocketConstructor` service.
 *
 * @category constructors
 * @since 4.0.0
 */
export const makeWebSocketChannel = (url, options) => Channel.unwrap(Effect.map(makeWebSocket(url, options), toChannelWith()));
/**
 * Layer that provides a `Socket` service backed by a WebSocket URL or URL
 * effect.
 *
 * @category layers
 * @since 4.0.0
 */
export const layerWebSocket = /*#__PURE__*/flow(makeWebSocket, /*#__PURE__*/Layer.effect(Socket));
/**
 * Context reference for socket send queue capacity, defaulting to `16`.
 *
 * @category fiber refs
 * @since 4.0.0
 */
export const SendQueueCapacity = /*#__PURE__*/Context.Reference("~effect/socket/Socket/SendQueueCapacity", {
  defaultValue: () => 16
});
/**
 * Builds a `Socket` from a scoped `InputTransformStream`, reading incoming
 * chunks through socket handlers and writing outgoing chunks to the writable
 * stream, encoding strings as UTF-8 and using close-code classification for
 * `CloseEvent` values.
 *
 * @category constructors
 * @since 4.0.0
 */
export const fromTransformStream = (acquire, options) => Effect.withFiber(fiber => {
  const latch = Latch.makeUnsafe(false);
  let currentStream;
  const acquireServices = fiber.context;
  const closeCodeIsError = options?.closeCodeIsError ?? defaultCloseCodeIsError;
  const runRaw = (handler, opts) => Effect.scopedWith(Effect.fnUntraced(function* (scope) {
    const stream = yield* Scope.provide(acquire, scope);
    const reader = stream.readable.getReader();
    yield* Scope.addFinalizer(scope, Effect.promise(() => reader.cancel()));
    const fiberSet = yield* FiberSet.make().pipe(Scope.provide(scope));
    const runFork = yield* FiberSet.runtime(fiberSet)();
    yield* Effect.tryPromise({
      try: async () => {
        while (true) {
          const {
            done,
            value
          } = await reader.read();
          if (done) {
            throw new SocketError({
              reason: new SocketCloseError({
                code: 1000
              })
            });
          }
          const result = handler(value);
          if (Effect.isEffect(result)) {
            runFork(result);
          }
        }
      },
      catch: cause => isSocketError(cause) ? cause : new SocketError({
        reason: new SocketReadError({
          cause
        })
      })
    }).pipe(FiberSet.run(fiberSet));
    currentStream = {
      stream,
      fiberSet
    };
    yield* latch.open;
    if (opts?.onOpen) yield* opts.onOpen;
    return yield* Effect.catchFilter(FiberSet.join(fiberSet), SocketCloseError.filterClean(_ => !closeCodeIsError(_)), () => Effect.void);
  })).pipe(_ => _, Effect.updateContext(input => Context.merge(acquireServices, input)), Effect.ensuring(Effect.sync(() => {
    latch.closeUnsafe();
    currentStream = undefined;
  })));
  const writers = new WeakMap();
  const getWriter = stream => {
    let writer = writers.get(stream);
    if (!writer) {
      writer = stream.writable.getWriter();
      writers.set(stream, writer);
    }
    return writer;
  };
  const write = chunk => latch.whenOpen(Effect.suspend(() => {
    const {
      fiberSet,
      stream
    } = currentStream;
    if (isCloseEvent(chunk)) {
      return Deferred.fail(fiberSet.deferred, new SocketError({
        reason: new SocketCloseError({
          code: chunk.code,
          closeReason: chunk.reason
        })
      }));
    }
    return Effect.promise(() => getWriter(stream).write(typeof chunk === "string" ? encoder.encode(chunk) : chunk));
  }));
  const writer = Effect.acquireRelease(Effect.succeed(write), () => Effect.promise(async () => {
    if (!currentStream) return;
    await getWriter(currentStream.stream).close();
  }));
  return Effect.succeed(make({
    runRaw,
    writer
  }));
});
//# sourceMappingURL=Socket.js.map