Function makePoolSerialized

packages/platform/src/internal/worker.ts:332–381

(
  options: Worker.SerializedWorkerPool.Options<I>
)

/** @internal */
export const makePoolSerialized = <I extends Schema.TaggedRequest.All>(
  options: Worker.SerializedWorkerPool.Options<I>
) =>
  Effect.gen(function*() {
    const manager = yield* WorkerManager
    const workers = new Set<Worker.SerializedWorker<I>>()
    const acquire = pipe(
      makeSerialized<I>(options),
      Effect.tap((worker) => Effect.sync(() => workers.add(worker))),
      Effect.tap((worker) => Effect.addFinalizer(() => Effect.sync(() => workers.delete(worker)))),
      options.onCreate
        ? Effect.tap(
          options.onCreate as (worker: Worker.SerializedWorker<I>) => Effect.Effect<void, WorkerError>
        )
        : identity,
      Effect.provideService(WorkerManager, manager)
    )
    const backing = yield* "timeToLive" in options ?
      Pool.makeWithTTL({
        acquire,
        min: options.minSize,
        max: options.maxSize,
        concurrency: options.concurrency,
        targetUtilization: options.targetUtilization,
        timeToLive: options.timeToLive
      }) :
      Pool.make({
        acquire,
        size: options.size,
        concurrency: options.concurrency,
        targetUtilization: options.targetUtilization
      })
    const pool: Worker.SerializedWorkerPool<I> = {
      backing,
      broadcast: <Req extends I>(message: Req) =>
        Effect.forEach(workers, (worker) => worker.executeEffect(message), {
          concurrency: "unbounded",
          discard: true
        }) as any,
      execute: <Req extends I>(message: Req) =>
        Stream.unwrapScoped(Effect.map(backing.get, (worker) => worker.execute(message))) as any,
      executeEffect: <Req extends I>(message: Req) =>
        Effect.scoped(Effect.flatMap(backing.get, (worker) => worker.executeEffect(message))) as any
    }

    // report any spawn errors
    yield* Effect.scoped(backing.get)

    return pool
  })

/** @internal */
export const makePoolSerializedLayer = <Tag, I extends Schema.TaggedRequest.All>(