effect-tsでRequirements ManagementでRuntime

ソースコードをうまくまとめようとしてRequirementsの機能を使ってみて以下ができるか試してみたメモ

• ログの切り替え
• 環境変数の切り替え
• モックの切り替え

イベント関数への登録あたりでEffect.runtimeを使用している箇所が、うまくいっておらず、Effect.runPromiseなどを使うようにするとログは出るが、AAA_Abort_Service_Test関連で実行時エラーが出る。そうでない場合はｒ

前回

参考

環境変数を使用する場合(.envファイル)

データを.envファイルに入れている

共通部分

test.spec.ts

import { describe, test } from 'vitest'
import {
  Effect,
  pipe,
  Either,
  Exit,
  Cause,
  Context,
  Data,
  Schema,
  Layer,
  ManagedRuntime,
  Config
} from 'effect'
import { UnknownException } from 'effect/Cause'

class SleepNGBeforeError extends Data.TaggedError('SleepNGBeforeError')<{}> {
  getMessage(site: string) {
    return `${site} - ng - Sleep NG before wait`
  }

  static isClass(test: any): test is SleepNGBeforeError {
    return test._tag === SleepNGBeforeError.name
  }
}

class SleepNGAfterError extends Data.TaggedError('SleepNGAfterError')<{ wait: number }> {
  getMessage(site: string) {
    return `${site} - ng - ${this.wait} - Sleep NG after wait`
  }

  static isClass(test: any): test is SleepNGAfterError {
    return test._tag === SleepNGAfterError.name
  }
}

class AlreadyAbortError extends Data.TaggedError('AlreadyAbortError')<{}> {
  static getMessage(site: string) {
    return `${site} - aborted - from event`
  }
}

class OriginAbortError extends Data.TaggedError('OriginAbortError')<{
  site: string
  reason: string
}> {
  getMessage() {
    return `${this.site} - abort - ${this.reason}`
  }

  static siteCheck(abortAction: Function, topIndex?: number) {
    return (site: string) => {
      return Effect.gen(function* () {
        const lastIndex = OkData.getLastIndex(site)
        let originAbortError = undefined
        if (Number.isNaN(lastIndex)) {
          const reason = `${lastIndex} is NaN`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (lastIndex === undefined) {
          const reason = `${lastIndex} is undefined`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (site[site.length - 2] !== '/') {
          const reason = `lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (topIndex && topIndex + lastIndex >= 10) {
          const reason = `topIndex + lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        }

        if (originAbortError !== undefined) {
          yield* Effect.log(originAbortError.getMessage())
          abortAction()
          return Effect.fail(originAbortError)
        }

        return Effect.succeed(site)
      })
    }
  }
}

class OkData extends Schema.Class<OkData>('OkData')({
  wait: Schema.Number
}) {
  static getLastIndex(site: string) {
    const index = site.slice(-1)
    return Number(index)
  }

  getMessage(site: string) {
    return `${site} - ok - ${this.wait}`
  }
}

const sleep = (wait: number) =>
  Effect.tryPromise<OkData>(
    () =>
      new Promise((resolve, reject) => {
        const waitSeconds = wait / 1000
        if (waitSeconds % 3 === 0) {
          reject(new SleepNGBeforeError())
        } else if (waitSeconds % 2 === 0) {
          setTimeout(() => reject(new SleepNGAfterError({ wait })), wait)
        } else {
          setTimeout(() => resolve(new OkData({ wait })), wait)
        }
      })
  )

const testOptions = {
  timeout: 1000 * 100
}

function getWaitNumber(index: number) {
  const wait = index * 1000
  return wait
}

function mapError_sleepNg<T>(effectAndThen: Effect.Effect<T, UnknownException, never>) {
  const effectAll = Effect.mapError(effectAndThen, (err) => {
    if (SleepNGBeforeError.isClass(err.error)) {
      return err.error
    } else if (SleepNGAfterError.isClass(err.error)) {
      return err.error
    } else {
      Effect.runPromise(Effect.log(`mapError_sleepNg : UnknownException : ${err}`))
      return err
    }
  })
  return effectAll
}

const do_sleep1 = (wait: number) =>
  Effect.try(() => {
    const sleepEffect = sleep(wait)
    const effectAll = mapError_sleepNg(sleepEffect)
    return effectAll
  })

test.spec.ts

class Abort_Service extends Context.Tag('Abort_Service')<
  Abort_Service,
  {
    readonly return_abort_error: (
      site: string
    ) => Effect.Effect<Either.Either<never, AlreadyAbortError>, never, never>
    readonly abort_event: (site: string) => Effect.Effect<void, never, never>
  }
>() {
  static readonly Live = Layer.succeed(Abort_Service, {
    return_abort_error: function (site: string) {
      return Effect.gen(function* () {
        const alreadyAbortError = new AlreadyAbortError()
        yield* Effect.log(AlreadyAbortError.getMessage(site))
        return Either.left(alreadyAbortError)
      })
    },
    abort_event: function (site: string) {
      return Effect.gen(function* () {
        yield* Effect.log(AlreadyAbortError.getMessage(site))
      })
    }
  })
}

class Main_Service extends Context.Tag('Main_Service')<
  Main_Service,
  {
    readonly abort_check: (
      abortAction: Function,
      topIndex?: number
    ) => (
      site: string
    ) => Effect.Effect<
      Effect.Effect<never, OriginAbortError, never> | Effect.Effect<string, never, never>,
      never,
      never
    >
    readonly do_effect: (
      abort_result: string
    ) => Effect.Effect<
      Effect.Effect<OkData, UnknownException | SleepNGBeforeError | SleepNGAfterError, never>,
      UnknownException,
      never
    >
  }
>() {
  static readonly Live = Layer.succeed(Main_Service, {
    abort_check: (abortAction, topIndex?) => (site: string) =>
      pipe(site, OriginAbortError.siteCheck(abortAction, topIndex)),
    do_effect: (abortResult) => pipe(abortResult, OkData.getLastIndex, getWaitNumber, do_sleep1)
  })
}

class Result_Service extends Context.Tag('Result_Service')<
  Result_Service,
  {
    readonly ok_action: (okData: OkData, site: string) => Effect.Effect<void, never, never>
    readonly ng_action: (
      ngData: UnknownException | SleepNGBeforeError | SleepNGAfterError,
      site: string
    ) => Effect.Effect<void, never, never>
  }
>() {
  static readonly Live = Layer.succeed(Result_Service, {
    ok_action: (okData, site) => {
      return Effect.gen(function* () {
        const message = okData.getMessage(site)
        yield* Effect.log(message)
      })
    },
    ng_action: (ngData, site) => {
      return Effect.gen(function* () {
        if (SleepNGBeforeError.isClass(ngData)) {
          const message = ngData.getMessage(site)
          yield* Effect.log(message)
        } else if (SleepNGAfterError.isClass(ngData)) {
          const message = ngData.getMessage(site)
          yield* Effect.log(message)
        }
      })
    }
  })
}

const MainLayer = Layer.mergeAll(Abort_Service.Live, Main_Service.Live, Result_Service.Live)

const MainRuntime = ManagedRuntime.make(MainLayer)

function mapper1(abortController: AbortController, topIndex?: number) {
  return (site: string) =>
    Effect.gen(function* () {
      const abortService = yield* Abort_Service
      const mainService = yield* Main_Service
      const resultService = yield* Result_Service

      const abortEvent = () => abortService.abort_event(site).pipe(Effect.runtime)
      if (abortController.signal.aborted) {
        return abortService.return_abort_error(site)
      }
      abortController.signal.addEventListener('abort', abortEvent)
      function removeEvent() {
        abortController.signal.removeEventListener('abort', abortEvent)
      }
      const abortAction = () => {
        removeEvent()
        abortController.abort()
      }

      const abortCheck = yield* yield* mainService.abort_check(abortAction, topIndex)(site)
      const effect = yield* mainService.do_effect(abortCheck)
      const either = yield* Effect.either(effect)

      if (Either.isRight(either)) {
        const okData = either.right
        yield* resultService.ok_action(okData, site)
      } else if (Either.isLeft(either)) {
        const ngData = either.left
        yield* resultService.ng_action(ngData, site)
      }

      removeEvent()
      return either
    })
}

const sitesData = Effect.gen(function* () {
  const config = yield* Config.array(Config.string(), 'VITE_MY_ARRAY')
  return config
})

describe('effect-ts 非同期処理１つの配列', () => {
  test('使い方１ー１: Requirements管理:本番or非同期テスト用', testOptions, async () => {
    const abortController = new AbortController()

    Effect.runPromise(Effect.log('[start]'))

    // const sites = [
    //   'https://main/1',
    //   'https://main/2',
    //   'https://main/3',
    //   'https://main/4',
    //   'https://main/5',
    //   'https://main/10'
    // ]
    const sites = await MainRuntime.runPromise(sitesData)

    MainRuntime.runPromise(Effect.log(`[sites] : ${sites}`))

    const effect = await Effect.all(sites.map(mapper1(abortController)), {
      concurrency: sites.length
    })
    const exit = await MainRuntime.runPromiseExit(effect, { signal: abortController.signal })

    MainRuntime.runPromise(
      Effect.log(
        Exit.match(exit, {
          onFailure: (cause) => `中断しました: ${Cause.pretty(cause)}`,
          onSuccess: (either) => either
        })
      )
    )
  })
})

中断データを含む場合

.env

VITE_MY_ARRAY=https://main/1, https://main/2, https://main/3, https://main/4, https://main/5, https://main/10

timestamp=2025-03-05T14:31:45.196Z level=INFO fiber=#1 message=[start]
timestamp=2025-03-05T14:31:45.221Z level=INFO fiber=#10 message="[sites] : https://main/1,https://main/2,https://main/3,https://main/4,https://main/5,https://main/10"
timestamp=2025-03-05T14:31:45.225Z level=INFO fiber=#18 message="https://main/10 - abort - lastIndex >= 10"
timestamp=2025-03-05T14:31:45.232Z level=INFO fiber=#19 message="中断しました: All fibers interrupted without errors."

中断データを含まない場合

.env

VITE_MY_ARRAY=https://main/1, https://main/2, https://main/3, https://main/4, https://main/5

timestamp=2025-03-05T14:32:30.655Z level=INFO fiber=#1 message=[start]
timestamp=2025-03-05T14:32:30.680Z level=INFO fiber=#10 message="[sites] : https://main/1,https://main/2,https://main/3,https://main/4,https://main/5"
timestamp=2025-03-05T14:32:30.686Z level=INFO fiber=#15 message="https://main/3 - ng - Sleep NG before wait"
timestamp=2025-03-05T14:32:31.688Z level=INFO fiber=#13 message="https://main/1 - ok - 1000"
timestamp=2025-03-05T14:32:32.684Z level=INFO fiber=#14 message="https://main/2 - ng - 2000 - Sleep NG after wait"
timestamp=2025-03-05T14:32:34.689Z level=INFO fiber=#16 message="https://main/4 - ng - 4000 - Sleep NG after wait"
timestamp=2025-03-05T14:32:35.691Z level=INFO fiber=#17 message="https://main/5 - ok - 5000"
timestamp=2025-03-05T14:32:35.694Z level=INFO fiber=#18 message="[
  {
    \"_id\": \"Either\",
    \"_tag\": \"Right\",
    \"right\": {
      \"wait\": 1000
    }
  },
  {
    \"_id\": \"Either\",
    \"_tag\": \"Left\",
    \"left\": {
      \"wait\": 2000,
      \"_tag\": \"SleepNGAfterError\"
    }
  },
  {
    \"_id\": \"Either\",
    \"_tag\": \"Left\",
    \"left\": {
      \"_tag\": \"SleepNGBeforeError\"
    }
  },
  {
    \"_id\": \"Either\",
    \"_tag\": \"Left\",
    \"left\": {
      \"wait\": 4000,
      \"_tag\": \"SleepNGAfterError\"
    }
  },
  {
    \"_id\": \"Either\",
    \"_tag\": \"Right\",
    \"right\": {
      \"wait\": 5000
    }
  }
]"

環境変数を使用せず、処理も切り替える場合

データはConfigProviderから取得している
Main_Service.LiveTest　の個所で処理を切り替えている
AAA_Abort_Service_TestでEffect.Serviceからサービス作成

共通部分

test.spec.ts

import { describe, test } from 'vitest'
import {
  Effect,
  Either,
  Exit,
  Cause,
  Context,
  Data,
  Schema,
  Layer,
  Duration,
  ManagedRuntime,
  Config,
  ConfigProvider
} from 'effect'
import { UnknownException } from 'effect/Cause'

class SleepNGBeforeError extends Data.TaggedError('SleepNGBeforeError')<{}> {
  getMessage(site: string) {
    return `${site} - ng - Sleep NG before wait`
  }

  static isClass(test: any): test is SleepNGBeforeError {
    return test._tag === SleepNGBeforeError.name
  }
}

class SleepNGAfterError extends Data.TaggedError('SleepNGAfterError')<{ wait: number }> {
  getMessage(site: string) {
    return `${site} - ng - ${this.wait} - Sleep NG after wait`
  }

  static isClass(test: any): test is SleepNGAfterError {
    return test._tag === SleepNGAfterError.name
  }
}

class AlreadyAbortError extends Data.TaggedError('AlreadyAbortError')<{}> {
  static getMessage(site: string) {
    return `${site} - aborted - from event`
  }
}

class OriginAbortError extends Data.TaggedError('OriginAbortError')<{
  site: string
  reason: string
}> {
  getMessage() {
    return `${this.site} - abort - ${this.reason}`
  }

  static siteCheck(abortAction: Function, topIndex?: number) {
    return (site: string) => {
      return Effect.gen(function* () {
        const lastIndex = OkData.getLastIndex(site)
        let originAbortError = undefined
        if (Number.isNaN(lastIndex)) {
          const reason = `${lastIndex} is NaN`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (lastIndex === undefined) {
          const reason = `${lastIndex} is undefined`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (site[site.length - 2] !== '/') {
          const reason = `lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (topIndex && topIndex + lastIndex >= 10) {
          const reason = `topIndex + lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        }

        if (originAbortError !== undefined) {
          yield* Effect.log(originAbortError.getMessage())
          abortAction()
          return Effect.fail(originAbortError)
        }

        return Effect.succeed(site)
      })
    }
  }
}

class OkData extends Schema.Class<OkData>('OkData')({
  wait: Schema.Number
}) {
  static getLastIndex(site: string) {
    const index = site.slice(-1)
    return Number(index)
  }

  getMessage(site: string) {
    return `${site} - ok - ${this.wait}`
  }
}

const testOptions = {
  timeout: 1000 * 100
}

test.spec.ts

class Abort_Service extends Context.Tag('Abort_Service')<
  Abort_Service,
  {
    readonly return_abort_error: (
      site: string
    ) => Effect.Effect<Either.Either<never, AlreadyAbortError>, never, never>
    readonly abort_event: (site: string) => Effect.Effect<void, never, never>
  }
>() {}

class AAA_Abort_Service_Test extends Effect.Service<Abort_Service>()('Abort_Service', {
  succeed: {
    return_abort_error: function (site: string) {
      return Effect.gen(function* () {
        const alreadyAbortError = new AlreadyAbortError()
        yield* Effect.log(`${site} - aborted - from event - test - return_abort_error`)
        return Either.left(alreadyAbortError)
      })
    },
    abort_event: function (site: string) {
      return () =>
        Effect.gen(function* () {
          yield* Effect.log(`${site} - aborted - from event - test - abort_event`)
        })
    }
  }
}) {}

class Main_Service extends Context.Tag('Main_Service')<
  Main_Service,
  {
    readonly abort_check: (
      abortAction: Function,
      topIndex?: number
    ) => (
      site: string
    ) => Effect.Effect<
      Effect.Effect<never, OriginAbortError, never> | Effect.Effect<string, never, never>,
      never,
      never
    >
    readonly do_effect: (
      abort_result: string
    ) => Effect.Effect<
      Effect.Effect<OkData, UnknownException | SleepNGBeforeError | SleepNGAfterError, never>,
      UnknownException,
      never
    >
  }
>() {
  static readonly LiveTest = Layer.succeed(Main_Service, {
    abort_check: (abortAction, topIndex?) => (site: string) => {
      return Effect.gen(function* () {
        const index = site.slice(-1)
        const lastIndex = Number(index)
        let originAbortError = undefined
        let reason = undefined
        if (Number.isNaN(lastIndex)) {
          reason = `${lastIndex} is NaN`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (lastIndex === undefined) {
          reason = `${lastIndex} is undefined`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (site[site.length - 2] !== '/') {
          reason = `lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (topIndex && topIndex + lastIndex >= 10) {
          reason = `topIndex + lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        }

        if (originAbortError !== undefined) {
          yield* Effect.log(`${site} - abort - ${reason}`)
          abortAction()
          return Effect.fail(originAbortError)
        }

        return Effect.succeed(site)
      })
    },
    do_effect: (abortResult) => {
      const index = abortResult.slice(-1)
      const lastIndex = Number(index)
      const wait = lastIndex * 1000
      const testWait = wait / 2

      if (lastIndex % 3 === 0) {
        return Effect.succeed(Effect.fail(new SleepNGBeforeError()))
      } else if (lastIndex % 2 === 0) {
        return Effect.andThen(
          Effect.sleep(Duration.millis(testWait)),
          Effect.succeed(Effect.fail(new SleepNGAfterError({ wait })))
        )
      } else {
        return Effect.andThen(
          Effect.sleep(Duration.millis(testWait)),
          Effect.succeed(Effect.succeed(new OkData({ wait: lastIndex * 1000 })))
        )
      }
    }
  })
}

class Result_Service extends Context.Tag('Result_Service')<
  Result_Service,
  {
    readonly ok_action: (okData: OkData, site: string) => Effect.Effect<void, never, never>
    readonly ng_action: (
      ngData: UnknownException | SleepNGBeforeError | SleepNGAfterError,
      site: string
    ) => Effect.Effect<void, never, never>
  }
>() {
  static readonly Live = Layer.succeed(Result_Service, {
    ok_action: (okData, site) => {
      return Effect.gen(function* () {
        const message = okData.getMessage(site)
        yield* Effect.log(message)
      })
    },
    ng_action: (ngData, site) => {
      return Effect.gen(function* () {
        if (SleepNGBeforeError.isClass(ngData)) {
          const message = ngData.getMessage(site)
          yield* Effect.log(message)
        } else if (SleepNGAfterError.isClass(ngData)) {
          const message = ngData.getMessage(site)
          yield* Effect.log(message)
        }
      })
    }
  })
}

const TestLayer = Layer.mergeAll(
  Layer.setConfigProvider(
    ConfigProvider.fromMap(
      new Map([
        [
          'VITE_MY_ARRAY',
          'https://test/1, https://test/2, https://test/3, https://test/4, https://test/5, https://test/10'
        ]
      ])
    )
  ),
  AAA_Abort_Service_Test.Default,
  Main_Service.LiveTest,
  Result_Service.Live
)

const TestRuntime = ManagedRuntime.make(TestLayer)

function mapper1(abortController: AbortController, topIndex?: number) {
  return (site: string) =>
    Effect.gen(function* () {
      const abortService = yield* Abort_Service
      const mainService = yield* Main_Service
      const resultService = yield* Result_Service

      const abortEvent = () => abortService.abort_event(site).pipe(Effect.runtime)
      if (abortController.signal.aborted) {
        return abortService.return_abort_error(site)
      }
      abortController.signal.addEventListener('abort', abortEvent)
      function removeEvent() {
        abortController.signal.removeEventListener('abort', abortEvent)
      }
      const abortAction = () => {
        removeEvent()
        abortController.abort()
      }

      const abortCheck = yield* yield* mainService.abort_check(abortAction, topIndex)(site)
      const effect = yield* mainService.do_effect(abortCheck)
      const either = yield* Effect.either(effect)

      if (Either.isRight(either)) {
        const okData = either.right
        yield* resultService.ok_action(okData, site)
      } else if (Either.isLeft(either)) {
        const ngData = either.left
        yield* resultService.ng_action(ngData, site)
      }

      removeEvent()
      return either
    })
}

const sitesData = Effect.gen(function* () {
  const config = yield* Config.array(Config.string(), 'VITE_MY_ARRAY')
  return config
})

describe('effect-ts 非同期処理１つの配列', () => {
  test('使い方１ー２: Requirements管理:高速テスト用', testOptions, async () => {
    const abortController = new AbortController()

    Effect.runPromise(Effect.log('[start]'))

    // const sites = [
    //   'https://test/1',
    //   'https://test/2',
    //   'https://test/3',
    //   'https://test/4',
    //   'https://test/5',
    //   'https://test/10'
    // ]
    const sites = await TestRuntime.runPromise(sitesData)

    TestRuntime.runPromise(Effect.log(`[sites] : ${sites}`))

    const effect = await Effect.all(sites.map(mapper1(abortController)), {
      concurrency: sites.length
    })
    const exit = await TestRuntime.runPromiseExit(effect, { signal: abortController.signal })

    TestRuntime.runPromise(
      Effect.log(
        Exit.match(exit, {
          onFailure: (cause) => `中断しました: ${Cause.pretty(cause)}`,
          onSuccess: (either) => either
        })
      )
    )
  })
})

中断データを含む場合

timestamp=2025-03-05T14:43:00.545Z level=INFO fiber=#1 message=[start]
timestamp=2025-03-05T14:43:00.573Z level=INFO fiber=#13 message="[sites] : https://test/1,https://test/2,https://test/3,https://test/4,https://test/5,https://test/10"
timestamp=2025-03-05T14:43:00.580Z level=INFO fiber=#18 message="https://test/3 - ng - Sleep NG before wait"
timestamp=2025-03-05T14:43:00.581Z level=INFO fiber=#21 message="https://test/10 - abort - lastIndex >= 10"
timestamp=2025-03-05T14:43:00.587Z level=INFO fiber=#22 message="中断しました: All fibers interrupted without errors."

中断データを含まない場合

timestamp=2025-03-05T14:43:23.267Z level=INFO fiber=#1 message=[start]
timestamp=2025-03-05T14:43:23.297Z level=INFO fiber=#13 message="[sites] : https://test/1,https://test/2,https://test/3,https://test/4,https://test/5"
timestamp=2025-03-05T14:43:23.303Z level=INFO fiber=#18 message="https://test/3 - ng - Sleep NG before wait"
timestamp=2025-03-05T14:43:23.801Z level=INFO fiber=#16 message="https://test/1 - ok - 1000"
timestamp=2025-03-05T14:43:24.303Z level=INFO fiber=#17 message="https://test/2 - ng - 2000 - Sleep NG after wait"
timestamp=2025-03-05T14:43:25.316Z level=INFO fiber=#19 message="https://test/4 - ng - 4000 - Sleep NG after wait"
timestamp=2025-03-05T14:43:25.811Z level=INFO fiber=#20 message="https://test/5 - ok - 5000"
timestamp=2025-03-05T14:43:25.814Z level=INFO fiber=#21 message="[
  {
    \"_id\": \"Either\",
    \"_tag\": \"Right\",
    \"right\": {
      \"wait\": 1000
    }
  },
  {
    \"_id\": \"Either\",
    \"_tag\": \"Left\",
    \"left\": {
      \"wait\": 2000,
      \"_tag\": \"SleepNGAfterError\"
    }
  },
  {
    \"_id\": \"Either\",
    \"_tag\": \"Left\",
    \"left\": {
      \"_tag\": \"SleepNGBeforeError\"
    }
  },
  {
    \"_id\": \"Either\",
    \"_tag\": \"Left\",
    \"left\": {
      \"wait\": 4000,
      \"_tag\": \"SleepNGAfterError\"
    }
  },
  {
    \"_id\": \"Either\",
    \"_tag\": \"Right\",
    \"right\": {
      \"wait\": 5000
    }
  }
]"

環境変数を使用せず、処理はそのままだが、ログ出力をやめる場合

データはConfigProviderから取得している
Logger.minimumLogLevel(LogLevel.None)でログの出力をやめている

共通部分

test.spec.ts

import { describe, test } from 'vitest'
import {
  Effect,
  pipe,
  Either,
  Exit,
  Cause,
  Context,
  Data,
  Schema,
  Layer,
  ManagedRuntime,
  Config,
  ConfigProvider,
  Logger,
  LogLevel
} from 'effect'
import { UnknownException } from 'effect/Cause'

class SleepNGBeforeError extends Data.TaggedError('SleepNGBeforeError')<{}> {
  getMessage(site: string) {
    return `${site} - ng - Sleep NG before wait`
  }

  static isClass(test: any): test is SleepNGBeforeError {
    return test._tag === SleepNGBeforeError.name
  }
}

class SleepNGAfterError extends Data.TaggedError('SleepNGAfterError')<{ wait: number }> {
  getMessage(site: string) {
    return `${site} - ng - ${this.wait} - Sleep NG after wait`
  }

  static isClass(test: any): test is SleepNGAfterError {
    return test._tag === SleepNGAfterError.name
  }
}

class AlreadyAbortError extends Data.TaggedError('AlreadyAbortError')<{}> {
  static getMessage(site: string) {
    return `${site} - aborted - from event`
  }
}

class OriginAbortError extends Data.TaggedError('OriginAbortError')<{
  site: string
  reason: string
}> {
  getMessage() {
    return `${this.site} - abort - ${this.reason}`
  }

  static siteCheck(abortAction: Function, topIndex?: number) {
    return (site: string) => {
      return Effect.gen(function* () {
        const lastIndex = OkData.getLastIndex(site)
        let originAbortError = undefined
        if (Number.isNaN(lastIndex)) {
          const reason = `${lastIndex} is NaN`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (lastIndex === undefined) {
          const reason = `${lastIndex} is undefined`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (site[site.length - 2] !== '/') {
          const reason = `lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (topIndex && topIndex + lastIndex >= 10) {
          const reason = `topIndex + lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        }

        if (originAbortError !== undefined) {
          yield* Effect.log(originAbortError.getMessage())
          abortAction()
          return Effect.fail(originAbortError)
        }

        return Effect.succeed(site)
      })
    }
  }
}

class OkData extends Schema.Class<OkData>('OkData')({
  wait: Schema.Number
}) {
  static getLastIndex(site: string) {
    const index = site.slice(-1)
    return Number(index)
  }

  getMessage(site: string) {
    return `${site} - ok - ${this.wait}`
  }
}

const sleep = (wait: number) =>
  Effect.tryPromise<OkData>(
    () =>
      new Promise((resolve, reject) => {
        const waitSeconds = wait / 1000
        if (waitSeconds % 3 === 0) {
          reject(new SleepNGBeforeError())
        } else if (waitSeconds % 2 === 0) {
          setTimeout(() => reject(new SleepNGAfterError({ wait })), wait)
        } else {
          setTimeout(() => resolve(new OkData({ wait })), wait)
        }
      })
  )

const testOptions = {
  timeout: 1000 * 100
}

function getWaitNumber(index: number) {
  const wait = index * 1000
  return wait
}

function mapError_sleepNg<T>(effectAndThen: Effect.Effect<T, UnknownException, never>) {
  const effectAll = Effect.mapError(effectAndThen, (err) => {
    if (SleepNGBeforeError.isClass(err.error)) {
      return err.error
    } else if (SleepNGAfterError.isClass(err.error)) {
      return err.error
    } else {
      Effect.runPromise(Effect.log(`mapError_sleepNg : UnknownException : ${err}`))
      return err
    }
  })
  return effectAll
}

const do_sleep1 = (wait: number) =>
  Effect.try(() => {
    const sleepEffect = sleep(wait)
    const effectAll = mapError_sleepNg(sleepEffect)
    return effectAll
  })

test.spec.ts

class Abort_Service extends Context.Tag('Abort_Service')<
  Abort_Service,
  {
    readonly return_abort_error: (
      site: string
    ) => Effect.Effect<Either.Either<never, AlreadyAbortError>, never, never>
    readonly abort_event: (site: string) => Effect.Effect<void, never, never>
  }
>() {
  static readonly Live = Layer.succeed(Abort_Service, {
    return_abort_error: function (site: string) {
      return Effect.gen(function* () {
        const alreadyAbortError = new AlreadyAbortError()
        yield* Effect.log(AlreadyAbortError.getMessage(site))
        return Either.left(alreadyAbortError)
      })
    },
    abort_event: function (site: string) {
      return Effect.gen(function* () {
        yield* Effect.log(AlreadyAbortError.getMessage(site))
      })
    }
  })
}

class Main_Service extends Context.Tag('Main_Service')<
  Main_Service,
  {
    readonly abort_check: (
      abortAction: Function,
      topIndex?: number
    ) => (
      site: string
    ) => Effect.Effect<
      Effect.Effect<never, OriginAbortError, never> | Effect.Effect<string, never, never>,
      never,
      never
    >
    readonly do_effect: (
      abort_result: string
    ) => Effect.Effect<
      Effect.Effect<OkData, UnknownException | SleepNGBeforeError | SleepNGAfterError, never>,
      UnknownException,
      never
    >
  }
>() {
  static readonly Live = Layer.succeed(Main_Service, {
    abort_check: (abortAction, topIndex?) => (site: string) =>
      pipe(site, OriginAbortError.siteCheck(abortAction, topIndex)),
    do_effect: (abortResult) => pipe(abortResult, OkData.getLastIndex, getWaitNumber, do_sleep1)
  })
}

class Result_Service extends Context.Tag('Result_Service')<
  Result_Service,
  {
    readonly ok_action: (okData: OkData, site: string) => Effect.Effect<void, never, never>
    readonly ng_action: (
      ngData: UnknownException | SleepNGBeforeError | SleepNGAfterError,
      site: string
    ) => Effect.Effect<void, never, never>
  }
>() {
  static readonly Live = Layer.succeed(Result_Service, {
    ok_action: (okData, site) => {
      return Effect.gen(function* () {
        const message = okData.getMessage(site)
        yield* Effect.log(message)
      })
    },
    ng_action: (ngData, site) => {
      return Effect.gen(function* () {
        if (SleepNGBeforeError.isClass(ngData)) {
          const message = ngData.getMessage(site)
          yield* Effect.log(message)
        } else if (SleepNGAfterError.isClass(ngData)) {
          const message = ngData.getMessage(site)
          yield* Effect.log(message)
        }
      })
    }
  })
}

const NoneLogLayer = Layer.mergeAll(
  Layer.setConfigProvider(
    ConfigProvider.fromMap(
      new Map([
        [
          'VITE_MY_ARRAY',
          'https://test/1, https://main/2, https://test/3, https://main/4, https://test/5, https://main/10'
        ]
      ])
    )
  ),
  Abort_Service.Live,
  Main_Service.Live,
  Result_Service.Live,
  Logger.minimumLogLevel(LogLevel.None)
)

const NoneLogRuntime = ManagedRuntime.make(NoneLogLayer)

function mapper1(abortController: AbortController, topIndex?: number) {
  return (site: string) =>
    Effect.gen(function* () {
      const abortService = yield* Abort_Service
      const mainService = yield* Main_Service
      const resultService = yield* Result_Service

      const abortEvent = () => abortService.abort_event(site).pipe(Effect.runtime)
      if (abortController.signal.aborted) {
        return abortService.return_abort_error(site)
      }
      abortController.signal.addEventListener('abort', abortEvent)
      function removeEvent() {
        abortController.signal.removeEventListener('abort', abortEvent)
      }
      const abortAction = () => {
        removeEvent()
        abortController.abort()
      }

      const abortCheck = yield* yield* mainService.abort_check(abortAction, topIndex)(site)
      const effect = yield* mainService.do_effect(abortCheck)
      const either = yield* Effect.either(effect)

      if (Either.isRight(either)) {
        const okData = either.right
        yield* resultService.ok_action(okData, site)
      } else if (Either.isLeft(either)) {
        const ngData = either.left
        yield* resultService.ng_action(ngData, site)
      }

      removeEvent()
      return either
    })
}

const sitesData = Effect.gen(function* () {
  const config = yield* Config.array(Config.string(), 'VITE_MY_ARRAY')
  return config
})

describe('effect-ts 非同期処理１つの配列', () => {
  test('使い方１ー３: Requirements管理:ログなしテスト用', testOptions, async () => {
    const abortController = new AbortController()

    Effect.runPromise(Effect.log('[start]'))

    // const sites = [
    //   'https://test/1',
    //   'https://main/2',
    //   'https://test/3',
    //   'https://main/4',
    //   'https://test/5',
    //   'https://main/10'
    // ]
    const sites = await NoneLogRuntime.runPromise(sitesData)

    NoneLogRuntime.runPromise(Effect.log(`[sites] : ${sites}`))

    const effect = await Effect.all(sites.map(mapper1(abortController)), {
      concurrency: sites.length
    })
    const exit = await NoneLogRuntime.runPromiseExit(effect, { signal: abortController.signal })

    NoneLogRuntime.runPromise(
      Effect.log(
        Exit.match(exit, {
          onFailure: (cause) => `中断しました: ${Cause.pretty(cause)}`,
          onSuccess: (either) => either
        })
      )
    )
  })
})

中断データを含む場合

timestamp=2025-03-05T14:50:08.490Z level=INFO fiber=#1 message=[start]

中断データを含まない場合

timestamp=2025-03-05T14:50:41.684Z level=INFO fiber=#1 message=[start]

全ソースコード

import { describe, test } from 'vitest'
import {
  Effect,
  pipe,
  Either,
  Exit,
  Cause,
  Context,
  Data,
  Schema,
  Layer,
  Duration,
  ManagedRuntime,
  Config,
  ConfigProvider,
  Logger,
  LogLevel
} from 'effect'
import { UnknownException } from 'effect/Cause'

class SleepNGBeforeError extends Data.TaggedError('SleepNGBeforeError')<{}> {
  getMessage(site: string) {
    return `${site} - ng - Sleep NG before wait`
  }

  static isClass(test: any): test is SleepNGBeforeError {
    return test._tag === SleepNGBeforeError.name
  }
}

class SleepNGAfterError extends Data.TaggedError('SleepNGAfterError')<{ wait: number }> {
  getMessage(site: string) {
    return `${site} - ng - ${this.wait} - Sleep NG after wait`
  }

  static isClass(test: any): test is SleepNGAfterError {
    return test._tag === SleepNGAfterError.name
  }
}

class AlreadyAbortError extends Data.TaggedError('AlreadyAbortError')<{}> {
  static getMessage(site: string) {
    return `${site} - aborted - from event`
  }
}

class OriginAbortError extends Data.TaggedError('OriginAbortError')<{
  site: string
  reason: string
}> {
  getMessage() {
    return `${this.site} - abort - ${this.reason}`
  }

  static siteCheck(abortAction: Function, topIndex?: number) {
    return (site: string) => {
      return Effect.gen(function* () {
        const lastIndex = OkData.getLastIndex(site)
        let originAbortError = undefined
        if (Number.isNaN(lastIndex)) {
          const reason = `${lastIndex} is NaN`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (lastIndex === undefined) {
          const reason = `${lastIndex} is undefined`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (site[site.length - 2] !== '/') {
          const reason = `lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (topIndex && topIndex + lastIndex >= 10) {
          const reason = `topIndex + lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        }

        if (originAbortError !== undefined) {
          yield* Effect.log(originAbortError.getMessage())
          abortAction()
          return Effect.fail(originAbortError)
        }

        return Effect.succeed(site)
      })
    }
  }
}

class OkData extends Schema.Class<OkData>('OkData')({
  wait: Schema.Number
}) {
  static getLastIndex(site: string) {
    const index = site.slice(-1)
    return Number(index)
  }

  getMessage(site: string) {
    return `${site} - ok - ${this.wait}`
  }
}

const sleep = (wait: number) =>
  Effect.tryPromise<OkData>(
    () =>
      new Promise((resolve, reject) => {
        const waitSeconds = wait / 1000
        if (waitSeconds % 3 === 0) {
          reject(new SleepNGBeforeError())
        } else if (waitSeconds % 2 === 0) {
          setTimeout(() => reject(new SleepNGAfterError({ wait })), wait)
        } else {
          setTimeout(() => resolve(new OkData({ wait })), wait)
        }
      })
  )

const testOptions = {
  timeout: 1000 * 100
}

function getWaitNumber(index: number) {
  const wait = index * 1000
  return wait
}

function mapError_sleepNg<T>(effectAndThen: Effect.Effect<T, UnknownException, never>) {
  const effectAll = Effect.mapError(effectAndThen, (err) => {
    if (SleepNGBeforeError.isClass(err.error)) {
      return err.error
    } else if (SleepNGAfterError.isClass(err.error)) {
      return err.error
    } else {
      Effect.runPromise(Effect.log(`mapError_sleepNg : UnknownException : ${err}`))
      return err
    }
  })
  return effectAll
}

const do_sleep1 = (wait: number) =>
  Effect.try(() => {
    const sleepEffect = sleep(wait)
    const effectAll = mapError_sleepNg(sleepEffect)
    return effectAll
  })

class Abort_Service extends Context.Tag('Abort_Service')<
  Abort_Service,
  {
    readonly return_abort_error: (
      site: string
    ) => Effect.Effect<Either.Either<never, AlreadyAbortError>, never, never>
    readonly abort_event: (site: string) => Effect.Effect<void, never, never>
  }
>() {
  static readonly Live = Layer.succeed(Abort_Service, {
    return_abort_error: function (site: string) {
      return Effect.gen(function* () {
        const alreadyAbortError = new AlreadyAbortError()
        yield* Effect.log(AlreadyAbortError.getMessage(site))
        return Either.left(alreadyAbortError)
      })
    },
    abort_event: function (site: string) {
      return Effect.gen(function* () {
        yield* Effect.log(AlreadyAbortError.getMessage(site))
      })
    }
  })

  static readonly LiveTest = Layer.succeed(Abort_Service, {
    return_abort_error: function (site: string) {
      return Effect.gen(function* () {
        const alreadyAbortError = new AlreadyAbortError()
        yield* Effect.log(`${site} - aborted - from event - test - return_abort_error`)
        return Either.left(alreadyAbortError)
      })
    },
    abort_event: function (site: string) {
      return Effect.gen(function* () {
        yield* Effect.log(`${site} - aborted - from event - test - abort_event`)
      })
    }
  })
}

class AAA_Abort_Service_Test extends Effect.Service<Abort_Service>()('Abort_Service', {
  succeed: {
    return_abort_error: function (site: string) {
      return Effect.gen(function* () {
        const alreadyAbortError = new AlreadyAbortError()
        yield* Effect.log(`${site} - aborted - from event - test - return_abort_error`)
        return Either.left(alreadyAbortError)
      })
    },
    abort_event: function (site: string) {
      return () =>
        Effect.gen(function* () {
          yield* Effect.log(`${site} - aborted - from event - test - abort_event`)
        })
    }
  }
}) {}

class Main_Service extends Context.Tag('Main_Service')<
  Main_Service,
  {
    readonly abort_check: (
      abortAction: Function,
      topIndex?: number
    ) => (
      site: string
    ) => Effect.Effect<
      Effect.Effect<never, OriginAbortError, never> | Effect.Effect<string, never, never>,
      never,
      never
    >
    readonly do_effect: (
      abort_result: string
    ) => Effect.Effect<
      Effect.Effect<OkData, UnknownException | SleepNGBeforeError | SleepNGAfterError, never>,
      UnknownException,
      never
    >
  }
>() {
  static readonly Live = Layer.succeed(Main_Service, {
    abort_check: (abortAction, topIndex?) => (site: string) =>
      pipe(site, OriginAbortError.siteCheck(abortAction, topIndex)),
    do_effect: (abortResult) => pipe(abortResult, OkData.getLastIndex, getWaitNumber, do_sleep1)
  })

  static readonly LiveTest = Layer.succeed(Main_Service, {
    abort_check: (abortAction, topIndex?) => (site: string) => {
      return Effect.gen(function* () {
        const index = site.slice(-1)
        const lastIndex = Number(index)
        let originAbortError = undefined
        let reason = undefined
        if (Number.isNaN(lastIndex)) {
          reason = `${lastIndex} is NaN`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (lastIndex === undefined) {
          reason = `${lastIndex} is undefined`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (site[site.length - 2] !== '/') {
          reason = `lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        } else if (topIndex && topIndex + lastIndex >= 10) {
          reason = `topIndex + lastIndex >= 10`
          originAbortError = new OriginAbortError({ site, reason })
        }

        if (originAbortError !== undefined) {
          yield* Effect.log(`${site} - abort - ${reason}`)
          abortAction()
          return Effect.fail(originAbortError)
        }

        return Effect.succeed(site)
      })
    },
    do_effect: (abortResult) => {
      const index = abortResult.slice(-1)
      const lastIndex = Number(index)
      const wait = lastIndex * 1000
      const testWait = wait / 2

      if (lastIndex % 3 === 0) {
        return Effect.succeed(Effect.fail(new SleepNGBeforeError()))
      } else if (lastIndex % 2 === 0) {
        return Effect.andThen(
          Effect.sleep(Duration.millis(testWait)),
          Effect.succeed(Effect.fail(new SleepNGAfterError({ wait })))
        )
      } else {
        return Effect.andThen(
          Effect.sleep(Duration.millis(testWait)),
          Effect.succeed(Effect.succeed(new OkData({ wait: lastIndex * 1000 })))
        )
      }
    }
  })
}

class Result_Service extends Context.Tag('Result_Service')<
  Result_Service,
  {
    readonly ok_action: (okData: OkData, site: string) => Effect.Effect<void, never, never>
    readonly ng_action: (
      ngData: UnknownException | SleepNGBeforeError | SleepNGAfterError,
      site: string
    ) => Effect.Effect<void, never, never>
  }
>() {
  static readonly Live = Layer.succeed(Result_Service, {
    ok_action: (okData, site) => {
      return Effect.gen(function* () {
        const message = okData.getMessage(site)
        yield* Effect.log(message)
      })
    },
    ng_action: (ngData, site) => {
      return Effect.gen(function* () {
        if (SleepNGBeforeError.isClass(ngData)) {
          const message = ngData.getMessage(site)
          yield* Effect.log(message)
        } else if (SleepNGAfterError.isClass(ngData)) {
          const message = ngData.getMessage(site)
          yield* Effect.log(message)
        }
      })
    }
  })
}

const MainLayer = Layer.mergeAll(Abort_Service.Live, Main_Service.Live, Result_Service.Live)

const TestLayer = Layer.mergeAll(
  Layer.setConfigProvider(
    ConfigProvider.fromMap(
      new Map([
        [
          'VITE_MY_ARRAY',
          'https://test/1, https://test/2, https://test/3, https://test/4, https://test/5, https://test/10'
        ]
      ])
    )
  ),
  AAA_Abort_Service_Test.Default,
  Main_Service.LiveTest,
  Result_Service.Live
)

const NoneLogLayer = Layer.mergeAll(
  Layer.setConfigProvider(
    ConfigProvider.fromMap(
      new Map([
        [
          'VITE_MY_ARRAY',
          'https://test/1, https://main/2, https://test/3, https://main/4, https://test/5, https://main/10'
        ]
      ])
    )
  ),
  Abort_Service.Live,
  Main_Service.Live,
  Result_Service.Live,
  Logger.minimumLogLevel(LogLevel.None)
)

const MainRuntime = ManagedRuntime.make(MainLayer)
const TestRuntime = ManagedRuntime.make(TestLayer)
const NoneLogRuntime = ManagedRuntime.make(NoneLogLayer)

function mapper1(abortController: AbortController, topIndex?: number) {
  return (site: string) =>
    Effect.gen(function* () {
      const abortService = yield* Abort_Service
      const mainService = yield* Main_Service
      const resultService = yield* Result_Service

      const abortEvent = () => abortService.abort_event(site).pipe(Effect.runtime)
      if (abortController.signal.aborted) {
        return abortService.return_abort_error(site)
      }
      abortController.signal.addEventListener('abort', abortEvent)
      function removeEvent() {
        abortController.signal.removeEventListener('abort', abortEvent)
      }
      const abortAction = () => {
        removeEvent()
        abortController.abort()
      }

      const abortCheck = yield* yield* mainService.abort_check(abortAction, topIndex)(site)
      const effect = yield* mainService.do_effect(abortCheck)
      const either = yield* Effect.either(effect)

      if (Either.isRight(either)) {
        const okData = either.right
        yield* resultService.ok_action(okData, site)
      } else if (Either.isLeft(either)) {
        const ngData = either.left
        yield* resultService.ng_action(ngData, site)
      }

      removeEvent()
      return either
    })
}

const sitesData = Effect.gen(function* () {
  const config = yield* Config.array(Config.string(), 'VITE_MY_ARRAY')
  return config
})

describe('effect-ts 非同期処理１つの配列', () => {
  test('使い方１ー１: Requirements管理:本番or非同期テスト用', testOptions, async () => {
    const abortController = new AbortController()

    Effect.runPromise(Effect.log('[start]'))

    // const sites = [
    //   'https://main/1',
    //   'https://main/2',
    //   'https://main/3',
    //   'https://main/4',
    //   'https://main/5',
    //   'https://main/10'
    // ]
    const sites = await MainRuntime.runPromise(sitesData)

    MainRuntime.runPromise(Effect.log(`[sites] : ${sites}`))

    const effect = await Effect.all(sites.map(mapper1(abortController)), {
      concurrency: sites.length
    })
    const exit = await MainRuntime.runPromiseExit(effect, { signal: abortController.signal })

    MainRuntime.runPromise(
      Effect.log(
        Exit.match(exit, {
          onFailure: (cause) => `中断しました: ${Cause.pretty(cause)}`,
          onSuccess: (either) => either
        })
      )
    )
  })

  test('使い方１ー２: Requirements管理:高速テスト用', testOptions, async () => {
    const abortController = new AbortController()

    Effect.runPromise(Effect.log('[start]'))

    // const sites = [
    //   'https://test/1',
    //   'https://test/2',
    //   'https://test/3',
    //   'https://test/4',
    //   'https://test/5',
    //   'https://test/10'
    // ]
    const sites = await TestRuntime.runPromise(sitesData)

    TestRuntime.runPromise(Effect.log(`[sites] : ${sites}`))

    const effect = await Effect.all(sites.map(mapper1(abortController)), {
      concurrency: sites.length
    })
    const exit = await TestRuntime.runPromiseExit(effect, { signal: abortController.signal })

    TestRuntime.runPromise(
      Effect.log(
        Exit.match(exit, {
          onFailure: (cause) => `中断しました: ${Cause.pretty(cause)}`,
          onSuccess: (either) => either
        })
      )
    )
  })
  test('使い方１ー３: Requirements管理:ログなしテスト用', testOptions, async () => {
    const abortController = new AbortController()

    Effect.runPromise(Effect.log('[start]'))

    // const sites = [
    //   'https://test/1',
    //   'https://main/2',
    //   'https://test/3',
    //   'https://main/4',
    //   'https://test/5',
    //   'https://main/10'
    // ]
    const sites = await NoneLogRuntime.runPromise(sitesData)

    NoneLogRuntime.runPromise(Effect.log(`[sites] : ${sites}`))

    const effect = await Effect.all(sites.map(mapper1(abortController)), {
      concurrency: sites.length
    })
    const exit = await NoneLogRuntime.runPromiseExit(effect, { signal: abortController.signal })

    NoneLogRuntime.runPromise(
      Effect.log(
        Exit.match(exit, {
          onFailure: (cause) => `中断しました: ${Cause.pretty(cause)}`,
          onSuccess: (either) => either
        })
      )
    )
  })
})