| 1 | import { assert, describe, it } from "@effect/vitest" |
| 2 | import { Cause, Effect, Fiber, Option, Result, TxQueue } from "effect" |
| 3 | |
| 4 | describe("TxQueue", () => { |
| 5 | describe("interfaces", () => { |
| 6 | it.effect("TxEnqueue provides write-only interface", () => |
| 7 | Effect.tx(Effect.gen(function*() { |
| 8 | const queue = yield* TxQueue.bounded<number>(5) |
| 9 | |
| 10 | // Can assign to TxEnqueue interface |
| 11 | const enqueue: TxQueue.TxEnqueue<number> = queue |
| 12 | |
| 13 | // TxEnqueue type guard should work |
| 14 | assert.strictEqual(TxQueue.isTxEnqueue(enqueue), true) |
| 15 | |
| 16 | // Can use enqueue operations |
| 17 | const offered = yield* TxQueue.offer(queue, 42) |
| 18 | assert.strictEqual(offered, true) |
| 19 | |
| 20 | const rejected = yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 21 | assert.deepStrictEqual(rejected, []) |
| 22 | }))) |
| 23 | |
| 24 | it.effect("TxDequeue provides read-only interface", () => |
| 25 | Effect.tx(Effect.gen(function*() { |
| 26 | const queue = yield* TxQueue.bounded<number>(5) |
| 27 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 28 | |
| 29 | // Can assign to TxDequeue interface |
| 30 | const dequeue: TxQueue.TxDequeue<number> = queue |
| 31 | |
| 32 | // TxDequeue type guard should work |
| 33 | assert.strictEqual(TxQueue.isTxDequeue(dequeue), true) |
| 34 | |
| 35 | // Can use dequeue operations |
| 36 | const item = yield* TxQueue.take(queue) |
| 37 | assert.strictEqual(item, 1) |
| 38 | |
| 39 | const maybe = yield* TxQueue.poll(queue) |
| 40 | assert.deepStrictEqual(maybe, Option.some(2)) |
| 41 | }))) |
| 42 | |
| 43 | it.effect("TxQueue provides both enqueue and dequeue operations", () => |
| 44 | Effect.tx(Effect.gen(function*() { |
| 45 | const queue = yield* TxQueue.bounded<number>(5) |
| 46 | |
| 47 | // TxQueue should be both TxEnqueue and TxDequeue |
| 48 | assert.strictEqual(TxQueue.isTxQueue(queue), true) |
| 49 | assert.strictEqual(TxQueue.isTxEnqueue(queue), true) |
| 50 | assert.strictEqual(TxQueue.isTxDequeue(queue), true) |
| 51 | |
| 52 | // Can use both enqueue and dequeue operations |
| 53 | yield* TxQueue.offer(queue, 42) |
| 54 | const item = yield* TxQueue.take(queue) |
| 55 | assert.strictEqual(item, 42) |
| 56 | }))) |
| 57 | }) |
| 58 | |
| 59 | describe("interface segregation", () => { |
| 60 | it.effect("TxEnqueue interface enforces write-only operations", () => |
| 61 | Effect.tx(Effect.gen(function*() { |
| 62 | const queue = yield* TxQueue.bounded<number, string | Cause.Done>(5) |
| 63 | const enqueue: TxQueue.TxEnqueue<number, string | Cause.Done> = queue |
| 64 | |
| 65 | // Enqueue operations should work |
| 66 | const accepted = yield* TxQueue.offer(enqueue, 42) |
| 67 | assert.strictEqual(accepted, true) |
| 68 | |
| 69 | const rejected = yield* TxQueue.offerAll(enqueue, [1, 2, 3]) |
| 70 | assert.deepStrictEqual(rejected, []) |
| 71 | |
| 72 | // State management operations should work |
| 73 | const result = yield* TxQueue.failCause(enqueue, Cause.interrupt()) |
| 74 | assert.strictEqual(result, true) |
| 75 | |
| 76 | const endResult = yield* TxQueue.end(enqueue) |
| 77 | assert.strictEqual(endResult, false) // Already done |
| 78 | }))) |
| 79 | |
| 80 | it.effect("TxDequeue interface enforces read-only operations", () => |
| 81 | Effect.tx(Effect.gen(function*() { |
| 82 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 83 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 84 | const dequeue: TxQueue.TxDequeue<number, string> = queue |
| 85 | |
| 86 | // Dequeue operations should work |
| 87 | const item = yield* TxQueue.take(dequeue) |
| 88 | assert.strictEqual(item, 1) |
| 89 | |
| 90 | const maybe = yield* TxQueue.poll(dequeue) |
| 91 | assert.deepStrictEqual(maybe, Option.some(2)) |
| 92 | |
| 93 | const peek = yield* TxQueue.peek(dequeue) |
| 94 | assert.strictEqual(peek, 3) // Should be 3 since we took 1 and poll took 2 |
| 95 | |
| 96 | // State inspection operations should work |
| 97 | const size = yield* TxQueue.size(dequeue) |
| 98 | assert.strictEqual(size, 1) |
| 99 | |
| 100 | const isEmpty = yield* TxQueue.isEmpty(dequeue) |
| 101 | assert.strictEqual(isEmpty, false) |
| 102 | |
| 103 | const isOpen = yield* TxQueue.isOpen(dequeue) |
| 104 | assert.strictEqual(isOpen, true) |
| 105 | }))) |
| 106 | |
| 107 | it.effect("Interface assignments maintain type safety", () => |
| 108 | Effect.tx(Effect.gen(function*() { |
| 109 | const queue = yield* TxQueue.bounded<number>(5) |
| 110 | |
| 111 | // Full queue can be assigned to either interface |
| 112 | const enqueue: TxQueue.TxEnqueue<number> = queue |
| 113 | const dequeue: TxQueue.TxDequeue<number> = queue |
| 114 | const fullQueue: TxQueue.TxQueue<number> = queue |
| 115 | |
| 116 | // Verify all interfaces maintain functionality |
| 117 | yield* TxQueue.offer(enqueue, 1) |
| 118 | yield* TxQueue.offer(fullQueue, 2) |
| 119 | |
| 120 | const item1 = yield* TxQueue.take(dequeue) |
| 121 | const item2 = yield* TxQueue.take(fullQueue) |
| 122 | |
| 123 | assert.strictEqual(item1, 1) |
| 124 | assert.strictEqual(item2, 2) |
| 125 | |
| 126 | // State queries work on all interfaces |
| 127 | const size1 = yield* TxQueue.size(dequeue) |
| 128 | const size2 = yield* TxQueue.size(fullQueue) |
| 129 | assert.strictEqual(size1, size2) |
| 130 | }))) |
| 131 | |
| 132 | it.effect("Interface type guards work correctly", () => |
| 133 | Effect.tx(Effect.gen(function*() { |
| 134 | const queue = yield* TxQueue.bounded<number>(5) |
| 135 | const enqueue: TxQueue.TxEnqueue<number> = queue |
| 136 | const dequeue: TxQueue.TxDequeue<number> = queue |
| 137 | |
| 138 | // All type guards should return true for interfaces implemented by TxQueue |
| 139 | assert.strictEqual(TxQueue.isTxQueue(queue), true) |
| 140 | assert.strictEqual(TxQueue.isTxEnqueue(enqueue), true) |
| 141 | assert.strictEqual(TxQueue.isTxDequeue(dequeue), true) |
| 142 | |
| 143 | // Cross-interface type guards should work since TxQueue implements both |
| 144 | assert.strictEqual(TxQueue.isTxEnqueue(queue), true) |
| 145 | assert.strictEqual(TxQueue.isTxDequeue(queue), true) |
| 146 | assert.strictEqual(TxQueue.isTxQueue(enqueue), true) |
| 147 | assert.strictEqual(TxQueue.isTxQueue(dequeue), true) |
| 148 | }))) |
| 149 | }) |
| 150 | |
| 151 | describe("constructors", () => { |
| 152 | it.effect("bounded creates queue with specified capacity", () => |
| 153 | Effect.tx(Effect.gen(function*() { |
| 154 | const queue = yield* TxQueue.bounded<number>(5) |
| 155 | const size = yield* TxQueue.size(queue) |
| 156 | const isEmpty = yield* TxQueue.isEmpty(queue) |
| 157 | |
| 158 | assert.strictEqual(size, 0) |
| 159 | assert.strictEqual(isEmpty, true) |
| 160 | }))) |
| 161 | |
| 162 | it.effect("unbounded creates queue with unlimited capacity", () => |
| 163 | Effect.tx(Effect.gen(function*() { |
| 164 | const queue = yield* TxQueue.unbounded<string>() |
| 165 | const size = yield* TxQueue.size(queue) |
| 166 | const isEmpty = yield* TxQueue.isEmpty(queue) |
| 167 | |
| 168 | assert.strictEqual(size, 0) |
| 169 | assert.strictEqual(isEmpty, true) |
| 170 | }))) |
| 171 | |
| 172 | it.effect("dropping creates queue with dropping strategy", () => |
| 173 | Effect.tx(Effect.gen(function*() { |
| 174 | const queue = yield* TxQueue.dropping<number>(2) |
| 175 | const size = yield* TxQueue.size(queue) |
| 176 | const isEmpty = yield* TxQueue.isEmpty(queue) |
| 177 | |
| 178 | assert.strictEqual(size, 0) |
| 179 | assert.strictEqual(isEmpty, true) |
| 180 | }))) |
| 181 | |
| 182 | it.effect("sliding creates queue with sliding strategy", () => |
| 183 | Effect.tx(Effect.gen(function*() { |
| 184 | const queue = yield* TxQueue.sliding<number>(2) |
| 185 | const size = yield* TxQueue.size(queue) |
| 186 | const isEmpty = yield* TxQueue.isEmpty(queue) |
| 187 | |
| 188 | assert.strictEqual(size, 0) |
| 189 | assert.strictEqual(isEmpty, true) |
| 190 | }))) |
| 191 | }) |
| 192 | |
| 193 | describe("basic operations", () => { |
| 194 | it.effect("offer and take work correctly", () => |
| 195 | Effect.tx(Effect.gen(function*() { |
| 196 | const queue = yield* TxQueue.bounded<string>(10) |
| 197 | |
| 198 | const offered = yield* TxQueue.offer(queue, "hello") |
| 199 | assert.strictEqual(offered, true) |
| 200 | |
| 201 | const item = yield* TxQueue.take(queue) |
| 202 | assert.strictEqual(item, "hello") |
| 203 | }))) |
| 204 | |
| 205 | it.effect("poll returns Option.none for empty queue", () => |
| 206 | Effect.tx(Effect.gen(function*() { |
| 207 | const queue = yield* TxQueue.bounded<number>(10) |
| 208 | |
| 209 | const maybe = yield* TxQueue.poll(queue) |
| 210 | assert.deepStrictEqual(maybe, Option.none()) |
| 211 | }))) |
| 212 | |
| 213 | it.effect("poll returns Option.some for non-empty queue", () => |
| 214 | Effect.tx(Effect.gen(function*() { |
| 215 | const queue = yield* TxQueue.bounded<number>(10) |
| 216 | yield* TxQueue.offer(queue, 42) |
| 217 | |
| 218 | const maybe = yield* TxQueue.poll(queue) |
| 219 | assert.deepStrictEqual(maybe, Option.some(42)) |
| 220 | }))) |
| 221 | |
| 222 | it.effect("poll completes a closing queue after removing the last item", () => |
| 223 | Effect.tx(Effect.gen(function*() { |
| 224 | const queue = yield* TxQueue.bounded<number>(1) |
| 225 | yield* TxQueue.offer(queue, 42) |
| 226 | yield* TxQueue.interrupt(queue) |
| 227 | |
| 228 | assert.deepStrictEqual(yield* TxQueue.poll(queue), Option.some(42)) |
| 229 | assert.strictEqual(yield* TxQueue.isDone(queue), true) |
| 230 | }))) |
| 231 | |
| 232 | it.effect("clear completes a closing queue after removing all items", () => |
| 233 | Effect.tx(Effect.gen(function*() { |
| 234 | const queue = yield* TxQueue.bounded<number>(1) |
| 235 | yield* TxQueue.offer(queue, 42) |
| 236 | yield* TxQueue.interrupt(queue) |
| 237 | |
| 238 | assert.deepStrictEqual(yield* TxQueue.clear(queue), [42]) |
| 239 | assert.strictEqual(yield* TxQueue.isDone(queue), true) |
| 240 | }))) |
| 241 | |
| 242 | it.effect("offerAll works correctly", () => |
| 243 | Effect.tx(Effect.gen(function*() { |
| 244 | const queue = yield* TxQueue.bounded<number>(10) |
| 245 | |
| 246 | const rejected = yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5]) |
| 247 | assert.deepStrictEqual(rejected, []) |
| 248 | |
| 249 | const size = yield* TxQueue.size(queue) |
| 250 | assert.strictEqual(size, 5) |
| 251 | }))) |
| 252 | |
| 253 | it.effect("offerAll preserves a one-shot iterable across retries and runs", () => |
| 254 | Effect.gen(function*() { |
| 255 | const queue = yield* TxQueue.bounded<number>(1) |
| 256 | yield* TxQueue.offer(queue, 1) |
| 257 | |
| 258 | const values = (function*() { |
| 259 | yield 2 |
| 260 | })() |
| 261 | const offer = TxQueue.offerAll(queue, values) |
| 262 | const fiber = yield* Effect.forkChild(offer, { startImmediately: true }) |
| 263 | |
| 264 | assert.strictEqual(yield* TxQueue.take(queue), 1) |
| 265 | assert.deepStrictEqual(yield* Fiber.join(fiber), []) |
| 266 | assert.deepStrictEqual(yield* TxQueue.poll(queue), Option.some(2)) |
| 267 | |
| 268 | assert.deepStrictEqual(yield* offer, []) |
| 269 | assert.deepStrictEqual(yield* TxQueue.poll(queue), Option.some(2)) |
| 270 | })) |
| 271 | |
| 272 | it.effect("takeAll works correctly with new signature", () => |
| 273 | Effect.tx(Effect.gen(function*() { |
| 274 | const queue = yield* TxQueue.bounded<number>(10) |
| 275 | yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5]) |
| 276 | |
| 277 | const items = yield* TxQueue.takeAll(queue) |
| 278 | assert.deepStrictEqual(items, [1, 2, 3, 4, 5]) |
| 279 | |
| 280 | const size = yield* TxQueue.size(queue) |
| 281 | assert.strictEqual(size, 0) |
| 282 | }))) |
| 283 | |
| 284 | it.effect("takeAll with interrupted queue returns done flag", () => |
| 285 | Effect.tx(Effect.gen(function*() { |
| 286 | const queue = yield* TxQueue.bounded<number>(10) |
| 287 | yield* TxQueue.offerAll(queue, [1, 2]) |
| 288 | yield* TxQueue.interrupt(queue) // Interrupt the queue |
| 289 | |
| 290 | // takeAll should return all items |
| 291 | const items = yield* TxQueue.takeAll(queue) |
| 292 | assert.deepStrictEqual(items, [1, 2]) |
| 293 | }))) |
| 294 | |
| 295 | it.effect("takeAll with failed queue propagates error", () => |
| 296 | Effect.tx(Effect.gen(function*() { |
| 297 | const queue = yield* TxQueue.bounded<number, string>(10) |
| 298 | yield* TxQueue.offerAll(queue, [1, 2]) |
| 299 | yield* TxQueue.fail(queue, "queue failed") |
| 300 | |
| 301 | // takeAll should fail with the queue error |
| 302 | const result = yield* Effect.flip(TxQueue.takeAll(queue)) |
| 303 | assert.strictEqual(result, "queue failed") |
| 304 | }))) |
| 305 | |
| 306 | it.effect("takeN works correctly with new signature", () => |
| 307 | Effect.tx(Effect.gen(function*() { |
| 308 | const queue = yield* TxQueue.bounded<number>(10) |
| 309 | yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5]) |
| 310 | |
| 311 | const items = yield* TxQueue.takeN(queue, 3) |
| 312 | assert.deepStrictEqual(items, [1, 2, 3]) |
| 313 | |
| 314 | const size = yield* TxQueue.size(queue) |
| 315 | assert.strictEqual(size, 2) |
| 316 | }))) |
| 317 | |
| 318 | it.effect("takeN and takeBetween normalize element counts before waiting", () => |
| 319 | Effect.tx(Effect.gen(function*() { |
| 320 | const takeNQueue = yield* TxQueue.unbounded<number>() |
| 321 | yield* TxQueue.offer(takeNQueue, 1) |
| 322 | assert.deepStrictEqual(yield* TxQueue.takeN(takeNQueue, 1.9), [1]) |
| 323 | |
| 324 | const takeBetweenQueue = yield* TxQueue.unbounded<number>() |
| 325 | yield* TxQueue.offer(takeBetweenQueue, 1) |
| 326 | assert.deepStrictEqual(yield* TxQueue.takeBetween(takeBetweenQueue, 1.9, 2.9), [1]) |
| 327 | |
| 328 | const invalidQueue = yield* TxQueue.unbounded<number>() |
| 329 | yield* TxQueue.offer(invalidQueue, 1) |
| 330 | assert.deepStrictEqual(yield* TxQueue.takeN(invalidQueue, Number.NaN), []) |
| 331 | assert.deepStrictEqual(yield* TxQueue.takeAll(invalidQueue), [1]) |
| 332 | }))) |
| 333 | |
| 334 | it.effect("takeN with queue completion gets available items", () => |
| 335 | Effect.tx(Effect.gen(function*() { |
| 336 | const queue = yield* TxQueue.bounded<number>(10) |
| 337 | yield* TxQueue.offerAll(queue, [1, 2]) |
| 338 | yield* TxQueue.interrupt(queue) // Interrupt the queue |
| 339 | |
| 340 | // Take more than available - should get all items |
| 341 | const items = yield* TxQueue.takeN(queue, 5) |
| 342 | assert.deepStrictEqual(items, [1, 2]) |
| 343 | }))) |
| 344 | |
| 345 | it.effect("takeN with failed queue propagates error", () => |
| 346 | Effect.tx(Effect.gen(function*() { |
| 347 | const queue = yield* TxQueue.bounded<number, string>(10) |
| 348 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 349 | yield* TxQueue.fail(queue, "queue failed") |
| 350 | |
| 351 | // takeN should fail with the queue error |
| 352 | const result = yield* Effect.flip(TxQueue.takeN(queue, 2)) |
| 353 | assert.strictEqual(result, "queue failed") |
| 354 | }))) |
| 355 | |
| 356 | it.effect("takeN with capacity constraint takes available items", () => |
| 357 | Effect.tx(Effect.gen(function*() { |
| 358 | const queue = yield* TxQueue.bounded<number>(3) // Capacity is 3 |
| 359 | yield* TxQueue.offerAll(queue, [1, 2, 3]) // Fill to capacity |
| 360 | |
| 361 | // Request more than capacity - should take all available immediately |
| 362 | const items = yield* TxQueue.takeN(queue, 10) |
| 363 | assert.deepStrictEqual(items, [1, 2, 3]) |
| 364 | |
| 365 | const size = yield* TxQueue.size(queue) |
| 366 | assert.strictEqual(size, 0) |
| 367 | }))) |
| 368 | |
| 369 | it.effect("takeAll() on empty closed queue gets interrupted", () => |
| 370 | Effect.tx(Effect.gen(function*() { |
| 371 | const queue = yield* TxQueue.bounded<number>(5) |
| 372 | yield* TxQueue.interrupt(queue) // Empty queue becomes Done immediately |
| 373 | |
| 374 | // takeAll should get interrupted since queue is Done |
| 375 | const result = yield* Effect.exit(TxQueue.takeAll(queue)) |
| 376 | assert.strictEqual(result._tag, "Failure") |
| 377 | if (result._tag === "Failure") { |
| 378 | assert.strictEqual(Cause.hasInterruptsOnly(result.cause), true) |
| 379 | } |
| 380 | }))) |
| 381 | |
| 382 | it.effect("takeAll() waits when empty then returns items when available", () => |
| 383 | Effect.tx(Effect.gen(function*() { |
| 384 | const queue = yield* TxQueue.bounded<number>(5) |
| 385 | |
| 386 | // Start takeAll in a fiber - it should wait because queue is empty |
| 387 | const fiber = yield* Effect.forkChild(TxQueue.takeAll(queue)) |
| 388 | |
| 389 | // Add items to the queue |
| 390 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 391 | |
| 392 | // Now takeAll should complete and return the items |
| 393 | const items = yield* Fiber.join(fiber) |
| 394 | assert.deepStrictEqual(items, [1, 2, 3]) |
| 395 | }))) |
| 396 | |
| 397 | it.effect("takeAll() during state transition handles race condition", () => |
| 398 | Effect.tx(Effect.gen(function*() { |
| 399 | const queue = yield* TxQueue.bounded<number>(5) |
| 400 | yield* TxQueue.offerAll(queue, [1, 2]) |
| 401 | |
| 402 | // Interrupt queue to put it in Closing state |
| 403 | yield* TxQueue.interrupt(queue) |
| 404 | |
| 405 | // takeAll should get items and transition queue to Done |
| 406 | const items = yield* TxQueue.takeAll(queue) |
| 407 | assert.deepStrictEqual(items, [1, 2]) |
| 408 | |
| 409 | // Verify queue is now Done |
| 410 | const isDone = yield* TxQueue.isDone(queue) |
| 411 | assert.strictEqual(isDone, true) |
| 412 | }))) |
| 413 | |
| 414 | it.effect("takeN(0) returns empty array", () => |
| 415 | Effect.tx(Effect.gen(function*() { |
| 416 | const queue = yield* TxQueue.bounded<number>(5) |
| 417 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 418 | |
| 419 | const items = yield* TxQueue.takeN(queue, 0) |
| 420 | assert.deepStrictEqual(items, []) |
| 421 | |
| 422 | // Queue should still have all items |
| 423 | const size = yield* TxQueue.size(queue) |
| 424 | assert.strictEqual(size, 3) |
| 425 | }))) |
| 426 | |
| 427 | it.effect("takeN() with closing queue returns available items", () => |
| 428 | Effect.tx(Effect.gen(function*() { |
| 429 | const queue = yield* TxQueue.bounded<number>(5) |
| 430 | yield* TxQueue.offerAll(queue, [1, 2]) |
| 431 | yield* TxQueue.interrupt(queue) // Put in Closing state |
| 432 | |
| 433 | // Request more than available - should get what's available and transition to Done |
| 434 | const items = yield* TxQueue.takeN(queue, 5) |
| 435 | assert.deepStrictEqual(items, [1, 2]) |
| 436 | |
| 437 | // Queue should now be Done |
| 438 | const isDone = yield* TxQueue.isDone(queue) |
| 439 | assert.strictEqual(isDone, true) |
| 440 | }))) |
| 441 | |
| 442 | it.effect("takeN() signature validation - returns Array<A> not [Array<A>, boolean]", () => |
| 443 | Effect.tx(Effect.gen(function*() { |
| 444 | const queue = yield* TxQueue.bounded<number>(5) |
| 445 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 446 | |
| 447 | const items = yield* TxQueue.takeN(queue, 2) |
| 448 | |
| 449 | // Verify return type is Array<A>, not [Array<A>, boolean] |
| 450 | assert.strictEqual(Array.isArray(items), true) |
| 451 | assert.deepStrictEqual(items, [1, 2]) |
| 452 | |
| 453 | // Verify it's not a tuple - should not have boolean as second element |
| 454 | assert.strictEqual(typeof items[2], "undefined") |
| 455 | }))) |
| 456 | |
| 457 | it.effect("takeBetween() basic functionality", () => |
| 458 | Effect.tx(Effect.gen(function*() { |
| 459 | const queue = yield* TxQueue.bounded<number>(10) |
| 460 | yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7, 8]) |
| 461 | |
| 462 | // Take between 2 and 5 items |
| 463 | const batch1 = yield* TxQueue.takeBetween(queue, 2, 5) |
| 464 | assert.strictEqual(batch1.length, 5) |
| 465 | assert.deepStrictEqual(batch1, [1, 2, 3, 4, 5]) |
| 466 | |
| 467 | // Take between 1 and 10 items (but only 3 remain) |
| 468 | const batch2 = yield* TxQueue.takeBetween(queue, 1, 10) |
| 469 | assert.strictEqual(batch2.length, 3) |
| 470 | assert.deepStrictEqual(batch2, [6, 7, 8]) |
| 471 | |
| 472 | // Verify queue is empty |
| 473 | const empty = yield* TxQueue.isEmpty(queue) |
| 474 | assert.strictEqual(empty, true) |
| 475 | }))) |
| 476 | |
| 477 | it.effect("takeBetween() takes up to maximum when available", () => |
| 478 | Effect.tx(Effect.gen(function*() { |
| 479 | const queue = yield* TxQueue.bounded<number>(10) |
| 480 | yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7]) |
| 481 | |
| 482 | // Take between 3 and 5 items - should take exactly 5 (the maximum) |
| 483 | const items = yield* TxQueue.takeBetween(queue, 3, 5) |
| 484 | assert.strictEqual(items.length, 5) |
| 485 | assert.deepStrictEqual(items, [1, 2, 3, 4, 5]) |
| 486 | |
| 487 | // Remaining items should still be in queue |
| 488 | const remaining = yield* TxQueue.takeAll(queue) |
| 489 | assert.deepStrictEqual(remaining, [6, 7]) |
| 490 | }))) |
| 491 | |
| 492 | it.effect("takeBetween() with invalid parameters", () => |
| 493 | Effect.tx(Effect.gen(function*() { |
| 494 | const queue = yield* TxQueue.bounded<number>(10) |
| 495 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 496 | |
| 497 | // Invalid parameters should return empty array |
| 498 | const result1 = yield* TxQueue.takeBetween(queue, 0, 5) |
| 499 | assert.deepStrictEqual(result1, []) |
| 500 | |
| 501 | const result2 = yield* TxQueue.takeBetween(queue, -1, 5) |
| 502 | assert.deepStrictEqual(result2, []) |
| 503 | |
| 504 | const result3 = yield* TxQueue.takeBetween(queue, 5, 2) // min > max |
| 505 | assert.deepStrictEqual(result3, []) |
| 506 | |
| 507 | // Queue should still have all items |
| 508 | const size = yield* TxQueue.size(queue) |
| 509 | assert.strictEqual(size, 3) |
| 510 | }))) |
| 511 | |
| 512 | it.effect("takeBetween() with failed queue propagates error", () => |
| 513 | Effect.tx(Effect.gen(function*() { |
| 514 | const queue = yield* TxQueue.bounded<number, string>(10) |
| 515 | yield* TxQueue.fail(queue, "test error") |
| 516 | |
| 517 | const result = yield* Effect.exit(TxQueue.takeBetween(queue, 1, 5)) |
| 518 | assert.strictEqual(result._tag, "Failure") |
| 519 | if (result._tag === "Failure") { |
| 520 | const error = Cause.findError(result.cause) |
| 521 | assert.ok(Result.isSuccess(error)) |
| 522 | assert.strictEqual(error.success, "test error") |
| 523 | } |
| 524 | }))) |
| 525 | |
| 526 | it.effect("takeBetween() with closing queue returns available items", () => |
| 527 | Effect.tx(Effect.gen(function*() { |
| 528 | const queue = yield* TxQueue.bounded<number>(10) |
| 529 | yield* TxQueue.offerAll(queue, [1, 2]) |
| 530 | |
| 531 | // Start closing process |
| 532 | yield* TxQueue.interrupt(queue) |
| 533 | |
| 534 | // Should return available items even if less than minimum |
| 535 | const items = yield* TxQueue.takeBetween(queue, 3, 5) |
| 536 | assert.deepStrictEqual(items, [1, 2]) |
| 537 | |
| 538 | // Queue should be done now |
| 539 | const done = yield* TxQueue.isDone(queue) |
| 540 | assert.strictEqual(done, true) |
| 541 | }))) |
| 542 | |
| 543 | it.effect("takeBetween() exact minimum and maximum", () => |
| 544 | Effect.tx(Effect.gen(function*() { |
| 545 | const queue = yield* TxQueue.bounded<number>(10) |
| 546 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 547 | |
| 548 | // Take exactly 3 items (min=max=3) |
| 549 | const items = yield* TxQueue.takeBetween(queue, 3, 3) |
| 550 | assert.strictEqual(items.length, 3) |
| 551 | assert.deepStrictEqual(items, [1, 2, 3]) |
| 552 | |
| 553 | const empty = yield* TxQueue.isEmpty(queue) |
| 554 | assert.strictEqual(empty, true) |
| 555 | }))) |
| 556 | |
| 557 | it.effect("takeBetween() returns Array<A> not tuple", () => |
| 558 | Effect.tx(Effect.gen(function*() { |
| 559 | const queue = yield* TxQueue.bounded<number>(5) |
| 560 | yield* TxQueue.offerAll(queue, [1, 2, 3, 4]) |
| 561 | |
| 562 | const items = yield* TxQueue.takeBetween(queue, 2, 3) |
| 563 | |
| 564 | // Verify return type is Array<A>, not [Array<A>, boolean] |
| 565 | assert.strictEqual(Array.isArray(items), true) |
| 566 | assert.deepStrictEqual(items, [1, 2, 3]) |
| 567 | |
| 568 | // Verify it's not a tuple - should not have boolean as extra element |
| 569 | assert.strictEqual(typeof items[3], "undefined") |
| 570 | }))) |
| 571 | |
| 572 | it.effect("peek works correctly", () => |
| 573 | Effect.tx(Effect.gen(function*() { |
| 574 | const queue = yield* TxQueue.bounded<number>(10) |
| 575 | yield* TxQueue.offer(queue, 42) |
| 576 | |
| 577 | const item = yield* TxQueue.peek(queue) |
| 578 | assert.strictEqual(item, 42) |
| 579 | |
| 580 | // Item should still be in queue |
| 581 | const size = yield* TxQueue.size(queue) |
| 582 | assert.strictEqual(size, 1) |
| 583 | }))) |
| 584 | }) |
| 585 | |
| 586 | describe("queue state", () => { |
| 587 | it.effect("size returns correct count", () => |
| 588 | Effect.tx(Effect.gen(function*() { |
| 589 | const queue = yield* TxQueue.bounded<number>(10) |
| 590 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 591 | |
| 592 | const size = yield* TxQueue.size(queue) |
| 593 | assert.strictEqual(size, 3) |
| 594 | }))) |
| 595 | |
| 596 | it.effect("isEmpty and isNonEmpty reflect queue contents", () => |
| 597 | Effect.tx(Effect.gen(function*() { |
| 598 | const queue = yield* TxQueue.bounded<number>(10) |
| 599 | |
| 600 | assert.strictEqual(yield* TxQueue.isEmpty(queue), true) |
| 601 | assert.strictEqual(yield* TxQueue.isNonEmpty(queue), false) |
| 602 | |
| 603 | yield* TxQueue.offer(queue, 1) |
| 604 | assert.strictEqual(yield* TxQueue.isEmpty(queue), false) |
| 605 | assert.strictEqual(yield* TxQueue.isNonEmpty(queue), true) |
| 606 | }))) |
| 607 | |
| 608 | it.effect("isFull works correctly for bounded queue", () => |
| 609 | Effect.tx(Effect.gen(function*() { |
| 610 | const queue = yield* TxQueue.bounded<number>(2) |
| 611 | |
| 612 | const full1 = yield* TxQueue.isFull(queue) |
| 613 | assert.strictEqual(full1, false) |
| 614 | |
| 615 | yield* TxQueue.offerAll(queue, [1, 2]) |
| 616 | const full2 = yield* TxQueue.isFull(queue) |
| 617 | assert.strictEqual(full2, true) |
| 618 | }))) |
| 619 | |
| 620 | it.effect("isFull returns false for unbounded queue", () => |
| 621 | Effect.tx(Effect.gen(function*() { |
| 622 | const queue = yield* TxQueue.unbounded<number>() |
| 623 | yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5]) |
| 624 | |
| 625 | const full = yield* TxQueue.isFull(queue) |
| 626 | assert.strictEqual(full, false) |
| 627 | }))) |
| 628 | }) |
| 629 | |
| 630 | describe("clear", () => { |
| 631 | it.effect("clear removes all items from queue and returns them", () => |
| 632 | Effect.tx(Effect.gen(function*() { |
| 633 | const queue = yield* TxQueue.bounded<number>(10) |
| 634 | yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5]) |
| 635 | |
| 636 | const sizeBefore = yield* TxQueue.size(queue) |
| 637 | assert.strictEqual(sizeBefore, 5) |
| 638 | |
| 639 | const cleared = yield* TxQueue.clear(queue) |
| 640 | assert.deepStrictEqual(cleared, [1, 2, 3, 4, 5]) |
| 641 | |
| 642 | const sizeAfter = yield* TxQueue.size(queue) |
| 643 | assert.strictEqual(sizeAfter, 0) |
| 644 | |
| 645 | const isEmpty = yield* TxQueue.isEmpty(queue) |
| 646 | assert.strictEqual(isEmpty, true) |
| 647 | }))) |
| 648 | |
| 649 | it.effect("clear does not affect queue state", () => |
| 650 | Effect.tx(Effect.gen(function*() { |
| 651 | const queue = yield* TxQueue.bounded<number>(10) |
| 652 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 653 | |
| 654 | const cleared = yield* TxQueue.clear(queue) |
| 655 | assert.deepStrictEqual(cleared, [1, 2, 3]) |
| 656 | |
| 657 | // Queue should still be open |
| 658 | const isOpen = yield* TxQueue.isOpen(queue) |
| 659 | assert.strictEqual(isOpen, true) |
| 660 | |
| 661 | // Should be able to offer new items |
| 662 | const offered = yield* TxQueue.offer(queue, 42) |
| 663 | assert.strictEqual(offered, true) |
| 664 | |
| 665 | const size = yield* TxQueue.size(queue) |
| 666 | assert.strictEqual(size, 1) |
| 667 | }))) |
| 668 | |
| 669 | it.effect("clear works on empty queue", () => |
| 670 | Effect.tx(Effect.gen(function*() { |
| 671 | const queue = yield* TxQueue.bounded<number>(10) |
| 672 | |
| 673 | const cleared = yield* TxQueue.clear(queue) |
| 674 | assert.deepStrictEqual(cleared, []) |
| 675 | |
| 676 | const size = yield* TxQueue.size(queue) |
| 677 | assert.strictEqual(size, 0) |
| 678 | |
| 679 | const isEmpty = yield* TxQueue.isEmpty(queue) |
| 680 | assert.strictEqual(isEmpty, true) |
| 681 | }))) |
| 682 | |
| 683 | it.effect("clear works on queue ended with Done", () => |
| 684 | Effect.tx(Effect.gen(function*() { |
| 685 | const queue = yield* TxQueue.bounded<number, Cause.Done>(10) |
| 686 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 687 | yield* TxQueue.end(queue) |
| 688 | |
| 689 | // Take all items to move from Closing to Done state |
| 690 | yield* TxQueue.takeAll(queue) |
| 691 | |
| 692 | // Verify it's done before clearing |
| 693 | const isDoneBefore = yield* TxQueue.isDone(queue) |
| 694 | assert.strictEqual(isDoneBefore, true) |
| 695 | |
| 696 | // clear() returns empty array for halt causes (like Done) |
| 697 | const cleared = yield* TxQueue.clear(queue) |
| 698 | assert.deepStrictEqual(cleared, []) |
| 699 | |
| 700 | const size = yield* TxQueue.size(queue) |
| 701 | assert.strictEqual(size, 0) |
| 702 | |
| 703 | // Should still be done after clearing |
| 704 | const isDoneAfter = yield* TxQueue.isDone(queue) |
| 705 | assert.strictEqual(isDoneAfter, true) |
| 706 | }))) |
| 707 | }) |
| 708 | |
| 709 | describe("shutdown", () => { |
| 710 | it.effect("shutdown and isShutdown work correctly", () => |
| 711 | Effect.tx(Effect.gen(function*() { |
| 712 | const queue = yield* TxQueue.bounded<number>(10) |
| 713 | |
| 714 | const isShutdown1 = yield* TxQueue.isShutdown(queue) |
| 715 | assert.strictEqual(isShutdown1, false) |
| 716 | |
| 717 | yield* TxQueue.shutdown(queue) |
| 718 | const isShutdown2 = yield* TxQueue.isShutdown(queue) |
| 719 | assert.strictEqual(isShutdown2, true) |
| 720 | }))) |
| 721 | |
| 722 | it.effect("offer fails after shutdown", () => |
| 723 | Effect.tx(Effect.gen(function*() { |
| 724 | const queue = yield* TxQueue.bounded<number>(10) |
| 725 | yield* TxQueue.shutdown(queue) |
| 726 | |
| 727 | const offered = yield* TxQueue.offer(queue, 42) |
| 728 | assert.strictEqual(offered, false) |
| 729 | }))) |
| 730 | |
| 731 | it.effect("poll returns none after shutdown", () => |
| 732 | Effect.tx(Effect.gen(function*() { |
| 733 | const queue = yield* TxQueue.bounded<number>(10) |
| 734 | yield* TxQueue.offer(queue, 42) |
| 735 | yield* TxQueue.shutdown(queue) |
| 736 | |
| 737 | const maybe = yield* TxQueue.poll(queue) |
| 738 | assert.deepStrictEqual(maybe, Option.none()) |
| 739 | }))) |
| 740 | }) |
| 741 | |
| 742 | describe("dropping strategy", () => { |
| 743 | it.effect("dropping queue rejects items when full", () => |
| 744 | Effect.tx(Effect.gen(function*() { |
| 745 | const queue = yield* TxQueue.dropping<number>(2) |
| 746 | |
| 747 | // Fill to capacity |
| 748 | const accepted1 = yield* TxQueue.offer(queue, 1) |
| 749 | const accepted2 = yield* TxQueue.offer(queue, 2) |
| 750 | assert.strictEqual(accepted1, true) |
| 751 | assert.strictEqual(accepted2, true) |
| 752 | |
| 753 | // This should be dropped |
| 754 | const accepted3 = yield* TxQueue.offer(queue, 3) |
| 755 | assert.strictEqual(accepted3, false) |
| 756 | |
| 757 | const size = yield* TxQueue.size(queue) |
| 758 | assert.strictEqual(size, 2) |
| 759 | }))) |
| 760 | }) |
| 761 | |
| 762 | describe("sliding strategy", () => { |
| 763 | it.effect("sliding queue evicts old items when full", () => |
| 764 | Effect.tx(Effect.gen(function*() { |
| 765 | const queue = yield* TxQueue.sliding<number>(2) |
| 766 | |
| 767 | // Fill to capacity |
| 768 | yield* TxQueue.offer(queue, 1) |
| 769 | yield* TxQueue.offer(queue, 2) |
| 770 | |
| 771 | // This should evict item 1 |
| 772 | const accepted = yield* TxQueue.offer(queue, 3) |
| 773 | assert.strictEqual(accepted, true) |
| 774 | |
| 775 | const size = yield* TxQueue.size(queue) |
| 776 | assert.strictEqual(size, 2) |
| 777 | |
| 778 | // First item should be 2 (item 1 was evicted) |
| 779 | const item = yield* TxQueue.take(queue) |
| 780 | assert.strictEqual(item, 2) |
| 781 | }))) |
| 782 | }) |
| 783 | |
| 784 | describe("E-channel and exit signaling", () => { |
| 785 | describe("type ergonomics", () => { |
| 786 | it.effect("default E parameter provides clean typing", () => |
| 787 | Effect.tx(Effect.gen(function*() { |
| 788 | // Clean, concise typing with default E = never |
| 789 | const queue = yield* TxQueue.bounded<number>(5) |
| 790 | |
| 791 | yield* TxQueue.offer(queue, 42) |
| 792 | const item = yield* TxQueue.take(queue) // Effect<number, never> |
| 793 | assert.strictEqual(item, 42) |
| 794 | }))) |
| 795 | |
| 796 | it.effect("explicit E parameter works correctly", () => |
| 797 | Effect.tx(Effect.gen(function*() { |
| 798 | // Explicit error channel typing |
| 799 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 800 | |
| 801 | yield* TxQueue.offer(queue, 42) |
| 802 | const item = yield* TxQueue.take(queue) // Effect<number, never> (no errors when queue is open) |
| 803 | assert.strictEqual(item, 42) |
| 804 | }))) |
| 805 | }) |
| 806 | |
| 807 | describe("state transitions", () => { |
| 808 | it.effect("queue starts in Open state", () => |
| 809 | Effect.tx(Effect.gen(function*() { |
| 810 | const queue = yield* TxQueue.bounded<number>(5) |
| 811 | |
| 812 | const isOpen = yield* TxQueue.isOpen(queue) |
| 813 | const isClosing = yield* TxQueue.isClosing(queue) |
| 814 | const isDone = yield* TxQueue.isDone(queue) |
| 815 | |
| 816 | assert.strictEqual(isOpen, true) |
| 817 | assert.strictEqual(isClosing, false) |
| 818 | assert.strictEqual(isDone, false) |
| 819 | }))) |
| 820 | |
| 821 | it.effect("interrupt() transitions Open → Done when empty", () => |
| 822 | Effect.tx(Effect.gen(function*() { |
| 823 | const queue = yield* TxQueue.bounded<number>(5) |
| 824 | |
| 825 | const result = yield* TxQueue.interrupt(queue) |
| 826 | assert.strictEqual(result, true) |
| 827 | |
| 828 | const isOpen = yield* TxQueue.isOpen(queue) |
| 829 | const isClosing = yield* TxQueue.isClosing(queue) |
| 830 | const isDone = yield* TxQueue.isDone(queue) |
| 831 | |
| 832 | assert.strictEqual(isOpen, false) |
| 833 | assert.strictEqual(isClosing, false) |
| 834 | assert.strictEqual(isDone, true) |
| 835 | }))) |
| 836 | |
| 837 | it.effect("interrupt() transitions Open → Closing → Done when not empty", () => |
| 838 | Effect.tx(Effect.gen(function*() { |
| 839 | const queue = yield* TxQueue.bounded<number>(5) |
| 840 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 841 | |
| 842 | const result = yield* TxQueue.interrupt(queue) |
| 843 | assert.strictEqual(result, true) |
| 844 | |
| 845 | // Should be in Closing state |
| 846 | const isOpen1 = yield* TxQueue.isOpen(queue) |
| 847 | const isClosing1 = yield* TxQueue.isClosing(queue) |
| 848 | const isDone1 = yield* TxQueue.isDone(queue) |
| 849 | |
| 850 | assert.strictEqual(isOpen1, false) |
| 851 | assert.strictEqual(isClosing1, true) |
| 852 | assert.strictEqual(isDone1, false) |
| 853 | |
| 854 | // Take all items to trigger Closing → Done |
| 855 | yield* TxQueue.takeAll(queue) |
| 856 | |
| 857 | // Should now be Done |
| 858 | const isOpen2 = yield* TxQueue.isOpen(queue) |
| 859 | const isClosing2 = yield* TxQueue.isClosing(queue) |
| 860 | const isDone2 = yield* TxQueue.isDone(queue) |
| 861 | |
| 862 | assert.strictEqual(isOpen2, false) |
| 863 | assert.strictEqual(isClosing2, false) |
| 864 | assert.strictEqual(isDone2, true) |
| 865 | }))) |
| 866 | |
| 867 | it.effect("fail() transitions directly to Done", () => |
| 868 | Effect.tx(Effect.gen(function*() { |
| 869 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 870 | yield* TxQueue.offer(queue, 42) |
| 871 | |
| 872 | const result = yield* TxQueue.fail(queue, "test error") |
| 873 | assert.strictEqual(result, true) |
| 874 | |
| 875 | const isOpen = yield* TxQueue.isOpen(queue) |
| 876 | const isClosing = yield* TxQueue.isClosing(queue) |
| 877 | const isDone = yield* TxQueue.isDone(queue) |
| 878 | |
| 879 | assert.strictEqual(isOpen, false) |
| 880 | assert.strictEqual(isClosing, false) |
| 881 | assert.strictEqual(isDone, true) |
| 882 | }))) |
| 883 | |
| 884 | it.effect("done() with custom cause works correctly", () => |
| 885 | Effect.tx(Effect.gen(function*() { |
| 886 | const queue = yield* TxQueue.bounded<number>(5) |
| 887 | |
| 888 | const customCause = Cause.interrupt() |
| 889 | const result = yield* TxQueue.failCause(queue, customCause) |
| 890 | assert.strictEqual(result, true) |
| 891 | |
| 892 | const isDone = yield* TxQueue.isDone(queue) |
| 893 | assert.strictEqual(isDone, true) |
| 894 | }))) |
| 895 | }) |
| 896 | |
| 897 | describe("completion operations", () => { |
| 898 | it.effect("interrupt() completes queue successfully", () => |
| 899 | Effect.tx(Effect.gen(function*() { |
| 900 | const queue = yield* TxQueue.bounded<number>(5) |
| 901 | |
| 902 | const interruptResult = yield* TxQueue.interrupt(queue) |
| 903 | assert.strictEqual(interruptResult, true) |
| 904 | |
| 905 | // awaitCompletion should succeed |
| 906 | yield* TxQueue.awaitCompletion(queue) |
| 907 | }))) |
| 908 | |
| 909 | it.effect("fail() completes queue with error", () => |
| 910 | Effect.tx(Effect.gen(function*() { |
| 911 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 912 | |
| 913 | const failResult = yield* TxQueue.fail(queue, "test error") |
| 914 | assert.strictEqual(failResult, true) |
| 915 | |
| 916 | // awaitCompletion should succeed once queue is done (not propagate the error) |
| 917 | yield* TxQueue.awaitCompletion(queue) |
| 918 | }))) |
| 919 | |
| 920 | it.effect("multiple interrupt() calls return false after first", () => |
| 921 | Effect.tx(Effect.gen(function*() { |
| 922 | const queue = yield* TxQueue.bounded<number>(5) |
| 923 | |
| 924 | const result1 = yield* TxQueue.interrupt(queue) |
| 925 | const result2 = yield* TxQueue.interrupt(queue) |
| 926 | |
| 927 | assert.strictEqual(result1, true) |
| 928 | assert.strictEqual(result2, false) |
| 929 | }))) |
| 930 | |
| 931 | it.effect("operations on closed queue fail appropriately", () => |
| 932 | Effect.tx(Effect.gen(function*() { |
| 933 | const queue = yield* TxQueue.bounded<number>(5) |
| 934 | yield* TxQueue.interrupt(queue) |
| 935 | |
| 936 | // Offer should return false |
| 937 | const offerResult = yield* TxQueue.offer(queue, 42) |
| 938 | assert.strictEqual(offerResult, false) |
| 939 | |
| 940 | // Poll should return none |
| 941 | const pollResult = yield* TxQueue.poll(queue) |
| 942 | assert.deepStrictEqual(pollResult, Option.none()) |
| 943 | }))) |
| 944 | }) |
| 945 | |
| 946 | describe("awaitCompletion operation", () => { |
| 947 | it.effect("awaitCompletion succeeds when queue is interrupted", () => |
| 948 | Effect.tx(Effect.gen(function*() { |
| 949 | const queue = yield* TxQueue.bounded<number>(5) |
| 950 | yield* TxQueue.interrupt(queue) |
| 951 | |
| 952 | // Should succeed immediately when queue is already interrupted |
| 953 | yield* TxQueue.awaitCompletion(queue) |
| 954 | }))) |
| 955 | |
| 956 | it.effect("awaitCompletion succeeds when queue fails", () => |
| 957 | Effect.tx(Effect.gen(function*() { |
| 958 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 959 | yield* TxQueue.fail(queue, "connection lost") |
| 960 | |
| 961 | // Should succeed immediately since queue is done (errors are not propagated) |
| 962 | yield* TxQueue.awaitCompletion(queue) |
| 963 | }))) |
| 964 | |
| 965 | it.effect("awaitCompletion on already completed queue", () => |
| 966 | Effect.tx(Effect.gen(function*() { |
| 967 | const queue = yield* TxQueue.bounded<number>(5) |
| 968 | yield* TxQueue.interrupt(queue) |
| 969 | |
| 970 | // Should immediately succeed |
| 971 | yield* TxQueue.awaitCompletion(queue) |
| 972 | }))) |
| 973 | |
| 974 | it.effect("awaitCompletion on already failed queue", () => |
| 975 | Effect.tx(Effect.gen(function*() { |
| 976 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 977 | yield* TxQueue.fail(queue, "already failed") |
| 978 | |
| 979 | // Should immediately succeed since queue is done |
| 980 | yield* TxQueue.awaitCompletion(queue) |
| 981 | }))) |
| 982 | |
| 983 | it.effect("awaitCompletion with custom Cause.die() succeeds", () => |
| 984 | Effect.tx(Effect.gen(function*() { |
| 985 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 986 | const defect = new Error("system failure") |
| 987 | const defectCause = Cause.die(defect) |
| 988 | |
| 989 | yield* TxQueue.failCause(queue, defectCause) |
| 990 | |
| 991 | // awaitCompletion should succeed since queue is done (defects are not propagated) |
| 992 | yield* TxQueue.awaitCompletion(queue) |
| 993 | }))) |
| 994 | |
| 995 | it.effect("awaitCompletion with custom interrupt cause succeeds", () => |
| 996 | Effect.tx(Effect.gen(function*() { |
| 997 | const queue = yield* TxQueue.bounded<number>(5) |
| 998 | const customInterrupt = Cause.interrupt(12345) |
| 999 | |
| 1000 | yield* TxQueue.failCause(queue, customInterrupt) |
| 1001 | |
| 1002 | // awaitCompletion should succeed with interrupt-only causes |
| 1003 | yield* TxQueue.awaitCompletion(queue) |
| 1004 | }))) |
| 1005 | |
| 1006 | it.effect("awaitCompletion with fail cause succeeds", () => |
| 1007 | Effect.tx(Effect.gen(function*() { |
| 1008 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 1009 | |
| 1010 | // Complete queue with failure cause |
| 1011 | yield* TxQueue.fail(queue, "priority error") |
| 1012 | |
| 1013 | // awaitCompletion should succeed since queue is done (failures are not propagated) |
| 1014 | yield* TxQueue.awaitCompletion(queue) |
| 1015 | }))) |
| 1016 | |
| 1017 | it.effect("awaitCompletion concurrent behavior works", () => |
| 1018 | Effect.tx(Effect.gen(function*() { |
| 1019 | const queue = yield* TxQueue.bounded<number>(5) |
| 1020 | |
| 1021 | // Interrupt first, then await completion - should succeed immediately |
| 1022 | yield* TxQueue.interrupt(queue) |
| 1023 | yield* TxQueue.awaitCompletion(queue) |
| 1024 | }))) |
| 1025 | }) |
| 1026 | |
| 1027 | describe("legacy compatibility", () => { |
| 1028 | it.effect("isShutdown works as alias for isDone", () => |
| 1029 | Effect.tx(Effect.gen(function*() { |
| 1030 | const queue = yield* TxQueue.bounded<number>(5) |
| 1031 | |
| 1032 | const shutdown1 = yield* TxQueue.isShutdown(queue) |
| 1033 | const done1 = yield* TxQueue.isDone(queue) |
| 1034 | assert.strictEqual(shutdown1, done1) |
| 1035 | assert.strictEqual(shutdown1, false) |
| 1036 | |
| 1037 | yield* TxQueue.shutdown(queue) |
| 1038 | |
| 1039 | const shutdown2 = yield* TxQueue.isShutdown(queue) |
| 1040 | const done2 = yield* TxQueue.isDone(queue) |
| 1041 | assert.strictEqual(shutdown2, done2) |
| 1042 | assert.strictEqual(shutdown2, true) |
| 1043 | }))) |
| 1044 | |
| 1045 | it.effect("shutdown() clears queue and marks as done", () => |
| 1046 | Effect.tx(Effect.gen(function*() { |
| 1047 | const queue = yield* TxQueue.bounded<number>(5) |
| 1048 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 1049 | |
| 1050 | yield* TxQueue.shutdown(queue) |
| 1051 | |
| 1052 | const size = yield* TxQueue.size(queue) |
| 1053 | const isDone = yield* TxQueue.isDone(queue) |
| 1054 | |
| 1055 | assert.strictEqual(size, 0) |
| 1056 | assert.strictEqual(isDone, true) |
| 1057 | }))) |
| 1058 | }) |
| 1059 | |
| 1060 | describe("error channel behavior", () => { |
| 1061 | it.effect("take() on successfully interrupted queue interrupts", () => |
| 1062 | Effect.tx(Effect.gen(function*() { |
| 1063 | const queue = yield* TxQueue.bounded<number>(5) |
| 1064 | yield* TxQueue.interrupt(queue) |
| 1065 | |
| 1066 | // take() should interrupt on successfully interrupted queue |
| 1067 | const result = yield* Effect.exit(TxQueue.take(queue)) |
| 1068 | assert.strictEqual(result._tag, "Failure") |
| 1069 | }))) |
| 1070 | |
| 1071 | it.effect("take() on failed queue fails with error directly", () => |
| 1072 | Effect.tx(Effect.gen(function*() { |
| 1073 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 1074 | yield* TxQueue.fail(queue, "test error") |
| 1075 | |
| 1076 | // take() should fail with the error directly |
| 1077 | const result = yield* Effect.flip(TxQueue.take(queue)) |
| 1078 | assert.strictEqual(result, "test error") |
| 1079 | }))) |
| 1080 | |
| 1081 | it.effect("take() on shutdown queue interrupts (success exit)", () => |
| 1082 | Effect.tx(Effect.gen(function*() { |
| 1083 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 1084 | yield* TxQueue.shutdown(queue) // shutdown creates success exit |
| 1085 | |
| 1086 | // take() should interrupt since shutdown creates success exit |
| 1087 | const result = yield* Effect.exit(TxQueue.take(queue)) |
| 1088 | assert.strictEqual(result._tag, "Failure") |
| 1089 | }))) |
| 1090 | |
| 1091 | it.effect("take() with explicit interrupt cause propagates interruption", () => |
| 1092 | Effect.tx(Effect.gen(function*() { |
| 1093 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 1094 | const interruptCause = Cause.interrupt(123) |
| 1095 | yield* TxQueue.failCause(queue, interruptCause) |
| 1096 | |
| 1097 | // take() should propagate the interruption directly, not wrap in Option |
| 1098 | const result = yield* Effect.exit(TxQueue.take(queue)) |
| 1099 | assert.strictEqual(result._tag, "Failure") |
| 1100 | // The cause should contain an interrupt, not a typed error |
| 1101 | if (result._tag === "Failure") { |
| 1102 | assert.strictEqual(Cause.hasInterrupts(result.cause), true) |
| 1103 | } |
| 1104 | }))) |
| 1105 | |
| 1106 | it.effect("take() with custom Cause.fail extracts error correctly", () => |
| 1107 | Effect.tx(Effect.gen(function*() { |
| 1108 | const queue = yield* TxQueue.bounded<number, Error>(5) |
| 1109 | const customError = new Error("custom failure") |
| 1110 | const customCause = Cause.fail(customError) |
| 1111 | |
| 1112 | yield* TxQueue.failCause(queue, customCause) |
| 1113 | |
| 1114 | // take() should extract the custom error directly |
| 1115 | const result = yield* Effect.flip(TxQueue.take(queue)) |
| 1116 | assert.strictEqual(result, customError) |
| 1117 | }))) |
| 1118 | |
| 1119 | it.effect("take() with Cause.die (defect) propagates defect", () => |
| 1120 | Effect.tx(Effect.gen(function*() { |
| 1121 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 1122 | const defect = new Error("unexpected defect") |
| 1123 | const defectCause = Cause.die(defect) |
| 1124 | |
| 1125 | yield* TxQueue.failCause(queue, defectCause) |
| 1126 | |
| 1127 | // take() should propagate the defect directly, not wrap in Option |
| 1128 | const result = yield* Effect.exit(TxQueue.take(queue)) |
| 1129 | assert.strictEqual(result._tag, "Failure") |
| 1130 | // The cause should contain a defect |
| 1131 | if (result._tag === "Failure") { |
| 1132 | assert.strictEqual(Cause.hasDies(result.cause), true) |
| 1133 | } |
| 1134 | }))) |
| 1135 | |
| 1136 | it.effect("take() with typed error cause extracts error directly", () => |
| 1137 | Effect.tx(Effect.gen(function*() { |
| 1138 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 1139 | |
| 1140 | // Create a cause with typed error |
| 1141 | const typedError = "typed error" |
| 1142 | const failCause = Cause.fail(typedError) |
| 1143 | |
| 1144 | yield* TxQueue.failCause(queue, failCause) |
| 1145 | |
| 1146 | // take() should extract the typed error directly |
| 1147 | const result = yield* Effect.flip(TxQueue.take(queue)) |
| 1148 | assert.strictEqual(result, "typed error") |
| 1149 | }))) |
| 1150 | |
| 1151 | it.effect("peek() on successfully interrupted queue interrupts", () => |
| 1152 | Effect.tx(Effect.gen(function*() { |
| 1153 | const queue = yield* TxQueue.bounded<number>(5) |
| 1154 | yield* TxQueue.interrupt(queue) |
| 1155 | |
| 1156 | // peek() should interrupt on interrupted queue |
| 1157 | const result = yield* Effect.exit(TxQueue.peek(queue)) |
| 1158 | assert.strictEqual(result._tag, "Failure") |
| 1159 | }))) |
| 1160 | |
| 1161 | it.effect("peek() on failed queue fails with error directly", () => |
| 1162 | Effect.tx(Effect.gen(function*() { |
| 1163 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 1164 | yield* TxQueue.fail(queue, "peek error") |
| 1165 | |
| 1166 | // peek() should fail with the error directly |
| 1167 | const result = yield* Effect.flip(TxQueue.peek(queue)) |
| 1168 | assert.strictEqual(result, "peek error") |
| 1169 | }))) |
| 1170 | |
| 1171 | it.effect("poll() on failed queue returns none", () => |
| 1172 | Effect.tx(Effect.gen(function*() { |
| 1173 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 1174 | yield* TxQueue.offer(queue, 42) |
| 1175 | yield* TxQueue.fail(queue, "poll error") |
| 1176 | |
| 1177 | // poll() should return none on failed queue (doesn't propagate errors) |
| 1178 | const result = yield* TxQueue.poll(queue) |
| 1179 | assert.deepStrictEqual(result, Option.none()) |
| 1180 | }))) |
| 1181 | |
| 1182 | it.effect("takeN() with partial items available before failure", () => |
| 1183 | Effect.tx(Effect.gen(function*() { |
| 1184 | const queue = yield* TxQueue.bounded<number, string>(5) |
| 1185 | yield* TxQueue.offerAll(queue, [1, 2]) |
| 1186 | yield* TxQueue.fail(queue, "partial error") |
| 1187 | |
| 1188 | // takeN should fail even with some items available when queue is failed |
| 1189 | const result = yield* Effect.flip(TxQueue.takeN(queue, 3)) |
| 1190 | assert.strictEqual(result, "partial error") |
| 1191 | }))) |
| 1192 | |
| 1193 | it.effect("error propagation with different error types", () => |
| 1194 | Effect.tx(Effect.gen(function*() { |
| 1195 | const queue1 = yield* TxQueue.bounded<number, Error>(5) |
| 1196 | const queue2 = yield* TxQueue.bounded<number, { code: number; message: string }>(5) |
| 1197 | |
| 1198 | const customError = new Error("custom error") |
| 1199 | const customObject = { code: 404, message: "not found" } |
| 1200 | |
| 1201 | yield* TxQueue.fail(queue1, customError) |
| 1202 | yield* TxQueue.fail(queue2, customObject) |
| 1203 | |
| 1204 | // Should propagate Error objects correctly |
| 1205 | const result1 = yield* Effect.flip(TxQueue.take(queue1)) |
| 1206 | assert.strictEqual(result1, customError) |
| 1207 | |
| 1208 | // Should propagate custom object errors correctly |
| 1209 | const result2 = yield* Effect.flip(TxQueue.take(queue2)) |
| 1210 | assert.deepStrictEqual(result2, customObject) |
| 1211 | }))) |
| 1212 | |
| 1213 | it.effect("closing queue allows remaining items to be consumed", () => |
| 1214 | Effect.tx(Effect.gen(function*() { |
| 1215 | const queue = yield* TxQueue.bounded<number>(5) |
| 1216 | yield* TxQueue.offerAll(queue, [1, 2, 3]) |
| 1217 | |
| 1218 | // Interrupt the queue (puts it in Closing state) |
| 1219 | yield* TxQueue.interrupt(queue) |
| 1220 | |
| 1221 | // Should still be able to take items |
| 1222 | const item1 = yield* TxQueue.take(queue) |
| 1223 | const item2 = yield* TxQueue.take(queue) |
| 1224 | |
| 1225 | assert.strictEqual(item1, 1) |
| 1226 | assert.strictEqual(item2, 2) |
| 1227 | |
| 1228 | // Queue should still be closing |
| 1229 | const isClosing = yield* TxQueue.isClosing(queue) |
| 1230 | assert.strictEqual(isClosing, true) |
| 1231 | |
| 1232 | // Take last item - should transition to Done |
| 1233 | const item3 = yield* TxQueue.take(queue) |
| 1234 | assert.strictEqual(item3, 3) |
| 1235 | |
| 1236 | // Queue should now be Done |
| 1237 | const isDone = yield* TxQueue.isDone(queue) |
| 1238 | assert.strictEqual(isDone, true) |
| 1239 | }))) |
| 1240 | }) |
| 1241 | }) |
| 1242 | |
| 1243 | describe("Done and end function", () => { |
| 1244 | it.effect("end() signals completion with Done", () => |
| 1245 | Effect.tx(Effect.gen(function*() { |
| 1246 | const queue = yield* TxQueue.bounded<number, Cause.Done>(5) |
| 1247 | yield* TxQueue.offer(queue, 1) |
| 1248 | yield* TxQueue.offer(queue, 2) |
| 1249 | |
| 1250 | // Signal the end of the queue |
| 1251 | const result = yield* TxQueue.end(queue) |
| 1252 | assert.strictEqual(result, true) |
| 1253 | |
| 1254 | // Queue should be closing since it has items |
| 1255 | const isClosing = yield* TxQueue.isClosing(queue) |
| 1256 | assert.strictEqual(isClosing, true) |
| 1257 | |
| 1258 | // Consume all items to transition to Done |
| 1259 | yield* TxQueue.takeAll(queue) |
| 1260 | |
| 1261 | // Queue should now be done |
| 1262 | const isDone = yield* TxQueue.isDone(queue) |
| 1263 | assert.strictEqual(isDone, true) |
| 1264 | }))) |
| 1265 | |
| 1266 | it.effect("take() on ended queue fails with Done", () => |
| 1267 | Effect.tx(Effect.gen(function*() { |
| 1268 | const queue = yield* TxQueue.bounded<number, Cause.Done>(5) |
| 1269 | yield* TxQueue.offer(queue, 42) |
| 1270 | yield* TxQueue.end(queue) |
| 1271 | |
| 1272 | // First take should still get the item since queue is in Closing state |
| 1273 | const item = yield* TxQueue.take(queue) |
| 1274 | assert.strictEqual(item, 42) |
| 1275 | |
| 1276 | // Second take should fail with Done since queue is now done |
| 1277 | const takeResult = yield* Effect.flip(TxQueue.take(queue)) |
| 1278 | assert.strictEqual(Cause.isDone(takeResult), true) |
| 1279 | assert.strictEqual(takeResult, Cause.Done()) |
| 1280 | }))) |
| 1281 | |
| 1282 | it.effect("end() works with TxEnqueue interface", () => |
| 1283 | Effect.tx(Effect.gen(function*() { |
| 1284 | const queue = yield* TxQueue.bounded<number, Cause.Done>(5) |
| 1285 | |
| 1286 | yield* TxQueue.offer(queue, 1) |
| 1287 | const result = yield* TxQueue.end(queue) |
| 1288 | assert.strictEqual(result, true) |
| 1289 | |
| 1290 | // Queue should be closing since it has items |
| 1291 | const isClosing = yield* TxQueue.isClosing(queue) |
| 1292 | assert.strictEqual(isClosing, true) |
| 1293 | |
| 1294 | // Consume the item to transition to Done |
| 1295 | yield* TxQueue.take(queue) |
| 1296 | |
| 1297 | // Now verify queue is done |
| 1298 | const isDone = yield* TxQueue.isDone(queue) |
| 1299 | assert.strictEqual(isDone, true) |
| 1300 | }))) |
| 1301 | |
| 1302 | it.effect("end() returns false if queue is already done", () => |
| 1303 | Effect.tx(Effect.gen(function*() { |
| 1304 | const queue = yield* TxQueue.bounded<number, Cause.Done>(5) |
| 1305 | |
| 1306 | // End the queue once |
| 1307 | const result1 = yield* TxQueue.end(queue) |
| 1308 | assert.strictEqual(result1, true) |
| 1309 | |
| 1310 | // Try to end it again |
| 1311 | const result2 = yield* TxQueue.end(queue) |
| 1312 | assert.strictEqual(result2, false) |
| 1313 | }))) |
| 1314 | }) |
| 1315 | }) |