Function makeSerialized

packages/platform/src/internal/worker.ts:291–329

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

/** @internal */
export const makeSerialized = <
  I extends Schema.TaggedRequest.All
>(
  options: Worker.SerializedWorker.Options<I>
): Effect.Effect<Worker.SerializedWorker<I>, WorkerError, Worker.WorkerManager | Worker.Spawner | Scope.Scope> =>
  Effect.gen(function*() {
    const manager = yield* WorkerManager
    const backing = yield* manager.spawn({
      ...options as any,
      encode(message) {
        return Effect.mapError(
          Schema.serialize(message as any),
          (cause) => new WorkerError({ reason: "encode", cause })
        )
      }
    })
    const execute = <Req extends I>(message: Req) => {
      const parseSuccess = Schema.decode(Schema.successSchema(message as any))
      const parseFailure = Schema.decode(Schema.failureSchema(message as any))
      return pipe(
        backing.execute(message),
        Stream.catchAll((error) => Effect.flatMap(parseFailure(error), Effect.fail)),
        Stream.mapEffect(parseSuccess)
      )
    }
    const executeEffect = <Req extends I>(message: Req) => {
      const parseSuccess = Schema.decode(Schema.successSchema(message as any))
      const parseFailure = Schema.decode(Schema.failureSchema(message as any))
      return Effect.matchEffect(backing.executeEffect(message), {
        onFailure: (error) => Effect.flatMap(parseFailure(error), Effect.fail),
        onSuccess: parseSuccess
      })
    }
    return identity<Worker.SerializedWorker<I>>({
      id: backing.id,
      execute: execute as any,
      executeEffect: executeEffect as any
    })
  })

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