Dequeue<A> and Queue<A> is subtype of Effect<A> (#3591)

Co-authored-by: maksim.khramtsov <maksim.khramtsov@btsdigital.kz>

Author: 2 people

.changeset/proud-cats-shake.md

@@ -0,0 +1,15 @@
+---
+"effect": minor
+---
+
+`Dequeue<A>` and `Queue<A>` is subtype of `Effect<A>`. This means that now it can be used as an `Effect`, and when called, it will automatically extract and return an item from the queue, without having to explicitly use the `Queue.take` function.
+
+```ts
+Effect.gen(function* () {
+  const queue = yield* Queue.unbounded<number>()
+  yield* Queue.offer(queue, 1)
+  yield* Queue.offer(queue, 2)
+  const oldWay = yield* Queue.take(queue)
+  const newWay = yield* queue
+})
+```

packages/effect/dtslint/Unify.ts

@@ -6,6 +6,7 @@ import type * as Fiber from "effect/Fiber"
 import type * as FiberRef from "effect/FiberRef"
 import type * as Micro from "effect/Micro"
 import type * as Option from "effect/Option"
+import type * as Queue from "effect/Queue"
 import type * as RcRef from "effect/RcRef"
 import type * as Ref from "effect/Ref"
 import type * as Stream from "effect/Stream"
@@ -90,7 +91,18 @@ export type RuntimeFiberUnify = Unify.Unify<
   | Fiber.RuntimeFiber<"a", "b">
 >
 
-// $ExpectType 0 | Option<string | number> | Ref<1> | SynchronizedRef<1> | SubscriptionRef<1> | Deferred<1, 2> | Deferred<"a", "b"> | Fiber<"a" | 1, "b" | 2> | RuntimeFiber<"a" | 1, "b" | 2> | Ref<"A"> | SynchronizedRef<"A"> | SubscriptionRef<"A"> | FiberRef<12> | FiberRef<"a2"> | Either<1 | "A", 0 | "E"> | Effect<1 | "A", 0 | "E", "R" | "R1"> | RcRef<1 | "A", 0 | "E">
+// $ExpectType Queue<1> | Queue<"a">
+export type QueueUnify = Unify.Unify<
+  | Queue.Queue<1>
+  | Queue.Queue<"a">
+>
+// $ExpectType Dequeue<"a" | 1>
+export type DequeueUnify = Unify.Unify<
+  | Queue.Dequeue<1>
+  | Queue.Dequeue<"a">
+>
+
+// $ExpectType 0 | Option<string | number> | Ref<1> | SynchronizedRef<1> | SubscriptionRef<1> | Deferred<1, 2> | Deferred<"a", "b"> | Fiber<"a" | 1, "b" | 2> | RuntimeFiber<"a" | 1, "b" | 2> | Queue<1> | Queue<"a"> | Dequeue<"a" | 1> | Ref<"A"> | SynchronizedRef<"A"> | SubscriptionRef<"A"> | FiberRef<12> | FiberRef<"a2"> | Either<1 | "A", 0 | "E"> | Effect<1 | "A", 0 | "E", "R" | "R1"> | RcRef<1 | "A", 0 | "E">
 export type AllUnify = Unify.Unify<
   | Either.Either<1, 0>
   | Either.Either<"A", "E">
@@ -114,5 +126,9 @@ export type AllUnify = Unify.Unify<
   | Fiber.Fiber<"a", "b">
   | Fiber.RuntimeFiber<1, 2>
   | Fiber.RuntimeFiber<"a", "b">
+  | Queue.Queue<1>
+  | Queue.Queue<"a">
+  | Queue.Dequeue<1>
+  | Queue.Dequeue<"a">
   | 0
 >

packages/effect/src/Queue.ts

@@ -10,6 +10,7 @@ import type * as MutableRef from "./MutableRef.js"
 import type * as Option from "./Option.js"
 import type { Pipeable } from "./Pipeable.js"
 import type * as Types from "./Types.js"
+import type * as Unify from "./Unify.js"
 
 /**
  * @since 2.0.0
@@ -74,6 +75,26 @@ export interface Queue<in out A> extends Enqueue<A>, Dequeue<A>, Pipeable {
   readonly shutdownFlag: MutableRef.MutableRef<boolean>
   /** @internal */
   readonly strategy: Strategy<A>
+
+  readonly [Unify.typeSymbol]?: unknown
+  readonly [Unify.unifySymbol]?: QueueUnify<this>
+  readonly [Unify.ignoreSymbol]?: QueueUnifyIgnore
+}
+
+/**
+ * @category models
+ * @since 3.8.0
+ */
+export interface QueueUnify<A extends { [Unify.typeSymbol]?: any }> extends DequeueUnify<A> {
+  Queue?: () => Extract<A[Unify.typeSymbol], Queue<any>>
+}
+
+/**
+ * @category models
+ * @since 3.8.0
+ */
+export interface QueueUnifyIgnore extends DequeueUnifyIgnore {
+  Dequeue?: true
 }
 
 /**
@@ -113,7 +134,7 @@ export interface Enqueue<in A> extends Queue.EnqueueVariance<A>, BaseQueue, Pipe
  * @since 2.0.0
  * @category models
  */
-export interface Dequeue<out A> extends Queue.DequeueVariance<A>, BaseQueue, Pipeable {
+export interface Dequeue<out A> extends Effect.Effect<A>, Queue.DequeueVariance<A>, BaseQueue, Pipeable {
   /**
    * Takes the oldest value in the queue. If the queue is empty, this will return
    * a computation that resumes when an item has been added to the queue.
@@ -137,6 +158,26 @@ export interface Dequeue<out A> extends Queue.DequeueVariance<A>, BaseQueue, Pip
    * suspends until at least the minimum number of elements have been collected.
    */
   takeBetween(min: number, max: number): Effect.Effect<Chunk.Chunk<A>>
+
+  readonly [Unify.typeSymbol]?: unknown
+  readonly [Unify.unifySymbol]?: DequeueUnify<this>
+  readonly [Unify.ignoreSymbol]?: DequeueUnifyIgnore
+}
+
+/**
+ * @category models
+ * @since 3.8.0
+ */
+export interface DequeueUnify<A extends { [Unify.typeSymbol]?: any }> extends Effect.EffectUnify<A> {
+  Dequeue?: () => A[Unify.typeSymbol] extends Dequeue<infer A0> | infer _ ? Dequeue<A0> : never
+}
+
+/**
+ * @category models
+ * @since 3.8.0
+ */
+export interface DequeueUnifyIgnore extends Effect.EffectUnifyIgnore {
+  Effect?: true
 }
 
 /**

packages/effect/src/internal/groupBy.ts

@@ -3,6 +3,7 @@ import type * as Channel from "../Channel.js"
 import * as Chunk from "../Chunk.js"
 import * as Deferred from "../Deferred.js"
 import * as Effect from "../Effect.js"
+import * as Effectable from "../Effectable.js"
 import * as Exit from "../Exit.js"
 import { dual, pipe } from "../Function.js"
 import type * as GroupBy from "../GroupBy.js"
@@ -316,7 +317,7 @@ export const bindEffect = dual<
 
 const mapDequeue = <A, B>(dequeue: Queue.Dequeue<A>, f: (a: A) => B): Queue.Dequeue<B> => new MapDequeue(dequeue, f)
 
-class MapDequeue<in out A, out B> implements Queue.Dequeue<B> {
+class MapDequeue<in out A, out B> extends Effectable.Class<B> implements Queue.Dequeue<B> {
   readonly [Queue.DequeueTypeId] = {
     _Out: (_: never) => _
   }
@@ -325,6 +326,7 @@ class MapDequeue<in out A, out B> implements Queue.Dequeue<B> {
     readonly dequeue: Queue.Dequeue<A>,
     readonly f: (a: A) => B
   ) {
+    super()
   }
 
   capacity(): number {
@@ -390,6 +392,10 @@ class MapDequeue<in out A, out B> implements Queue.Dequeue<B> {
   pipe() {
     return pipeArguments(this, arguments)
   }
+
+  commit() {
+    return this.take
+  }
 }
 
 /** @internal */

packages/effect/src/internal/pubsub.ts

@@ -1,6 +1,7 @@
 import * as Chunk from "../Chunk.js"
 import type * as Deferred from "../Deferred.js"
 import type * as Effect from "../Effect.js"
+import * as Effectable from "../Effectable.js"
 import { dual, pipe } from "../Function.js"
 import * as MutableQueue from "../MutableQueue.js"
 import * as MutableRef from "../MutableRef.js"
@@ -909,7 +910,7 @@ class UnboundedPubSubSubscription<in out A> implements Subscription<A> {
 }
 
 /** @internal */
-class SubscriptionImpl<in out A> implements Queue.Dequeue<A> {
+class SubscriptionImpl<in out A> extends Effectable.Class<A> implements Queue.Dequeue<A> {
   [queue.DequeueTypeId] = queue.dequeueVariance
 
   constructor(
@@ -921,7 +922,13 @@ class SubscriptionImpl<in out A> implements Queue.Dequeue<A> {
     readonly shutdownFlag: MutableRef.MutableRef<boolean>,
     readonly strategy: PubSubStrategy<A>,
     readonly replayWindow: ReplayWindow<A>
-  ) {}
+  ) {
+    super()
+  }
+
+  commit() {
+    return this.take
+  }
 
   pipe() {
     return pipeArguments(this, arguments)

packages/effect/src/internal/queue.ts

@@ -2,6 +2,7 @@ import * as Arr from "../Array.js"
 import * as Chunk from "../Chunk.js"
 import type * as Deferred from "../Deferred.js"
 import type * as Effect from "../Effect.js"
+import * as Effectable from "../Effectable.js"
 import { dual, pipe } from "../Function.js"
 import * as MutableQueue from "../MutableQueue.js"
 import * as MutableRef from "../MutableRef.js"
@@ -63,7 +64,7 @@ export const dequeueVariance = {
 }
 
 /** @internal */
-class QueueImpl<in out A> implements Queue.Queue<A> {
+class QueueImpl<in out A> extends Effectable.Class<A> implements Queue.Queue<A> {
   readonly [EnqueueTypeId] = enqueueVariance
   readonly [DequeueTypeId] = dequeueVariance
 
@@ -79,12 +80,17 @@ class QueueImpl<in out A> implements Queue.Queue<A> {
     /** @internal */
     readonly strategy: Queue.Strategy<A>
   ) {
+    super()
   }
 
   pipe() {
     return pipeArguments(this, arguments)
   }
 
+  commit() {
+    return this.take
+  }
+
   capacity(): number {
     return this.queue.capacity()
   }

packages/effect/test/Queue.test.ts

@@ -775,4 +775,14 @@ describe("Queue", () => {
         expect(queue.pipe(identity)).toBe(queue)
       })
   )
+  it.effect(
+    "is subtype of Effect",
+    () =>
+      Effect.gen(function*() {
+        const queue = yield* Queue.unbounded<number>()
+        yield* Queue.offer(queue, 1)
+        const result1 = yield* queue
+        assert.strictEqual(result1, 1)
+      })
+  )
 })

packages/effect/test/Sink/constructors.test.ts

@@ -1,6 +1,7 @@
 import type * as Chunk from "effect/Chunk"
 import * as Deferred from "effect/Deferred"
 import * as Effect from "effect/Effect"
+import * as Effectable from "effect/Effectable"
 import * as Exit from "effect/Exit"
 import * as Fiber from "effect/Fiber"
 import { pipe } from "effect/Function"
@@ -87,7 +88,7 @@ describe("Sink", () => {
 
 const createQueueSpy = <A>(queue: Queue.Queue<A>): Queue.Queue<A> => new QueueSpy(queue)
 
-class QueueSpy<A> implements Queue.Queue<A> {
+class QueueSpy<A> extends Effectable.Class<A> implements Queue.Queue<A> {
   readonly [Queue.DequeueTypeId] = internalQueue.dequeueVariance
   readonly [Queue.EnqueueTypeId] = internalQueue.enqueueVariance
   private isShutdownInternal = false
@@ -98,13 +99,18 @@ class QueueSpy<A> implements Queue.Queue<A> {
   readonly takers: MutableQueue.MutableQueue<Deferred.Deferred<A, never>>
 
   constructor(readonly backingQueue: Queue.Queue<A>) {
+    super()
     this.queue = backingQueue.queue
     this.shutdownFlag = backingQueue.shutdownFlag
     this.shutdownHook = backingQueue.shutdownHook
     this.strategy = backingQueue.strategy
     this.takers = backingQueue.takers
   }
 
+  commit() {
+    return this.take
+  }
+
   pipe() {
     return pipeArguments(this, arguments)
   }