// ets_tracing: off
import * as ChunkFilter from "../Collections/Immutable/Chunk/api/filter.js"
import * as Chunk from "../Collections/Immutable/Chunk/core.js"
import type { AtomicBoolean } from "../Support/AtomicBoolean/index.js"
import type { MutableQueue } from "../Support/MutableQueue/index.js"
import { EmptyQueue } from "../Support/MutableQueue/index.js"
import * as T from "./effect.js"
import * as P from "./promise.js"
import type { XQueue } from "./xqueue.js"
import { concreteQueue } from "./xqueue.js"
export { Dequeue, Queue, XQueue } from "./xqueue.js"
export interface Strategy {
readonly handleSurplus: (
as: Chunk.Chunk,
queue: MutableQueue,
takers: MutableQueue>,
isShutdown: AtomicBoolean
) => T.UIO
readonly unsafeOnQueueEmptySpace: (
queue: MutableQueue,
takers: MutableQueue>
) => void
readonly surplusSize: number
readonly shutdown: T.UIO
}
export class DroppingStrategy implements Strategy {
handleSurplus(
_as: Chunk.Chunk,
_queue: MutableQueue,
_takers: MutableQueue>,
_isShutdown: AtomicBoolean
): T.UIO {
return T.succeed(false)
}
unsafeOnQueueEmptySpace(_queue: MutableQueue) {
//
}
get shutdown(): T.UIO {
return T.unit
}
get surplusSize(): number {
return 0
}
}
export class SlidingStrategy implements Strategy {
handleSurplus(
as: Chunk.Chunk,
queue: MutableQueue,
takers: MutableQueue>,
_isShutdown: AtomicBoolean
): T.UIO {
return T.succeedWith(() => {
this.unsafeSlidingOffer(queue, as)
unsafeCompleteTakers(this, queue, takers)
return true
})
}
unsafeOnQueueEmptySpace(_queue: MutableQueue) {
//
}
get shutdown(): T.UIO {
return T.unit
}
get surplusSize(): number {
return 0
}
private unsafeSlidingOffer(queue: MutableQueue, as: Chunk.Chunk) {
let bs = as
while (Chunk.size(bs) > 0) {
if (queue.capacity === 0) {
return
}
// poll 1 and retry
queue.poll(EmptyQueue)
if (queue.offer(Chunk.unsafeGet_(bs, 0))) {
bs = Chunk.drop_(bs, 1)
}
}
}
}
export function unsafeCompletePromise(p: P.Promise, a: A) {
return P.unsafeDone(T.succeed(a))(p)
}
export function unsafeCompleteTakers(
strategy: Strategy,
queue: MutableQueue,
takers: MutableQueue>
) {
let keepPolling = true
while (keepPolling && !queue.isEmpty) {
const taker = takers.poll(EmptyQueue)
if (taker !== EmptyQueue) {
const element = queue.poll(EmptyQueue)
if (element !== EmptyQueue) {
unsafeCompletePromise(taker, element)
strategy.unsafeOnQueueEmptySpace(queue, takers)
} else {
unsafeOfferAll(takers, Chunk.prepend_(unsafePollAll(takers), taker))
}
keepPolling = true
} else {
keepPolling = false
}
}
}
export function unsafeRemove(q: MutableQueue, a: A) {
unsafeOfferAll(
q,
ChunkFilter.filter_(unsafePollAll(q), (b) => a !== b)
)
}
export function unsafePollN(q: MutableQueue, max: number): Chunk.Chunk {
return q.pollUpTo(max)
}
export function unsafeOfferAll(
q: MutableQueue,
as: Chunk.Chunk
): Chunk.Chunk {
return q.offerAll(as)
}
export function unsafePollAll(q: MutableQueue): Chunk.Chunk {
let as = Chunk.empty()
while (!q.isEmpty) {
const elem = q.poll(EmptyQueue)!
if (elem !== EmptyQueue) {
as = Chunk.append_(as, elem)
}
}
return as
}
/**
* Waits until the queue is shutdown.
* The `IO` returned by this method will not resume until the queue has been shutdown.
* If the queue is already shutdown, the `IO` will resume right away.
*/
export function awaitShutdown(
self: XQueue
) {
concreteQueue(self)
return self.awaitShutdown
}
/**
* How many elements can hold in the queue
*/
export function capacity(self: XQueue) {
concreteQueue(self)
return self.capacity
}
/**
* `true` if `shutdown` has been called.
*/
export function isShutdown(self: XQueue) {
concreteQueue(self)
return self.isShutdown
}
/**
* Places one value in the queue.
*
* @ets_data_first offer_
*/
export function offer(a: A) {
return (self: XQueue) => offer_(self, a)
}
/**
* Places one value in the queue.
*/
export function offer_(self: XQueue, a: A) {
concreteQueue(self)
return self.offer(a)
}
/**
* Places one value in the queue.
*
* @ets_data_first offerTo_
*/
export function offerTo(self: XQueue) {
return (a: A) => offer_(self, a)
}
/**
* Places one value in the queue.
*
* @ets_data_first offerTo_
*/
export function offerTo_(
a: A,
self: XQueue
) {
return offer_(self, a)
}
/**
* For Bounded Queue: uses the `BackPressure` Strategy, places the values in the queue and always returns true.
* If the queue has reached capacity, then
* the fiber performing the `offerAll` will be suspended until there is room in
* the queue.
*
* For Unbounded Queue:
* Places all values in the queue and returns true.
*
* For Sliding Queue: uses `Sliding` Strategy
* If there is room in the queue, it places the values otherwise it removes the old elements and
* enqueues the new ones. Always returns true.
*
* For Dropping Queue: uses `Dropping` Strategy,
* It places the values in the queue but if there is no room it will not enqueue them and return false.
*
* @ets_data_first offerAll_
*/
export function offerAll(as: Iterable) {
return (self: XQueue) => offerAll_(self, as)
}
/**
* For Bounded Queue: uses the `BackPressure` Strategy, places the values in the queue and always returns true.
* If the queue has reached capacity, then
* the fiber performing the `offerAll` will be suspended until there is room in
* the queue.
*
* For Unbounded Queue:
* Places all values in the queue and returns true.
*
* For Sliding Queue: uses `Sliding` Strategy
* If there is room in the queue, it places the values otherwise it removes the old elements and
* enqueues the new ones. Always returns true.
*
* For Dropping Queue: uses `Dropping` Strategy,
* It places the values in the queue but if there is no room it will not enqueue them and return false.
*/
export function offerAll_(
self: XQueue,
as: Iterable
) {
concreteQueue(self)
return self.offerAll(as)
}
/**
* Interrupts any fibers that are suspended on `offer` or `take`.
* Future calls to `offer*` and `take*` will be interrupted immediately.
*/
export function shutdown(self: XQueue) {
concreteQueue(self)
return self.shutdown
}
/**
* Retrieves the size of the queue, which is equal to the number of elements
* in the queue. This may be negative if fibers are suspended waiting for
* elements to be added to the queue.
*/
export function size(self: XQueue) {
concreteQueue(self)
return self.size
}
/**
* Removes 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.
*/
export function take(self: XQueue) {
concreteQueue(self)
return self.take
}
/**
* Removes all the values in the queue and returns the list of the values. If the queue
* is empty returns empty list.
*/
export function takeAll(self: XQueue) {
concreteQueue(self)
return self.takeAll
}
/**
* Takes up to max number of values in the queue.
*
* @ets_data_first takeAllUpTo_
*/
export function takeAllUpTo(n: number) {
return (self: XQueue) =>
takeAllUpTo_(self, n)
}
/**
* Takes up to max number of values in the queue.
*/
export function takeAllUpTo_(
self: XQueue,
n: number
) {
concreteQueue(self)
return self.takeUpTo(n)
}