Socket
Socket
Sign inDemoInstall

@effect/stream

Package Overview
Dependencies
Maintainers
3
Versions
62
Alerts
File Explorer

Advanced tools

License
Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

@effect/stream - npm Package Compare versions

Comparing version 0.23.0 to 0.24.0

249

internal/channel.js

@@ -9,3 +9,2 @@ "use strict";

var Context = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/data/Context"));
var Debug = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/data/Debug"));
var Either = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/data/Either"));

@@ -37,74 +36,74 @@ var Equal = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/data/Equal"));

/** @internal */
const acquireUseRelease = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => (acquire, use, release) => core.flatMap(core.fromEffect(Ref.make(() => Effect.unit())), ref => core.ensuringWith(exit => Effect.flatMap(Ref.get(ref), f => f(exit)))(core.flatMap(restore(use))(core.fromEffect(Effect.uninterruptible(Effect.tap(acquire, a => Ref.set(ref, exit => restore(release)(a, exit)))))))).traced(trace));
const acquireUseRelease = (acquire, use, release) => core.flatMap(core.fromEffect(Ref.make(() => Effect.unit())), ref => core.ensuringWith(exit => Effect.flatMap(Ref.get(ref), f => f(exit)))(core.flatMap(use)(core.fromEffect(Effect.uninterruptible(Effect.tap(acquire, a => Ref.set(ref, exit => release(a, exit))))))));
/** @internal */
exports.acquireUseRelease = acquireUseRelease;
const as = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, value) => map(self, () => value).traced(trace));
const as = /*#__PURE__*/(0, _Function.dual)(2, (self, value) => map(self, () => value));
/** @internal */
exports.as = as;
const asUnit = /*#__PURE__*/Debug.methodWithTrace(trace => self => map(self, _Function.constVoid).traced(trace));
const asUnit = self => map(self, _Function.constVoid);
/** @internal */
exports.asUnit = asUnit;
const buffer = /*#__PURE__*/Debug.methodWithTrace(trace => (empty, isEmpty, ref) => core.suspend(() => {
const buffer = (empty, isEmpty, ref) => core.suspend(() => {
const doBuffer = (empty, isEmpty, ref) => unwrap(Ref.modify(ref, inElem => isEmpty(inElem) ? [core.readWith(input => core.flatMap(core.write(input), () => doBuffer(empty, isEmpty, ref)), error => core.fail(error), done => core.succeedNow(done)), inElem] : [core.flatMap(core.write(inElem), () => doBuffer(empty, isEmpty, ref)), empty]));
return doBuffer(empty, isEmpty, ref);
}).traced(trace));
});
/** @internal */
exports.buffer = buffer;
const bufferChunk = /*#__PURE__*/Debug.methodWithTrace(trace => ref => buffer(Chunk.empty(), Chunk.isEmpty, ref).traced(trace));
const bufferChunk = ref => buffer(Chunk.empty(), Chunk.isEmpty, ref);
/** @internal */
exports.bufferChunk = bufferChunk;
const catchAll = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => core.catchAllCause(self, cause => Either.match(Cause.failureOrCause(cause), restore(f), core.failCause)).traced(trace));
const catchAll = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => core.catchAllCause(self, cause => Either.match(Cause.failureOrCause(cause), f, core.failCause)));
/** @internal */
exports.catchAll = catchAll;
const concatMap = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => core.concatMapWith(self, restore(f), () => void 0, () => void 0).traced(trace));
const concatMap = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => core.concatMapWith(self, f, () => void 0, () => void 0));
/** @internal */
exports.concatMap = concatMap;
const collect = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, pf) => {
const collector = core.readWith(out => Option.match(restore(pf)(out), () => collector, out2 => core.flatMap(core.write(out2), () => collector)), core.fail, core.succeedNow);
return core.pipeTo(self, collector).traced(trace);
const collect = /*#__PURE__*/(0, _Function.dual)(2, (self, pf) => {
const collector = core.readWith(out => Option.match(pf(out), () => collector, out2 => core.flatMap(core.write(out2), () => collector)), core.fail, core.succeedNow);
return core.pipeTo(self, collector);
});
/** @internal */
exports.collect = collect;
const concatOut = /*#__PURE__*/Debug.methodWithTrace(trace => self => core.concatAll(self).traced(trace));
const concatOut = self => core.concatAll(self);
/** @internal */
exports.concatOut = concatOut;
const contramap = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.write(inElem), () => reader), core.fail, done => core.succeedNow(restore(f)(done)));
return core.pipeTo(reader, self).traced(trace);
const contramap = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.write(inElem), () => reader), core.fail, done => core.succeedNow(f(done)));
return core.pipeTo(reader, self);
});
/** @internal */
exports.contramap = contramap;
const contramapEffect = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.write(inElem), () => reader), core.fail, done => core.fromEffect(restore(f)(done)));
return core.pipeTo(reader, self).traced(trace);
const contramapEffect = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.write(inElem), () => reader), core.fail, done => core.fromEffect(f(done)));
return core.pipeTo(reader, self);
});
/** @internal */
exports.contramapEffect = contramapEffect;
const contramapError = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.write(inElem), () => reader), error => core.fail(restore(f)(error)), core.succeedNow);
return core.pipeTo(reader, self).traced(trace);
const contramapError = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.write(inElem), () => reader), error => core.fail(f(error)), core.succeedNow);
return core.pipeTo(reader, self);
});
/** @internal */
exports.contramapError = contramapError;
const contramapErrorEffect = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.write(inElem), () => reader), error => core.fromEffect(restore(f)(error)), core.succeedNow);
return core.pipeTo(reader, self).traced(trace);
const contramapErrorEffect = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.write(inElem), () => reader), error => core.fromEffect(f(error)), core.succeedNow);
return core.pipeTo(reader, self);
});
/** @internal */
exports.contramapErrorEffect = contramapErrorEffect;
const contramapIn = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.write(restore(f)(inElem)), () => reader), core.fail, core.succeedNow);
return core.pipeTo(reader, self).traced(trace);
const contramapIn = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.write(f(inElem)), () => reader), core.fail, core.succeedNow);
return core.pipeTo(reader, self);
});
exports.contramapIn = contramapIn;
const contramapInEffect = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.flatMap(core.fromEffect(restore(f)(inElem)), core.write), () => reader), core.fail, core.succeedNow);
return core.pipeTo(reader, self).traced(trace);
const contramapInEffect = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => {
const reader = core.readWith(inElem => core.flatMap(core.flatMap(core.fromEffect(f(inElem)), core.write), () => reader), core.fail, core.succeedNow);
return core.pipeTo(reader, self);
});
/** @internal */
exports.contramapInEffect = contramapInEffect;
const doneCollect = /*#__PURE__*/Debug.methodWithTrace(trace => self => core.suspend(() => {
const doneCollect = self => core.suspend(() => {
const builder = [];
return core.flatMap(outDone => core.succeed([Chunk.unsafeFromArray(builder), outDone]))(core.pipeTo(self, doneCollectReader(builder))).traced(trace);
}));
return core.flatMap(outDone => core.succeed([Chunk.unsafeFromArray(builder), outDone]))(core.pipeTo(self, doneCollectReader(builder)));
});
/** @internal */

@@ -118,30 +117,30 @@ exports.doneCollect = doneCollect;

/** @internal */
const drain = /*#__PURE__*/Debug.methodWithTrace(trace => self => {
const drain = self => {
const drainer = core.readWithCause(() => drainer, core.failCause, core.succeed);
return core.pipeTo(self, drainer).traced(trace);
});
return core.pipeTo(self, drainer);
};
/** @internal */
exports.drain = drain;
const emitCollect = /*#__PURE__*/Debug.methodWithTrace(trace => self => core.flatMap(doneCollect(self), core.write).traced(trace));
const emitCollect = self => core.flatMap(doneCollect(self), core.write);
/** @internal */
exports.emitCollect = emitCollect;
const ensuring = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, finalizer) => core.ensuringWith(self, () => finalizer).traced(trace));
const ensuring = /*#__PURE__*/(0, _Function.dual)(2, (self, finalizer) => core.ensuringWith(self, () => finalizer));
/** @internal */
exports.ensuring = ensuring;
const context = /*#__PURE__*/Debug.methodWithTrace(trace => () => core.fromEffect(Effect.context()).traced(trace));
const context = () => core.fromEffect(Effect.context());
/** @internal */
exports.context = context;
const contextWith = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => f => map(context(), restore(f)).traced(trace));
const contextWith = f => map(context(), f);
/** @internal */
exports.contextWith = contextWith;
const contextWithChannel = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => f => core.flatMap(context(), restore(f)).traced(trace));
const contextWithChannel = f => core.flatMap(context(), f);
/** @internal */
exports.contextWithChannel = contextWithChannel;
const contextWithEffect = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => f => mapEffect(context(), restore(f)).traced(trace));
const contextWithEffect = f => mapEffect(context(), f);
/** @internal */
exports.contextWithEffect = contextWithEffect;
const flatten = /*#__PURE__*/Debug.methodWithTrace(trace => self => core.flatMap(self, _Function.identity).traced(trace));
const flatten = self => core.flatMap(self, _Function.identity);
/** @internal */
exports.flatten = flatten;
const foldChannel = /*#__PURE__*/Debug.dualWithTrace(3, (trace, restore) => (self, onError, onSuccess) => core.foldCauseChannel(self, cause => {
const foldChannel = /*#__PURE__*/(0, _Function.dual)(3, (self, onError, onSuccess) => core.foldCauseChannel(self, cause => {
const either = Cause.failureOrCause(cause);

@@ -151,3 +150,3 @@ switch (either._tag) {

{
return restore(onError)(either.left);
return onError(either.left);
}

@@ -159,21 +158,21 @@ case "Right":

}
}, restore(onSuccess)).traced(trace));
}, onSuccess));
/** @internal */
exports.foldChannel = foldChannel;
const fromEither = /*#__PURE__*/Debug.methodWithTrace(trace => either => core.suspend(() => Either.match(either, core.fail, core.succeed)).traced(trace));
const fromEither = either => core.suspend(() => Either.match(either, core.fail, core.succeed));
/** @internal */
exports.fromEither = fromEither;
const fromInput = /*#__PURE__*/Debug.methodWithTrace(trace => input => unwrap(input.takeWith(core.failCause, elem => core.flatMap(core.write(elem), () => fromInput(input)), core.succeed)).traced(trace));
const fromInput = input => unwrap(input.takeWith(core.failCause, elem => core.flatMap(core.write(elem), () => fromInput(input)), core.succeed));
/** @internal */
exports.fromInput = fromInput;
const fromHub = /*#__PURE__*/Debug.methodWithTrace(trace => hub => unwrapScoped(Effect.map(Hub.subscribe(hub), fromQueue)).traced(trace));
const fromHub = hub => unwrapScoped(Effect.map(Hub.subscribe(hub), fromQueue));
/** @internal */
exports.fromHub = fromHub;
const fromHubScoped = /*#__PURE__*/Debug.methodWithTrace(trace => hub => Effect.map(Hub.subscribe(hub), fromQueue).traced(trace));
const fromHubScoped = hub => Effect.map(Hub.subscribe(hub), fromQueue);
/** @internal */
exports.fromHubScoped = fromHubScoped;
const fromOption = /*#__PURE__*/Debug.methodWithTrace(trace => option => core.suspend(() => Option.match(option, () => core.fail(Option.none()), core.succeed)).traced(trace));
const fromOption = option => core.suspend(() => Option.match(option, () => core.fail(Option.none()), core.succeed));
/** @internal */
exports.fromOption = fromOption;
const fromQueue = /*#__PURE__*/Debug.methodWithTrace(trace => queue => core.suspend(() => fromQueueInternal(queue)).traced(trace));
const fromQueue = queue => core.suspend(() => fromQueueInternal(queue));
/** @internal */

@@ -183,36 +182,36 @@ exports.fromQueue = fromQueue;

/** @internal */
const identityChannel = /*#__PURE__*/Debug.methodWithTrace(trace => () => core.readWith(input => core.flatMap(core.write(input), () => identityChannel()), core.fail, core.succeedNow).traced(trace));
const identityChannel = () => core.readWith(input => core.flatMap(core.write(input), () => identityChannel()), core.fail, core.succeedNow);
/** @internal */
exports.identityChannel = identityChannel;
const interruptWhen = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, effect) => mergeWith(self, core.fromEffect(effect), selfDone => mergeDecision.Done(Effect.done(selfDone)), effectDone => mergeDecision.Done(Effect.done(effectDone))).traced(trace));
const interruptWhen = /*#__PURE__*/(0, _Function.dual)(2, (self, effect) => mergeWith(self, core.fromEffect(effect), selfDone => mergeDecision.Done(Effect.done(selfDone)), effectDone => mergeDecision.Done(Effect.done(effectDone))));
/** @internal */
exports.interruptWhen = interruptWhen;
const interruptWhenDeferred = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, deferred) => interruptWhen(self, Deferred.await(deferred)).traced(trace));
const interruptWhenDeferred = /*#__PURE__*/(0, _Function.dual)(2, (self, deferred) => interruptWhen(self, Deferred.await(deferred)));
/** @internal */
exports.interruptWhenDeferred = interruptWhenDeferred;
const map = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => core.flatMap(self, a => core.sync(() => restore(f)(a))).traced(trace));
const map = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => core.flatMap(self, a => core.sync(() => f(a))));
/** @internal */
exports.map = map;
const mapEffect = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => core.flatMap(self, z => core.fromEffect(restore(f)(z))).traced(trace));
const mapEffect = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => core.flatMap(self, z => core.fromEffect(f(z))));
/** @internal */
exports.mapEffect = mapEffect;
const mapError = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => mapErrorCause(self, Cause.map(restore(f))).traced(trace));
const mapError = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => mapErrorCause(self, Cause.map(f)));
/** @internal */
exports.mapError = mapError;
const mapErrorCause = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => core.catchAllCause(self, cause => core.failCause(restore(f)(cause))).traced(trace));
const mapErrorCause = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => core.catchAllCause(self, cause => core.failCause(f(cause))));
/** @internal */
exports.mapErrorCause = mapErrorCause;
const mapOut = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => {
const reader = core.readWith(outElem => core.flatMap(core.write(restore(f)(outElem)), () => reader), core.fail, core.succeedNow);
return core.pipeTo(self, reader).traced(trace);
const mapOut = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => {
const reader = core.readWith(outElem => core.flatMap(core.write(f(outElem)), () => reader), core.fail, core.succeedNow);
return core.pipeTo(self, reader);
});
/** @internal */
exports.mapOut = mapOut;
const mapOutEffect = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => {
const reader = core.readWith(outElem => core.flatMap(() => reader)(core.flatMap(core.write)(core.fromEffect(restore(f)(outElem)))), core.fail, core.succeedNow);
return core.pipeTo(self, reader).traced(trace);
const mapOutEffect = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => {
const reader = core.readWithCause(outElem => core.flatMap(() => reader)(core.flatMap(core.write)(core.fromEffect(f(outElem)))), core.failCause, core.succeedNow);
return core.pipeTo(self, reader);
});
/** @internal */
exports.mapOutEffect = mapOutEffect;
const mapOutEffectPar = /*#__PURE__*/Debug.dualWithTrace(3, (trace, restoreTrace) => (self, f, n) => unwrapScoped(Effect.map(queue => {
const mapOutEffectPar = /*#__PURE__*/(0, _Function.dual)(3, (self, f, n) => unwrapScoped(Effect.map(queue => {
const consumer = unwrap(Effect.matchCause(Effect.flatten(Queue.take(queue)), core.failCause, Either.match(core.succeedNow, outElem => core.flatMap(core.write(outElem), () => consumer))));

@@ -232,21 +231,21 @@ return consumer;

yield* $(Effect.asUnit(Queue.offer(queue, Effect.map(Deferred.await(deferred), Either.right))));
yield* $(Effect.forkScoped(withPermits(1)(Effect.zipRight(Effect.intoDeferred(deferred)(Effect.tapErrorCause(cause => Deferred.failCause(errorSignal, cause))(Effect.uninterruptibleMask(restore => Effect.flatMap(Effect.done)(Effect.raceFirst(Effect.exit(restore(restoreTrace(f)(outElem))))(Effect.exit(restore(Deferred.await(errorSignal)))))))))(Deferred.succeed(latch, void 0)))));
yield* $(Effect.forkScoped(withPermits(1)(Effect.zipRight(Effect.intoDeferred(deferred)(Effect.tapErrorCause(cause => Deferred.failCause(errorSignal, cause))(Effect.uninterruptibleMask(restore => Effect.flatMap(Effect.done)(Effect.raceFirst(Effect.exit(restore(f(outElem))))(Effect.exit(restore(Deferred.await(errorSignal)))))))))(Deferred.succeed(latch, void 0)))));
yield* $(Deferred.await(latch));
})))))));
return queue;
}))).traced(trace));
}))));
/** @internal */
exports.mapOutEffectPar = mapOutEffectPar;
const mergeAll = /*#__PURE__*/Debug.methodWithTrace(trace => (n, bufferSize = 16, mergeStrategy = _mergeStrategy.BackPressure()) => {
return channels => mergeAllWith(n, bufferSize, mergeStrategy)(channels, _Function.constVoid).traced(trace);
});
const mergeAll = (n, bufferSize = 16, mergeStrategy = _mergeStrategy.BackPressure()) => {
return channels => mergeAllWith(n, bufferSize, mergeStrategy)(channels, _Function.constVoid);
};
/** @internal */
exports.mergeAll = mergeAll;
const mergeAllUnbounded = /*#__PURE__*/Debug.methodWithTrace(trace => channels => mergeAllWith(Number.POSITIVE_INFINITY)(channels, _Function.constVoid).traced(trace));
const mergeAllUnbounded = channels => mergeAllWith(Number.POSITIVE_INFINITY)(channels, _Function.constVoid);
/** @internal */
exports.mergeAllUnbounded = mergeAllUnbounded;
const mergeAllUnboundedWith = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => (channels, f) => mergeAllWith(Number.POSITIVE_INFINITY)(channels, restore(f)).traced(trace));
const mergeAllUnboundedWith = (channels, f) => mergeAllWith(Number.POSITIVE_INFINITY)(channels, f);
/** @internal */
exports.mergeAllUnboundedWith = mergeAllUnboundedWith;
const mergeAllWith = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => (n, bufferSize = 16, mergeStrategy = _mergeStrategy.BackPressure()) => (channels, f) => unwrapScoped(Effect.map(([queue, input]) => {
const mergeAllWith = (n, bufferSize = 16, mergeStrategy = _mergeStrategy.BackPressure()) => (channels, f) => unwrapScoped(Effect.map(([queue, input]) => {
const consumer = unwrap(Effect.matchCause(core.failCause, Either.match(core.succeedNow, outElem => core.flatMap(core.write(outElem), () => consumer)))(Effect.flatten(Queue.take(queue))));

@@ -264,3 +263,3 @@ return core.embedInput(consumer, input);

const evaluatePull = pull => Effect.catchAllCause(cause => Cause.isInterrupted(cause) ? Effect.failCause(cause) : Effect.asUnit(Effect.zipRight(Deferred.succeed(errorSignal, void 0))(Queue.offer(queue, Effect.failCause(cause)))))(Effect.flatMap(Option.match(() => Effect.unit(), outDone => Ref.update(lastDone, Option.match(() => Option.some(outDone), lastDone => Option.some(f(lastDone, outDone))))))(Effect.repeatUntil(Option.isSome)(Effect.flatMap(Either.match(done => Effect.succeed(Option.some(done)), outElem => Effect.as(Option.none())(Queue.offer(queue, Effect.succeed(Either.right(outElem))))))(pull))));
yield* $(Effect.forkScoped(Effect.repeatWhileEquals(true)(Effect.matchCauseEffect(pull, cause => Effect.zipRight(Effect.succeed(false))(Queue.offer(queue, Effect.failCause(cause))), Either.match(outDone => Effect.raceWith(withPermits(n)(Effect.unit()), (_, permitAcquisition) => Effect.as(false)(Fiber.interrupt(permitAcquisition)), (_, failureAwait) => Effect.zipRight(Effect.as(false)(Effect.flatMap(Option.match(() => Queue.offer(queue, Effect.succeed(Either.left(outDone))), lastDone => Queue.offer(queue, Effect.succeed(Either.left(restore(f)(lastDone, outDone))))))(Ref.get(lastDone))))(Fiber.interrupt(failureAwait)))(Deferred.await(errorSignal)), channel => _mergeStrategy.match(() => Effect.gen(function* ($) {
yield* $(Effect.forkScoped(Effect.repeatWhileEquals(true)(Effect.matchCauseEffect(pull, cause => Effect.zipRight(Effect.succeed(false))(Queue.offer(queue, Effect.failCause(cause))), Either.match(outDone => Effect.raceWith(withPermits(n)(Effect.unit()), (_, permitAcquisition) => Effect.as(false)(Fiber.interrupt(permitAcquisition)), (_, failureAwait) => Effect.zipRight(Effect.as(false)(Effect.flatMap(Option.match(() => Queue.offer(queue, Effect.succeed(Either.left(outDone))), lastDone => Queue.offer(queue, Effect.succeed(Either.left(f(lastDone, outDone))))))(Ref.get(lastDone))))(Fiber.interrupt(failureAwait)))(Deferred.await(errorSignal)), channel => _mergeStrategy.match(() => Effect.gen(function* ($) {
const latch = yield* $(Deferred.make());

@@ -285,24 +284,24 @@ const raceEffects = Effect.scoped(Effect.flatMap(pull => Effect.raceAwait(Deferred.await(errorSignal))(evaluatePull(pull)))(toPull(core.pipeTo(channel)(queueReader))));

return [queue, input];
}))).traced(trace));
})));
/** @internal */
exports.mergeAllWith = mergeAllWith;
const mergeMap = /*#__PURE__*/Debug.dualWithTrace(3, (trace, restore) => (self, f, n) => mergeMapBufferStrategy(self, restore(f), n, 16, _mergeStrategy.BackPressure()).traced(trace));
const mergeMap = /*#__PURE__*/(0, _Function.dual)(3, (self, f, n) => mergeMapBufferStrategy(self, f, n, 16, _mergeStrategy.BackPressure()));
/** @internal */
exports.mergeMap = mergeMap;
const mergeMapBuffer = /*#__PURE__*/Debug.dualWithTrace(4, (trace, restore) => (self, f, n, bufferSize) => mergeMapBufferStrategy(self, restore(f), n, bufferSize, _mergeStrategy.BackPressure()).traced(trace));
const mergeMapBuffer = /*#__PURE__*/(0, _Function.dual)(4, (self, f, n, bufferSize) => mergeMapBufferStrategy(self, f, n, bufferSize, _mergeStrategy.BackPressure()));
/** @internal */
exports.mergeMapBuffer = mergeMapBuffer;
const mergeMapStrategy = /*#__PURE__*/Debug.dualWithTrace(4, (trace, restore) => (self, f, n, mergeStrategy) => mergeMapBufferStrategy(self, restore(f), n, 16, mergeStrategy).traced(trace));
const mergeMapStrategy = /*#__PURE__*/(0, _Function.dual)(4, (self, f, n, mergeStrategy) => mergeMapBufferStrategy(self, f, n, 16, mergeStrategy));
/** @internal */
exports.mergeMapStrategy = mergeMapStrategy;
const mergeMapBufferStrategy = /*#__PURE__*/Debug.dualWithTrace(5, (trace, restore) => (self, f, n, bufferSize, mergeStrategy) => mergeAll(n, bufferSize, mergeStrategy)(mapOut(self, restore(f))).traced(trace));
const mergeMapBufferStrategy = /*#__PURE__*/(0, _Function.dual)(5, (self, f, n, bufferSize, mergeStrategy) => mergeAll(n, bufferSize, mergeStrategy)(mapOut(self, f)));
/** @internal */
exports.mergeMapBufferStrategy = mergeMapBufferStrategy;
const mergeOut = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, n) => mergeAll(n)(mapOut(self, _Function.identity)).traced(trace));
const mergeOut = /*#__PURE__*/(0, _Function.dual)(2, (self, n) => mergeAll(n)(mapOut(self, _Function.identity)));
/** @internal */
exports.mergeOut = mergeOut;
const mergeOutWith = /*#__PURE__*/Debug.dualWithTrace(3, (trace, restore) => (self, n, f) => mergeAllWith(n)(mapOut(self, _Function.identity), restore(f)).traced(trace));
const mergeOutWith = /*#__PURE__*/(0, _Function.dual)(3, (self, n, f) => mergeAllWith(n)(mapOut(self, _Function.identity), f));
/** @internal */
exports.mergeOutWith = mergeOutWith;
const mergeWith = /*#__PURE__*/Debug.dualWithTrace(4, (trace, restore) => (self, that, leftDone, rightDone) => unwrapScoped(Effect.flatMap(input => {
const mergeWith = /*#__PURE__*/(0, _Function.dual)(4, (self, that, leftDone, rightDone) => unwrapScoped(Effect.flatMap(input => {
const queueReader = fromInput(input);

@@ -326,3 +325,3 @@ return Effect.map(Effect.zip(toPull(core.pipeTo(queueReader, self)), toPull(core.pipeTo(queueReader, that))), ([pullL, pullR]) => {

const rightJoin = Effect.interruptible(Fiber.join(state.right));
return unwrap(Effect.raceWith(leftJoin, rightJoin, (leftExit, rf) => Effect.zipRight(Fiber.interrupt(rf), handleSide(leftExit, state.right, pullL)(restore(leftDone), mergeState.BothRunning, f => mergeState.LeftDone(f))), (rightExit, lf) => Effect.zipRight(Fiber.interrupt(lf), handleSide(rightExit, state.left, pullR)(restore(rightDone), (left, right) => mergeState.BothRunning(right, left), f => mergeState.RightDone(f)))));
return unwrap(Effect.raceWith(leftJoin, rightJoin, (leftExit, rf) => Effect.zipRight(Fiber.interrupt(rf), handleSide(leftExit, state.right, pullL)(leftDone, mergeState.BothRunning, f => mergeState.LeftDone(f))), (rightExit, lf) => Effect.zipRight(Fiber.interrupt(lf), handleSide(rightExit, state.left, pullR)(rightDone, (left, right) => mergeState.BothRunning(right, left), f => mergeState.RightDone(f)))));
}

@@ -341,20 +340,20 @@ case MergeStateOpCodes.OP_LEFT_DONE:

});
})(singleProducerAsyncInput.make())).traced(trace));
})(singleProducerAsyncInput.make())));
/** @internal */
exports.mergeWith = mergeWith;
const never = /*#__PURE__*/Debug.methodWithTrace(trace => () => core.fromEffect(Effect.never()).traced(trace));
const never = () => core.fromEffect(Effect.never());
/** @internal */
exports.never = never;
const orDie = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, error) => orDieWith(self, restore(error)).traced(trace));
const orDie = /*#__PURE__*/(0, _Function.dual)(2, (self, error) => orDieWith(self, error));
/** @internal */
exports.orDie = orDie;
const orDieWith = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => catchAll(self, e => {
throw restore(f)(e);
}).traced(trace));
const orDieWith = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => catchAll(self, e => {
throw f(e);
}));
/** @internal */
exports.orDieWith = orDieWith;
const orElse = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, that) => catchAll(self, that).traced(trace));
const orElse = /*#__PURE__*/(0, _Function.dual)(2, (self, that) => catchAll(self, that));
/** @internal */
exports.orElse = orElse;
const pipeToOrFail = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, that) => core.suspend(() => {
const pipeToOrFail = /*#__PURE__*/(0, _Function.dual)(2, (self, that) => core.suspend(() => {
let channelException = undefined;

@@ -369,59 +368,59 @@ const reader = core.readWith(outElem => core.flatMap(core.write(outElem), () => reader), outErr => {

}, core.succeedNow);
return core.pipeTo(core.pipeTo(core.pipeTo(self, reader), that), writer).traced(trace);
return core.pipeTo(core.pipeTo(core.pipeTo(self, reader), that), writer);
}));
/** @internal */
exports.pipeToOrFail = pipeToOrFail;
const provideService = /*#__PURE__*/Debug.dualWithTrace(3, trace => (self, tag, service) => {
return core.flatMap(context(), context => core.provideContext(self, Context.add(context, tag, service))).traced(trace);
const provideService = /*#__PURE__*/(0, _Function.dual)(3, (self, tag, service) => {
return core.flatMap(context(), context => core.provideContext(self, Context.add(context, tag, service)));
});
/** @internal */
exports.provideService = provideService;
const provideLayer = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, layer) => unwrapScoped(Effect.map(Layer.build(layer), env => core.provideContext(self, env))).traced(trace));
const provideLayer = /*#__PURE__*/(0, _Function.dual)(2, (self, layer) => unwrapScoped(Effect.map(Layer.build(layer), env => core.provideContext(self, env))));
/** @internal */
exports.provideLayer = provideLayer;
const contramapContext = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => contextWithChannel(context => core.provideContext(self, restore(f)(context))).traced(trace));
const contramapContext = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => contextWithChannel(context => core.provideContext(self, f(context))));
/** @internal */
exports.contramapContext = contramapContext;
const provideSomeLayer = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, layer) =>
const provideSomeLayer = /*#__PURE__*/(0, _Function.dual)(2, (self, layer) =>
// @ts-expect-error
provideLayer(self, Layer.merge(Layer.context(), layer)).traced(trace));
provideLayer(self, Layer.merge(Layer.context(), layer)));
/** @internal */
exports.provideSomeLayer = provideSomeLayer;
const read = /*#__PURE__*/Debug.methodWithTrace(trace => () => core.readOrFail(Option.none()).traced(trace));
const read = () => core.readOrFail(Option.none());
/** @internal */
exports.read = read;
const repeated = /*#__PURE__*/Debug.methodWithTrace(trace => self => core.flatMap(self, () => repeated(self)).traced(trace));
const repeated = self => core.flatMap(self, () => repeated(self));
/** @internal */
exports.repeated = repeated;
const run = /*#__PURE__*/Debug.methodWithTrace(trace => self => Effect.scoped(executor.runScoped(self)).traced(trace));
const run = self => Effect.scoped(executor.runScoped(self));
/** @internal */
exports.run = run;
const runCollect = /*#__PURE__*/Debug.methodWithTrace(trace => self => executor.run(core.collectElements(self)).traced(trace));
const runCollect = self => executor.run(core.collectElements(self));
/** @internal */
exports.runCollect = runCollect;
const runDrain = /*#__PURE__*/Debug.methodWithTrace(trace => self => executor.run(drain(self)).traced(trace));
const runDrain = self => executor.run(drain(self));
/** @internal */
exports.runDrain = runDrain;
const scoped = /*#__PURE__*/Debug.methodWithTrace(trace => effect => unwrap(Effect.uninterruptibleMask(restore => Effect.map(Scope.make(), scope => core.acquireReleaseOut(Effect.tapErrorCause(restore(Scope.extend(scope)(effect)), cause => Scope.close(scope, Exit.failCause(cause))), (_, exit) => Scope.close(scope, exit))))).traced(trace));
const scoped = effect => unwrap(Effect.uninterruptibleMask(restore => Effect.map(Scope.make(), scope => core.acquireReleaseOut(Effect.tapErrorCause(restore(Scope.extend(scope)(effect)), cause => Scope.close(scope, Exit.failCause(cause))), (_, exit) => Scope.close(scope, exit)))));
/** @internal */
exports.scoped = scoped;
const service = /*#__PURE__*/Debug.methodWithTrace(trace => tag => core.fromEffect(tag).traced(trace));
const service = tag => core.fromEffect(tag);
/** @internal */
exports.service = service;
const serviceWith = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => tag => f => map(service(tag), restore(f)).traced(trace));
const serviceWith = tag => f => map(service(tag), f);
/** @internal */
exports.serviceWith = serviceWith;
const serviceWithChannel = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => tag => f => core.flatMap(service(tag), restore(f)).traced(trace));
const serviceWithChannel = tag => f => core.flatMap(service(tag), f);
/** @internal */
exports.serviceWithChannel = serviceWithChannel;
const serviceWithEffect = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => tag => f => mapEffect(service(tag), restore(f)).traced(trace));
const serviceWithEffect = tag => f => mapEffect(service(tag), f);
/** @internal */
exports.serviceWithEffect = serviceWithEffect;
const toHub = /*#__PURE__*/Debug.methodWithTrace(trace => hub => toQueue(hub).traced(trace));
const toHub = hub => toQueue(hub);
/** @internal */
exports.toHub = toHub;
const toPull = /*#__PURE__*/Debug.methodWithTrace(trace => self => Effect.map(Effect.acquireRelease(Effect.sync(() => new executor.ChannelExecutor(self, void 0, _Function.identity)), (exec, exit) => {
const toPull = self => Effect.map(Effect.acquireRelease(Effect.sync(() => new executor.ChannelExecutor(self, void 0, _Function.identity)), (exec, exit) => {
const finalize = exec.close(exit);
return finalize === undefined ? Effect.unit() : finalize;
}), exec => Effect.suspend(() => interpretToPull(exec.run(), exec))).traced(trace));
}), exec => Effect.suspend(() => interpretToPull(exec.run(), exec)));
/** @internal */

@@ -451,3 +450,3 @@ exports.toPull = toPull;

/** @internal */
const toQueue = /*#__PURE__*/Debug.methodWithTrace(trace => queue => core.suspend(() => toQueueInternal(queue)).traced(trace));
const toQueue = queue => core.suspend(() => toQueueInternal(queue));
/** @internal */

@@ -459,15 +458,15 @@ exports.toQueue = toQueue;

/** @internal */
const unwrap = /*#__PURE__*/Debug.methodWithTrace(trace => channel => flatten(core.fromEffect(channel)).traced(trace));
const unwrap = channel => flatten(core.fromEffect(channel));
/** @internal */
exports.unwrap = unwrap;
const unwrapScoped = /*#__PURE__*/Debug.methodWithTrace(trace => self => core.concatAllWith(scoped(self), (d, _) => d, (d, _) => d).traced(trace));
const unwrapScoped = self => core.concatAllWith(scoped(self), (d, _) => d, (d, _) => d);
/** @internal */
exports.unwrapScoped = unwrapScoped;
const updateService = /*#__PURE__*/Debug.dualWithTrace(3, (trace, restore) => (self, tag, f) => contramapContext(self, context => Context.merge(context, Context.make(tag, restore(f)(Context.unsafeGet(context, tag))))).traced(trace));
const updateService = /*#__PURE__*/(0, _Function.dual)(3, (self, tag, f) => contramapContext(self, context => Context.merge(context, Context.make(tag, f(Context.unsafeGet(context, tag))))));
/** @internal */
exports.updateService = updateService;
const writeAll = /*#__PURE__*/Debug.methodWithTrace(trace => (...outs) => writeChunk(Chunk.fromIterable(outs)).traced(trace));
const writeAll = (...outs) => writeChunk(Chunk.fromIterable(outs));
/** @internal */
exports.writeAll = writeAll;
const writeChunk = /*#__PURE__*/Debug.methodWithTrace(trace => outs => writeChunkWriter(0, outs.length, outs).traced(trace));
const writeChunk = outs => writeChunkWriter(0, outs.length, outs);
/** @internal */

@@ -479,18 +478,18 @@ exports.writeChunk = writeChunk;

/** @internal */
const zip = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, that) => core.flatMap(self, a => map(b => [a, b])(that)).traced(trace));
const zip = /*#__PURE__*/(0, _Function.dual)(2, (self, that) => core.flatMap(self, a => map(that, b => [a, b])));
/** @internal */
exports.zip = zip;
const zipLeft = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, that) => core.flatMap(self, z => as(that, z)).traced(trace));
const zipLeft = /*#__PURE__*/(0, _Function.dual)(2, (self, that) => core.flatMap(self, z => as(that, z)));
/** @internal */
exports.zipLeft = zipLeft;
const zipRight = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, that) => core.flatMap(self, () => that).traced(trace));
const zipRight = /*#__PURE__*/(0, _Function.dual)(2, (self, that) => core.flatMap(self, () => that));
/** @internal */
exports.zipRight = zipRight;
const zipPar = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, that) => mergeWith(self, that, exit1 => mergeDecision.Await(exit2 => Effect.done(Exit.zip(exit1, exit2))), exit2 => mergeDecision.Await(exit1 => Effect.done(Exit.zip(exit1, exit2)))).traced(trace));
const zipPar = /*#__PURE__*/(0, _Function.dual)(2, (self, that) => mergeWith(self, that, exit1 => mergeDecision.Await(exit2 => Effect.done(Exit.zip(exit1, exit2))), exit2 => mergeDecision.Await(exit1 => Effect.done(Exit.zip(exit1, exit2)))));
/** @internal */
exports.zipPar = zipPar;
const zipParLeft = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, that) => map(zipPar(self, that), tuple => tuple[0]).traced(trace));
const zipParLeft = /*#__PURE__*/(0, _Function.dual)(2, (self, that) => map(zipPar(self, that), tuple => tuple[0]));
/** @internal */
exports.zipParLeft = zipParLeft;
const zipParRight = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, that) => map(zipPar(self, that), tuple => tuple[1]).traced(trace));
const zipParRight = /*#__PURE__*/(0, _Function.dual)(2, (self, that) => map(zipPar(self, that), tuple => tuple[1]));
/** @internal */

@@ -497,0 +496,0 @@ exports.zipParRight = zipParRight;

5

internal/channel/channelExecutor.d.ts

@@ -1,4 +0,2 @@

import * as Chunk from "@effect/data/Chunk";
import type * as Context from "@effect/data/Context";
import * as Debug from "@effect/data/Debug";
import * as Option from "@effect/data/Option";

@@ -24,3 +22,2 @@ import * as Cause from "@effect/io/Cause";

private _emitted;
private _traceStack;
private _executeCloseLastSubstream;

@@ -32,3 +29,2 @@ private _input;

run(): ChannelState.ChannelState<Env, unknown>;
stackToLines(): Chunk.Chunk<Debug.SourceLocation>;
getDone(): Exit.Exit<OutErr, OutDone>;

@@ -44,3 +40,2 @@ getEmit(): OutElem;

doneSucceed(value: unknown): ChannelState.ChannelState<Env, unknown> | undefined;
annotate<E>(_cause: Cause.Cause<E>): Cause.Cause<E>;
doneHalt(cause: Cause.Cause<unknown>): ChannelState.ChannelState<Env, unknown> | undefined;

@@ -47,0 +42,0 @@ processCancellation(): ChannelState.ChannelState<Env, unknown>;

155

internal/channel/channelExecutor.js

@@ -7,6 +7,3 @@ "use strict";

exports.runScoped = exports.run = exports.readUpstream = exports.ChannelExecutor = void 0;
var Chunk = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/data/Chunk"));
var Debug = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/data/Debug"));
var _Function = /*#__PURE__*/require("@effect/data/Function");
var MRef = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/data/MutableRef"));
var Option = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/data/Option"));

@@ -40,3 +37,2 @@ var Cause = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/io/Cause"));

this._emitted = undefined;
this._traceStack = [];
this._input = undefined;

@@ -182,3 +178,7 @@ this._inProgressFinalizer = undefined;

result = ChannelState.Read(this._input, _Function.identity, emitted => {
this._currentChannel = read.more(emitted);
try {
this._currentChannel = read.more(emitted);
} catch (error) {
this._currentChannel = read.done.onExit(Exit.die(error));
}
return undefined;

@@ -209,11 +209,2 @@ }, exit => {

}
case ChannelOpCodes.OP_TRACED:
{
this._traceStack.push(this._currentChannel.trace);
this.addFinalizer(() => Effect.sync(() => {
this._traceStack.pop();
}));
this._currentChannel = this._currentChannel.channel;
break;
}
default:

@@ -234,15 +225,2 @@ {

}
stackToLines() {
if (this._traceStack.length === 0) {
return Chunk.empty();
}
const lines = [];
let current = this._traceStack.length - 1;
while (current >= 0 && lines.length < Debug.runtimeDebug.traceStackLimit) {
const value = this._traceStack[current];
lines.push(value);
current = current - 1;
}
return Chunk.unsafeFromArray(lines);
}
getDone() {

@@ -264,15 +242,13 @@ return this._done;

popAllFinalizers(exit) {
return Debug.untraced(() => {
const finalizers = [];
let next = this._doneStack.pop();
while (next) {
if (next._tag === "ContinuationFinalizer") {
finalizers.push(next.finalizer);
}
next = this._doneStack.pop();
const finalizers = [];
let next = this._doneStack.pop();
while (next) {
if (next._tag === "ContinuationFinalizer") {
finalizers.push(next.finalizer);
}
const effect = finalizers.length === 0 ? Effect.unit() : runFinalizers(finalizers, exit);
this.storeInProgressFinalizer(effect);
return effect;
});
next = this._doneStack.pop();
}
const effect = finalizers.length === 0 ? Effect.unit() : runFinalizers(finalizers, exit);
this.storeInProgressFinalizer(effect);
return effect;
}

@@ -292,30 +268,26 @@ popNextFinalizers() {

restorePipe(exit, prev) {
return Debug.untraced(() => {
const currInput = this._input;
this._input = prev;
if (currInput !== undefined) {
const effect = currInput.close(exit);
return effect;
}
return Effect.unit();
});
const currInput = this._input;
this._input = prev;
if (currInput !== undefined) {
const effect = currInput.close(exit);
return effect;
}
return Effect.unit();
}
close(exit) {
return Debug.untraced(() => {
let runInProgressFinalizers = undefined;
const finalizer = this._inProgressFinalizer;
if (finalizer !== undefined) {
runInProgressFinalizers = Effect.ensuring(Effect.sync(() => this.clearInProgressFinalizer()))(finalizer);
}
let closeSelf = undefined;
const selfFinalizers = this.popAllFinalizers(exit);
if (selfFinalizers !== undefined) {
closeSelf = Effect.ensuring(Effect.sync(() => this.clearInProgressFinalizer()))(selfFinalizers);
}
const closeSubexecutors = this._activeSubexecutor === undefined ? undefined : this._activeSubexecutor.close(exit);
if (closeSubexecutors === undefined && runInProgressFinalizers === undefined && closeSelf === undefined) {
return undefined;
}
return Effect.flatMap(Effect.done)(Effect.uninterruptible(Effect.map(([[exit1, exit2], exit3]) => Exit.zipRight(exit3)(Exit.zipRight(exit2)(exit1)))(Effect.zip(Effect.exit(ifNotNull(closeSelf)))(Effect.zip(Effect.exit(ifNotNull(runInProgressFinalizers)))(Effect.exit(ifNotNull(closeSubexecutors)))))));
});
let runInProgressFinalizers = undefined;
const finalizer = this._inProgressFinalizer;
if (finalizer !== undefined) {
runInProgressFinalizers = Effect.ensuring(Effect.sync(() => this.clearInProgressFinalizer()))(finalizer);
}
let closeSelf = undefined;
const selfFinalizers = this.popAllFinalizers(exit);
if (selfFinalizers !== undefined) {
closeSelf = Effect.ensuring(Effect.sync(() => this.clearInProgressFinalizer()))(selfFinalizers);
}
const closeSubexecutors = this._activeSubexecutor === undefined ? undefined : this._activeSubexecutor.close(exit);
if (closeSubexecutors === undefined && runInProgressFinalizers === undefined && closeSelf === undefined) {
return undefined;
}
return Effect.flatMap(Effect.done)(Effect.uninterruptible(Effect.map(([[exit1, exit2], exit3]) => Exit.zipRight(exit3)(Exit.zipRight(exit2)(exit1)))(Effect.zip(Effect.exit(ifNotNull(closeSelf)))(Effect.zip(Effect.exit(ifNotNull(runInProgressFinalizers)))(Effect.exit(ifNotNull(closeSubexecutors)))))));
}

@@ -346,13 +318,2 @@ doneSucceed(value) {

}
annotate(_cause) {
let cause = _cause;
if (Cause.isAnnotatedType(cause) && Cause.isStackAnnotation(cause.annotation)) {
const stack = cause.annotation.stack;
const currentStack = this.stackToLines();
cause = Cause.annotated(cause.cause, new Cause.StackAnnotation(Chunk.take(Debug.runtimeDebug.traceStackLimit)(Chunk.dedupeAdjacent(stack.length === 0 ? currentStack : currentStack.length === 0 ? stack : Chunk.unsafeLast(stack) === Chunk.unsafeLast(currentStack) ? stack : Chunk.concat(currentStack)(stack))), cause.annotation.seq));
} else {
cause = Cause.annotated(cause, new Cause.StackAnnotation(this.stackToLines(), MRef.getAndIncrement(Cause.globalErrorSeq)));
}
return cause;
}
doneHalt(cause) {

@@ -438,7 +399,5 @@ if (this._doneStack.length === 0) {

finishWithExit(exit) {
return Debug.untraced(() => {
const state = Exit.match(cause => this.doneHalt(cause), value => this.doneSucceed(value))(exit);
this._activeSubexecutor = undefined;
return state === undefined ? Effect.unit() : ChannelState.effect(state);
});
const state = Exit.match(cause => this.doneHalt(cause), value => this.doneSucceed(value))(exit);
this._activeSubexecutor = undefined;
return state === undefined ? Effect.unit() : ChannelState.effect(state);
}

@@ -604,10 +563,10 @@ finishSubexecutorWithCloseEffect(subexecutorDone, ...closeFuncs) {

exports.ChannelExecutor = ChannelExecutor;
const ifNotNull = /*#__PURE__*/Debug.untracedMethod(() => effect => effect !== undefined ? effect : Effect.unit());
const runFinalizers = /*#__PURE__*/Debug.untracedMethod(() => (finalizers, exit) => {
const ifNotNull = effect => effect !== undefined ? effect : Effect.unit();
const runFinalizers = (finalizers, exit) => {
return Effect.flatMap(exit => Effect.done(exit))(Effect.map(exits => Option.getOrElse(() => Exit.unit())(Exit.collectAll(exits)))(Effect.forEach(finalizers, fin => Effect.exit(fin(exit)))));
});
};
/**
* @internal
*/
const readUpstream = /*#__PURE__*/Debug.methodWithTrace(trace => (r, onSuccess, onFailure) => {
const readUpstream = (r, onSuccess, onFailure) => {
const readStack = [r];

@@ -628,8 +587,8 @@ const read = () => {

}
return Effect.matchCauseEffect(onFailure, onSuccess)(emitEffect).traced(trace);
return Effect.matchCauseEffect(onFailure, onSuccess)(emitEffect);
}
if (emitEffect === undefined) {
return Effect.suspend(() => read()).traced(trace);
return Effect.suspend(() => read());
}
return Effect.matchCauseEffect(onFailure, () => read())(emitEffect).traced(trace);
return Effect.matchCauseEffect(onFailure, () => read())(emitEffect);
}

@@ -641,10 +600,10 @@ case ChannelStateOpCodes.OP_DONE:

if (doneEffect === undefined) {
return Effect.suspend(onSuccess).traced(trace);
return Effect.suspend(onSuccess);
}
return Effect.matchCauseEffect(onFailure, () => onSuccess())(doneEffect).traced(trace);
return Effect.matchCauseEffect(onFailure, () => onSuccess())(doneEffect);
}
if (doneEffect === undefined) {
return Effect.suspend(() => read()).traced(trace);
return Effect.suspend(() => read());
}
return Effect.matchCauseEffect(onFailure, () => read())(doneEffect).traced(trace);
return Effect.matchCauseEffect(onFailure, () => read())(doneEffect);
}

@@ -657,3 +616,3 @@ case ChannelStateOpCodes.OP_FROM_EFFECT:

return doneEffect === undefined ? Effect.unit() : doneEffect;
}))(current.onEffect(state.effect))).traced(trace);
}))(current.onEffect(state.effect)));
}

@@ -664,3 +623,3 @@ case ChannelStateOpCodes.OP_READ:

readStack.push(state);
return Effect.suspend(() => read()).traced(trace);
return Effect.suspend(() => read());
}

@@ -670,10 +629,10 @@ }

return read();
});
};
/** @internal */
exports.readUpstream = readUpstream;
const run = /*#__PURE__*/Debug.methodWithTrace(trace => self => Effect.scoped(runScoped(self)).traced(trace));
const run = self => Effect.scoped(runScoped(self));
/** @internal */
exports.run = run;
const runScoped = self => {
const run = (deferred, scope) => Effect.acquireUseRelease(Effect.sync(() => new ChannelExecutor(self, void 0, _Function.identity)), exec => Effect.suspend(() => Effect.zipLeft(Effect.never())(Effect.zipRight(Deferred.await(deferred))(Effect.intoDeferred(deferred)(runScopedInterpret(exec.run(), exec))))), (exec, exit) => {
const run = (channelDeferred, scopeDeferred, scope) => Effect.acquireUseRelease(Effect.sync(() => new ChannelExecutor(self, void 0, _Function.identity)), exec => Effect.suspend(() => Effect.zipLeft(Deferred.await(scopeDeferred))(Effect.zipRight(Deferred.await(channelDeferred))(Effect.intoDeferred(channelDeferred)(runScopedInterpret(exec.run(), exec))))), (exec, exit) => {
const finalize = exec.close(exit);

@@ -685,3 +644,3 @@ if (finalize === undefined) {

});
return Effect.flatMap(Effect.scope(), parent => Effect.flatMap(Scope.fork(parent, ExecutionStrategy.sequential), child => Effect.flatMap(Deferred.make(), deferred => Effect.flatMap(Effect.forkScoped(run(deferred, child)), fiber => Effect.zipLeft(Deferred.await(deferred), Fiber.inheritAll(fiber))))));
return Effect.uninterruptibleMask(restore => Effect.flatMap(Effect.scope(), parent => Effect.flatMap(([child, channelDeferred, scopeDeferred]) => Effect.flatMap(fiber => Effect.zipLeft(Fiber.inheritAll(fiber))(Effect.zipRight(restore(Deferred.await(channelDeferred)))(Effect.addFinalizer(() => Deferred.succeed(scopeDeferred, void 0)))))(Effect.forkScoped(restore(run(channelDeferred, scopeDeferred, child)))))(Effect.all(Scope.fork(parent, ExecutionStrategy.sequential), Deferred.make(), Deferred.make()))));
};

@@ -688,0 +647,0 @@ /** @internal */

2

internal/core.d.ts
import type { LazyArg } from "@effect/data/Function";
import type * as Channel from "@effect/stream/Channel";
export type Primitive = BracketOut | Bridge | ConcatAll | Emit | Ensuring | Fail | Fold | FromEffect | PipeTo | Provide | Read | Succeed | SucceedNow | Suspend | Traced;
export type Primitive = BracketOut | Bridge | ConcatAll | Emit | Ensuring | Fail | Fold | FromEffect | PipeTo | Provide | Read | Succeed | SucceedNow | Suspend;
export declare const sync: <OutDone>(evaluate: LazyArg<OutDone>) => Channel.Channel<never, unknown, unknown, unknown, never, never, OutDone>;
//# sourceMappingURL=core.d.ts.map

173

internal/core.js

@@ -8,3 +8,2 @@ "use strict";

var Chunk = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/data/Chunk"));
var Debug = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/data/Debug"));
var Either = /*#__PURE__*/_interopRequireWildcard( /*#__PURE__*/require("@effect/data/Either"));

@@ -37,45 +36,29 @@ var _Function = /*#__PURE__*/require("@effect/data/Function");

const proto = {
[ChannelTypeId]: channelVariance,
traced(trace) {
if (trace) {
return Object.create(proto, {
_tag: {
value: OpCodes.OP_TRACED
},
channel: {
value: this
},
trace: {
value: trace
}
});
}
return this;
}
[ChannelTypeId]: channelVariance
};
/** @internal */
const acquireReleaseOut = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, release) => {
const acquireReleaseOut = /*#__PURE__*/(0, _Function.dual)(2, (self, release) => {
const op = Object.create(proto);
op._tag = OpCodes.OP_BRACKET_OUT;
op.acquire = () => self;
op.finalizer = restore(release);
return op.traced(trace);
op.finalizer = release;
return op;
});
/** @internal */
exports.acquireReleaseOut = acquireReleaseOut;
const catchAllCause = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => {
const catchAllCause = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => {
const op = Object.create(proto);
op._tag = OpCodes.OP_FOLD;
op.channel = self;
op.k = new _continuation.ContinuationKImpl(succeed, restore(f));
return op.traced(trace);
op.k = new _continuation.ContinuationKImpl(succeed, f);
return op;
});
/** @internal */
exports.catchAllCause = catchAllCause;
const collectElements = /*#__PURE__*/Debug.methodWithTrace(trace => self => {
const collectElements = self => {
return suspend(() => {
const builder = [];
return flatMap(pipeTo(self, collectElementsReader(builder)), value => sync(() => [Chunk.fromIterable(builder), value]));
}).traced(trace);
});
});
};
/** @internal */

@@ -89,10 +72,10 @@ exports.collectElements = collectElements;

/** @internal */
const concatAll = /*#__PURE__*/Debug.methodWithTrace(trace => channels => concatAllWith(channels, _Function.constVoid, _Function.constVoid).traced(trace));
const concatAll = channels => concatAllWith(channels, _Function.constVoid, _Function.constVoid);
/** @internal */
exports.concatAll = concatAll;
const concatAllWith = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => (channels, f, g) => {
const concatAllWith = (channels, f, g) => {
const op = Object.create(proto);
op._tag = OpCodes.OP_CONCAT_ALL;
op.combineInners = restore(f);
op.combineAll = restore(g);
op.combineInners = f;
op.combineAll = g;
op.onPull = () => upstreamPullStrategy.PullAfterNext(Option.none());

@@ -102,33 +85,33 @@ op.onEmit = () => childExecutorDecision.Continue;

op.k = _Function.identity;
return op.traced(trace);
});
return op;
};
/** @internal */
exports.concatAllWith = concatAllWith;
const concatMapWith = /*#__PURE__*/Debug.dualWithTrace(4, (trace, restore) => (self, f, g, h) => {
const concatMapWith = /*#__PURE__*/(0, _Function.dual)(4, (self, f, g, h) => {
const op = Object.create(proto);
op._tag = OpCodes.OP_CONCAT_ALL;
op.combineInners = restore(g);
op.combineAll = restore(h);
op.combineInners = g;
op.combineAll = h;
op.onPull = () => upstreamPullStrategy.PullAfterNext(Option.none());
op.onEmit = () => childExecutorDecision.Continue;
op.value = () => self;
op.k = restore(f);
return op.traced(trace);
op.k = f;
return op;
});
/** @internal */
exports.concatMapWith = concatMapWith;
const concatMapWithCustom = /*#__PURE__*/Debug.dualWithTrace(6, (trace, restore) => (self, f, g, h, onPull, onEmit) => {
const concatMapWithCustom = /*#__PURE__*/(0, _Function.dual)(6, (self, f, g, h, onPull, onEmit) => {
const op = Object.create(proto);
op._tag = OpCodes.OP_CONCAT_ALL;
op.combineInners = restore(g);
op.combineAll = restore(h);
op.onPull = restore(onPull);
op.onEmit = restore(onEmit);
op.combineInners = g;
op.combineAll = h;
op.onPull = onPull;
op.onEmit = onEmit;
op.value = () => self;
op.k = restore(f);
return op.traced(trace);
op.k = f;
return op;
});
/** @internal */
exports.concatMapWithCustom = concatMapWithCustom;
const embedInput = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, input) => {
const embedInput = /*#__PURE__*/(0, _Function.dual)(2, (self, input) => {
const op = Object.create(proto);

@@ -138,59 +121,59 @@ op._tag = OpCodes.OP_BRIDGE;

op.channel = self;
return op.traced(trace);
return op;
});
/** @internal */
exports.embedInput = embedInput;
const ensuringWith = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, finalizer) => {
const ensuringWith = /*#__PURE__*/(0, _Function.dual)(2, (self, finalizer) => {
const op = Object.create(proto);
op._tag = OpCodes.OP_ENSURING;
op.channel = self;
op.finalizer = restore(finalizer);
return op.traced(trace);
op.finalizer = finalizer;
return op;
});
/** @internal */
exports.ensuringWith = ensuringWith;
const fail = /*#__PURE__*/Debug.methodWithTrace(trace => error => failCause(Cause.fail(error)).traced(trace));
const fail = error => failCause(Cause.fail(error));
/** @internal */
exports.fail = fail;
const failSync = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => evaluate => failCauseSync(() => Cause.fail(restore(evaluate)())).traced(trace));
const failSync = evaluate => failCauseSync(() => Cause.fail(evaluate()));
/** @internal */
exports.failSync = failSync;
const failCause = /*#__PURE__*/Debug.methodWithTrace(trace => cause => failCauseSync(() => cause).traced(trace));
const failCause = cause => failCauseSync(() => cause);
/** @internal */
exports.failCause = failCause;
const failCauseSync = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => evaluate => {
const failCauseSync = evaluate => {
const op = Object.create(proto);
op._tag = OpCodes.OP_FAIL;
op.error = restore(evaluate);
return op.traced(trace);
});
op.error = evaluate;
return op;
};
/** @internal */
exports.failCauseSync = failCauseSync;
const flatMap = /*#__PURE__*/Debug.dualWithTrace(2, (trace, restore) => (self, f) => {
const flatMap = /*#__PURE__*/(0, _Function.dual)(2, (self, f) => {
const op = Object.create(proto);
op._tag = OpCodes.OP_FOLD;
op.channel = self;
op.k = new _continuation.ContinuationKImpl(restore(f), failCause);
return op.traced(trace);
op.k = new _continuation.ContinuationKImpl(f, failCause);
return op;
});
/** @internal */
exports.flatMap = flatMap;
const foldCauseChannel = /*#__PURE__*/Debug.dualWithTrace(3, (trace, restore) => (self, onError, onSuccess) => {
const foldCauseChannel = /*#__PURE__*/(0, _Function.dual)(3, (self, onError, onSuccess) => {
const op = Object.create(proto);
op._tag = OpCodes.OP_FOLD;
op.channel = self;
op.k = new _continuation.ContinuationKImpl(restore(onSuccess), restore(onError));
return op.traced(trace);
op.k = new _continuation.ContinuationKImpl(onSuccess, onError);
return op;
});
/** @internal */
exports.foldCauseChannel = foldCauseChannel;
const fromEffect = /*#__PURE__*/Debug.methodWithTrace(trace => effect => {
const fromEffect = effect => {
const op = Object.create(proto);
op._tag = OpCodes.OP_FROM_EFFECT;
op.effect = () => effect;
return op.traced(trace);
});
return op;
};
/** @internal */
exports.fromEffect = fromEffect;
const pipeTo = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, that) => {
const pipeTo = /*#__PURE__*/(0, _Function.dual)(2, (self, that) => {
const op = Object.create(proto);

@@ -200,7 +183,7 @@ op._tag = OpCodes.OP_PIPE_TO;

op.right = () => that;
return op.traced(trace);
return op;
});
/** @internal */
exports.pipeTo = pipeTo;
const provideContext = /*#__PURE__*/Debug.dualWithTrace(2, trace => (self, env) => {
const provideContext = /*#__PURE__*/(0, _Function.dual)(2, (self, env) => {
const op = Object.create(proto);

@@ -210,7 +193,7 @@ op._tag = OpCodes.OP_PROVIDE;

op.inner = self;
return op.traced(trace);
return op;
});
/** @internal */
exports.provideContext = provideContext;
const readOrFail = /*#__PURE__*/Debug.methodWithTrace(trace => error => {
const readOrFail = error => {
const op = Object.create(proto);

@@ -220,54 +203,54 @@ op._tag = OpCodes.OP_READ;

op.done = new _continuation.ContinuationKImpl(() => fail(error), () => fail(error));
return op.traced(trace);
});
return op;
};
/** @internal */
exports.readOrFail = readOrFail;
const readWith = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => (input, error, done) => readWithCause(restore(input), cause => Either.match(Cause.failureOrCause(cause), restore(error), failCause), restore(done)));
const readWith = (input, error, done) => readWithCause(input, cause => Either.match(Cause.failureOrCause(cause), error, failCause), done);
/** @internal */
exports.readWith = readWith;
const readWithCause = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => (input, halt, done) => {
const readWithCause = (input, halt, done) => {
const op = Object.create(proto);
op._tag = OpCodes.OP_READ;
op.more = restore(input);
op.done = new _continuation.ContinuationKImpl(restore(done), restore(halt));
return op.traced(trace);
});
op.more = input;
op.done = new _continuation.ContinuationKImpl(done, halt);
return op;
};
/** @internal */
exports.readWithCause = readWithCause;
const succeed = /*#__PURE__*/Debug.methodWithTrace(trace => value => sync(() => value).traced(trace));
const succeed = value => sync(() => value);
/** @internal */
exports.succeed = succeed;
const succeedNow = /*#__PURE__*/Debug.methodWithTrace(trace => result => {
const succeedNow = result => {
const op = Object.create(proto);
op._tag = OpCodes.OP_SUCCEED_NOW;
op.terminal = result;
return op.traced(trace);
});
return op;
};
/** @internal */
exports.succeedNow = succeedNow;
const suspend = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => evaluate => {
const suspend = evaluate => {
const op = Object.create(proto);
op._tag = OpCodes.OP_SUSPEND;
op.channel = restore(evaluate);
return op.traced(trace);
});
op.channel = evaluate;
return op;
};
exports.suspend = suspend;
const sync = /*#__PURE__*/Debug.methodWithTrace((trace, restore) => evaluate => {
const sync = evaluate => {
const op = Object.create(proto);
op._tag = OpCodes.OP_SUCCEED;
op.evaluate = restore(evaluate);
return op.traced(trace);
});
op.evaluate = evaluate;
return op;
};
/** @internal */
exports.sync = sync;
const unit = /*#__PURE__*/Debug.methodWithTrace(trace => () => succeedNow(void 0).traced(trace));
const unit = () => succeedNow(void 0);
/** @internal */
exports.unit = unit;
const write = /*#__PURE__*/Debug.methodWithTrace(trace => out => {
const write = out => {
const op = Object.create(proto);
op._tag = OpCodes.OP_EMIT;
op.out = out;
return op.traced(trace);
});
return op;
};
exports.write = write;
//# sourceMappingURL=core.js.map

5

internal/opCodes/channel.js

@@ -6,3 +6,3 @@ "use strict";

});
exports.OP_TRACED = exports.OP_SUSPEND = exports.OP_SUCCEED_NOW = exports.OP_SUCCEED = exports.OP_READ = exports.OP_PROVIDE = exports.OP_PIPE_TO = exports.OP_FROM_EFFECT = exports.OP_FOLD = exports.OP_FAIL = exports.OP_ENSURING = exports.OP_EMIT = exports.OP_CONCAT_ALL = exports.OP_BRIDGE = exports.OP_BRACKET_OUT = void 0;
exports.OP_SUSPEND = exports.OP_SUCCEED_NOW = exports.OP_SUCCEED = exports.OP_READ = exports.OP_PROVIDE = exports.OP_PIPE_TO = exports.OP_FROM_EFFECT = exports.OP_FOLD = exports.OP_FAIL = exports.OP_ENSURING = exports.OP_EMIT = exports.OP_CONCAT_ALL = exports.OP_BRIDGE = exports.OP_BRACKET_OUT = void 0;
/** @internal */

@@ -49,6 +49,3 @@ const OP_BRACKET_OUT = "BracketOut";

const OP_SUSPEND = "Suspend";
/** @internal */
exports.OP_SUSPEND = OP_SUSPEND;
const OP_TRACED = "Traced";
exports.OP_TRACED = OP_TRACED;
//# sourceMappingURL=channel.js.map

21

internal/sink.js

@@ -60,7 +60,16 @@ "use strict";

/** @internal */
const collectAllN = n => {
return flatMap(builder => foldUntil(builder, n, (chunk, input) => Chunk.append(input)(chunk)))(fromEffect(Effect.sync(() => Chunk.empty())));
};
const collectAllN = n => suspend(() => fromChannel(collectAllNLoop(n, Chunk.empty())));
/** @internal */
exports.collectAllN = collectAllN;
const collectAllNLoop = (n, acc) => core.readWithCause(chunk => {
const [collected, leftovers] = Chunk.splitAt(chunk, n);
if (collected.length < n) {
return collectAllNLoop(n - collected.length, Chunk.concat(acc, collected));
}
if (Chunk.isEmpty(leftovers)) {
return core.succeed(Chunk.concat(acc, collected));
}
return core.flatMap(core.write(leftovers), () => core.succeed(Chunk.concat(acc, collected)));
}, core.failCause, () => core.succeed(acc));
/** @internal */
const collectAllFrom = self => collectAllWhileWith(Chunk.empty(), _Function.constTrue, (chunk, z) => Chunk.append(z)(chunk))(self);

@@ -163,3 +172,3 @@ /** @internal */

exports.contramapChunksEffect = contramapChunksEffect;
const count = () => foldLeft(0, (s, _) => s + 1);
const count = () => foldLeftChunks(0, (acc, chunk) => acc + chunk.length);
/** @internal */

@@ -549,3 +558,3 @@ exports.count = count;

exports.ignoreLeftover = ignoreLeftover;
const last = () => foldLeft(Option.none(), (_, input) => Option.some(input));
const last = () => foldLeftChunks(Option.none(), (s, input) => Option.orElse(Chunk.last(input), () => s));
/** @internal */

@@ -740,3 +749,3 @@ exports.last = last;

exports.succeed = succeed;
const sum = () => foldLeft(0, (a, b) => a + b);
const sum = () => foldLeftChunks(0, (acc, chunk) => acc + Chunk.reduce(chunk, 0, (s, a) => s + a));
/** @internal */

@@ -743,0 +752,0 @@ exports.sum = sum;

7

internal/stream.d.ts

@@ -7,5 +7,10 @@ import * as Option from "@effect/data/Option";

import * as Schedule from "@effect/io/Schedule";
import type * as Scope from "@effect/io/Scope";
import * as Scope from "@effect/io/Scope";
import type * as Stream from "@effect/stream/Stream";
export declare const distributedWithDynamicCallback: (<E, A, _>(maximumLag: number, decide: (a: A) => Effect.Effect<never, never, Predicate<number>>, done: (exit: Exit.Exit<Option.Option<E>, never>) => Effect.Effect<never, never, _>) => <R>(self: Stream.Stream<R, E, A>) => Effect.Effect<Scope.Scope | R, never, Effect.Effect<never, never, readonly [number, Queue.Dequeue<Exit.Exit<Option.Option<E>, A>>]>>) & (<R_1, E_1, A_1, __1>(self: Stream.Stream<R_1, E_1, A_1>, maximumLag: number, decide: (a: A_1) => Effect.Effect<never, never, Predicate<number>>, done: (exit: Exit.Exit<Option.Option<E_1>, never>) => Effect.Effect<never, never, __1>) => Effect.Effect<Scope.Scope | R_1, never, Effect.Effect<never, never, readonly [number, Queue.Dequeue<Exit.Exit<Option.Option<E_1>, A_1>>]>>);
export declare class ReadableStreamError {
readonly reason: unknown;
readonly _tag = "ReadableStreamError";
constructor(reason: unknown);
}
/**

@@ -12,0 +17,0 @@ * Repeats the value using the provided schedule.

6

internal/stream/sinkEndReason.js

@@ -6,3 +6,3 @@ "use strict";

});
exports.UpstreamEnd = exports.SchedulEnd = exports.OP_UPSTREAM_END = exports.OP_SCHEDULE_END = void 0;
exports.UpstreamEnd = exports.ScheduleEnd = exports.OP_UPSTREAM_END = exports.OP_SCHEDULE_END = void 0;
/** @internal */

@@ -15,7 +15,7 @@ const OP_SCHEDULE_END = "ScheduleEnd";

exports.OP_UPSTREAM_END = OP_UPSTREAM_END;
const SchedulEnd = {
const ScheduleEnd = {
_tag: OP_SCHEDULE_END
};
/** @internal */
exports.SchedulEnd = SchedulEnd;
exports.ScheduleEnd = ScheduleEnd;
const UpstreamEnd = {

@@ -22,0 +22,0 @@ _tag: OP_UPSTREAM_END

6

package.json
{
"name": "@effect/stream",
"version": "0.23.0",
"version": "0.24.0",
"license": "MIT",

@@ -10,4 +10,4 @@ "repository": {

"dependencies": {
"@effect/data": "~0.12.9",
"@effect/io": "~0.27.0"
"@effect/data": "~0.12.10",
"@effect/io": "~0.29.2"
},

@@ -14,0 +14,0 @@ "publishConfig": {

5

Sink.d.ts

@@ -95,3 +95,6 @@ /**

*/
export declare const as: (<Z2>(z: Z2) => <R, E, In, L, Z>(self: Sink<R, E, In, L, Z>) => Sink<R, E, In, L, Z2>) & (<R_1, E_1, In_1, L_1, Z_1, Z2_1>(self: Sink<R_1, E_1, In_1, L_1, Z_1>, z: Z2_1) => Sink<R_1, E_1, In_1, L_1, Z2_1>);
export declare const as: {
<Z2>(z: Z2): <R, E, In, L, Z>(self: Sink<R, E, In, L, Z>) => Sink<R, E, In, L, Z2>;
<R, E, In, L, Z, Z2>(self: Sink<R, E, In, L, Z>, z: Z2): Sink<R, E, In, L, Z2>;
};
/**

@@ -98,0 +101,0 @@ * A sink that collects all elements into a `Chunk`.

410

src/internal/channel/channelExecutor.ts

@@ -1,6 +0,3 @@

import * as Chunk from "@effect/data/Chunk"
import type * as Context from "@effect/data/Context"
import * as Debug from "@effect/data/Debug"
import { identity, pipe } from "@effect/data/Function"
import * as MRef from "@effect/data/MutableRef"
import * as Option from "@effect/data/Option"

@@ -80,4 +77,2 @@ import * as Cause from "@effect/io/Cause"

private _traceStack: Array<Debug.SourceLocation> = []
private _executeCloseLastSubstream: (

@@ -326,3 +321,7 @@ effect: Effect.Effect<Env, never, unknown>

(emitted) => {
this._currentChannel = read.more(emitted) as core.Primitive
try {
this._currentChannel = read.more(emitted) as core.Primitive
} catch (error) {
this._currentChannel = read.done.onExit(Exit.die(error)) as core.Primitive
}
return undefined

@@ -356,13 +355,2 @@ },

case ChannelOpCodes.OP_TRACED: {
this._traceStack.push(this._currentChannel.trace)
this.addFinalizer(() =>
Effect.sync(() => {
this._traceStack.pop()
})
)
this._currentChannel = this._currentChannel.channel as core.Primitive
break
}
default: {

@@ -383,16 +371,2 @@ // @ts-expect-error

stackToLines(): Chunk.Chunk<Debug.SourceLocation> {
if (this._traceStack.length === 0) {
return Chunk.empty()
}
const lines: Array<Debug.SourceLocation> = []
let current = this._traceStack.length - 1
while (current >= 0 && lines.length < Debug.runtimeDebug.traceStackLimit) {
const value = this._traceStack[current]!
lines.push(value)
current = current - 1
}
return Chunk.unsafeFromArray(lines)
}
getDone(): Exit.Exit<OutErr, OutDone> {

@@ -419,19 +393,17 @@ return this._done as Exit.Exit<OutErr, OutDone>

popAllFinalizers(exit: Exit.Exit<unknown, unknown>): Effect.Effect<Env, never, unknown> {
return Debug.untraced(() => {
const finalizers: Array<ErasedFinalizer<Env>> = []
let next = this._doneStack.pop() as Continuation.Primitive | undefined
while (next) {
if (next._tag === "ContinuationFinalizer") {
finalizers.push(next.finalizer as ErasedFinalizer<Env>)
}
next = this._doneStack.pop() as Continuation.Primitive | undefined
const finalizers: Array<ErasedFinalizer<Env>> = []
let next = this._doneStack.pop() as Continuation.Primitive | undefined
while (next) {
if (next._tag === "ContinuationFinalizer") {
finalizers.push(next.finalizer as ErasedFinalizer<Env>)
}
const effect = (finalizers.length === 0 ? Effect.unit() : runFinalizers(finalizers, exit)) as Effect.Effect<
Env,
never,
unknown
>
this.storeInProgressFinalizer(effect)
return effect
})
next = this._doneStack.pop() as Continuation.Primitive | undefined
}
const effect = (finalizers.length === 0 ? Effect.unit() : runFinalizers(finalizers, exit)) as Effect.Effect<
Env,
never,
unknown
>
this.storeInProgressFinalizer(effect)
return effect
}

@@ -456,54 +428,50 @@

): Effect.Effect<Env, never, unknown> | undefined {
return Debug.untraced(() => {
const currInput = this._input
this._input = prev
if (currInput !== undefined) {
const effect = currInput.close(exit)
return effect
}
return Effect.unit()
})
const currInput = this._input
this._input = prev
if (currInput !== undefined) {
const effect = currInput.close(exit)
return effect
}
return Effect.unit()
}
close(exit: Exit.Exit<unknown, unknown>): Effect.Effect<Env, never, unknown> | undefined {
return Debug.untraced(() => {
let runInProgressFinalizers: Effect.Effect<Env, never, unknown> | undefined = undefined
const finalizer = this._inProgressFinalizer
if (finalizer !== undefined) {
runInProgressFinalizers = pipe(
finalizer,
Effect.ensuring(Effect.sync(() => this.clearInProgressFinalizer()))
)
}
let runInProgressFinalizers: Effect.Effect<Env, never, unknown> | undefined = undefined
const finalizer = this._inProgressFinalizer
if (finalizer !== undefined) {
runInProgressFinalizers = pipe(
finalizer,
Effect.ensuring(Effect.sync(() => this.clearInProgressFinalizer()))
)
}
let closeSelf: Effect.Effect<Env, never, unknown> | undefined = undefined
const selfFinalizers = this.popAllFinalizers(exit)
if (selfFinalizers !== undefined) {
closeSelf = pipe(
selfFinalizers,
Effect.ensuring(Effect.sync(() => this.clearInProgressFinalizer()))
)
}
let closeSelf: Effect.Effect<Env, never, unknown> | undefined = undefined
const selfFinalizers = this.popAllFinalizers(exit)
if (selfFinalizers !== undefined) {
closeSelf = pipe(
selfFinalizers,
Effect.ensuring(Effect.sync(() => this.clearInProgressFinalizer()))
)
}
const closeSubexecutors = this._activeSubexecutor === undefined ?
undefined :
this._activeSubexecutor.close(exit)
const closeSubexecutors = this._activeSubexecutor === undefined ?
undefined :
this._activeSubexecutor.close(exit)
if (
closeSubexecutors === undefined &&
runInProgressFinalizers === undefined &&
closeSelf === undefined
) {
return undefined
}
if (
closeSubexecutors === undefined &&
runInProgressFinalizers === undefined &&
closeSelf === undefined
) {
return undefined
}
return pipe(
Effect.exit(ifNotNull(closeSubexecutors)),
Effect.zip(Effect.exit(ifNotNull(runInProgressFinalizers))),
Effect.zip(Effect.exit(ifNotNull(closeSelf))),
Effect.map(([[exit1, exit2], exit3]) => pipe(exit1, Exit.zipRight(exit2), Exit.zipRight(exit3))),
Effect.uninterruptible,
Effect.flatMap(Effect.done)
)
})
return pipe(
Effect.exit(ifNotNull(closeSubexecutors)),
Effect.zip(Effect.exit(ifNotNull(runInProgressFinalizers))),
Effect.zip(Effect.exit(ifNotNull(closeSelf))),
Effect.map(([[exit1, exit2], exit3]) => pipe(exit1, Exit.zipRight(exit2), Exit.zipRight(exit3))),
Effect.uninterruptible,
Effect.flatMap(Effect.done)
)
}

@@ -546,36 +514,2 @@

annotate<E>(_cause: Cause.Cause<E>) {
let cause = _cause
if (Cause.isAnnotatedType(cause) && Cause.isStackAnnotation(cause.annotation)) {
const stack = cause.annotation.stack
const currentStack = this.stackToLines()
cause = Cause.annotated(
cause.cause,
new Cause.StackAnnotation(
pipe(
stack.length === 0 ?
currentStack :
currentStack.length === 0 ?
stack :
Chunk.unsafeLast(stack) === Chunk.unsafeLast(currentStack) ?
stack :
pipe(
stack,
Chunk.concat(currentStack)
),
Chunk.dedupeAdjacent,
Chunk.take(Debug.runtimeDebug.traceStackLimit)
),
cause.annotation.seq
)
)
} else {
cause = Cause.annotated(
cause,
new Cause.StackAnnotation(this.stackToLines(), MRef.getAndIncrement(Cause.globalErrorSeq))
)
}
return cause
}
doneHalt(cause: Cause.Cause<unknown>): ChannelState.ChannelState<Env, unknown> | undefined {

@@ -691,15 +625,13 @@ if (this._doneStack.length === 0) {

finishWithExit(exit: Exit.Exit<unknown, unknown>): Effect.Effect<Env, unknown, unknown> {
return Debug.untraced(() => {
const state = pipe(
exit,
Exit.match(
(cause) => this.doneHalt(cause),
(value) => this.doneSucceed(value)
)
const state = pipe(
exit,
Exit.match(
(cause) => this.doneHalt(cause),
(value) => this.doneSucceed(value)
)
this._activeSubexecutor = undefined
return state === undefined ?
Effect.unit() :
ChannelState.effect(state)
})
)
this._activeSubexecutor = undefined
return state === undefined ?
Effect.unit() :
ChannelState.effect(state)
}

@@ -1094,21 +1026,15 @@

const ifNotNull = Debug.untracedMethod(() =>
<Env>(effect: Effect.Effect<Env, never, unknown> | undefined): Effect.Effect<Env, never, unknown> =>
effect !== undefined ?
effect :
Effect.unit()
)
const ifNotNull = <Env>(effect: Effect.Effect<Env, never, unknown> | undefined): Effect.Effect<Env, never, unknown> =>
effect !== undefined ? effect : Effect.unit()
const runFinalizers = Debug.untracedMethod(() =>
<Env>(
finalizers: Array<ErasedFinalizer<Env>>,
exit: Exit.Exit<unknown, unknown>
): Effect.Effect<Env, never, unknown> => {
return pipe(
Effect.forEach(finalizers, (fin) => Effect.exit(fin(exit))),
Effect.map((exits) => pipe(Exit.collectAll(exits), Option.getOrElse(() => Exit.unit()))),
Effect.flatMap((exit) => Effect.done(exit as Exit.Exit<never, unknown>))
)
}
)
const runFinalizers = <Env>(
finalizers: Array<ErasedFinalizer<Env>>,
exit: Exit.Exit<unknown, unknown>
): Effect.Effect<Env, never, unknown> => {
return pipe(
Effect.forEach(finalizers, (fin) => Effect.exit(fin(exit))),
Effect.map((exits) => pipe(Exit.collectAll(exits), Option.getOrElse(() => Exit.unit()))),
Effect.flatMap((exit) => Effect.done(exit as Exit.Exit<never, unknown>))
)
}

@@ -1118,87 +1044,83 @@ /**

*/
export const readUpstream = Debug.methodWithTrace((trace) =>
<R, E, E2, A>(
r: ChannelState.Read,
onSuccess: () => Effect.Effect<R, E2, A>,
onFailure: (cause: Cause.Cause<E>) => Effect.Effect<R, E2, A>
): Effect.Effect<R, E2, A> => {
const readStack = [r as ChannelState.Read]
const read = (): Effect.Effect<R, E2, A> => {
const current = readStack.pop()
if (current === undefined || current.upstream === undefined) {
return Effect.dieMessage("Unexpected end of input for channel execution")
}
const state = current.upstream.run() as ChannelState.Primitive
switch (state._tag) {
case ChannelStateOpCodes.OP_EMIT: {
const emitEffect = current.onEmit(current.upstream.getEmit())
if (readStack.length === 0) {
if (emitEffect === undefined) {
return Effect.suspend(onSuccess)
}
return pipe(
emitEffect as Effect.Effect<never, never, void>,
Effect.matchCauseEffect(onFailure, onSuccess)
).traced(trace)
}
export const readUpstream = <R, E, E2, A>(
r: ChannelState.Read,
onSuccess: () => Effect.Effect<R, E2, A>,
onFailure: (cause: Cause.Cause<E>) => Effect.Effect<R, E2, A>
): Effect.Effect<R, E2, A> => {
const readStack = [r as ChannelState.Read]
const read = (): Effect.Effect<R, E2, A> => {
const current = readStack.pop()
if (current === undefined || current.upstream === undefined) {
return Effect.dieMessage("Unexpected end of input for channel execution")
}
const state = current.upstream.run() as ChannelState.Primitive
switch (state._tag) {
case ChannelStateOpCodes.OP_EMIT: {
const emitEffect = current.onEmit(current.upstream.getEmit())
if (readStack.length === 0) {
if (emitEffect === undefined) {
return Effect.suspend(() => read()).traced(trace)
return Effect.suspend(onSuccess)
}
return pipe(
emitEffect as Effect.Effect<never, never, void>,
Effect.matchCauseEffect(onFailure, () => read())
).traced(trace)
Effect.matchCauseEffect(onFailure, onSuccess)
)
}
if (emitEffect === undefined) {
return Effect.suspend(() => read())
}
return pipe(
emitEffect as Effect.Effect<never, never, void>,
Effect.matchCauseEffect(onFailure, () => read())
)
}
case ChannelStateOpCodes.OP_DONE: {
const doneEffect = current.onDone(current.upstream.getDone())
if (readStack.length === 0) {
if (doneEffect === undefined) {
return Effect.suspend(onSuccess).traced(trace)
}
return pipe(
doneEffect as Effect.Effect<never, never, void>,
Effect.matchCauseEffect(onFailure, () => onSuccess())
).traced(trace)
}
case ChannelStateOpCodes.OP_DONE: {
const doneEffect = current.onDone(current.upstream.getDone())
if (readStack.length === 0) {
if (doneEffect === undefined) {
return Effect.suspend(() => read()).traced(trace)
return Effect.suspend(onSuccess)
}
return pipe(
doneEffect as Effect.Effect<never, never, void>,
Effect.matchCauseEffect(onFailure, () => read())
).traced(trace)
Effect.matchCauseEffect(onFailure, () => onSuccess())
)
}
case ChannelStateOpCodes.OP_FROM_EFFECT: {
readStack.push(current)
return pipe(
current.onEffect(state.effect as Effect.Effect<never, never, void>) as Effect.Effect<never, never, void>,
Effect.catchAllCause((cause) =>
Effect.suspend(() => {
const doneEffect = current.onDone(Exit.failCause(cause)) as Effect.Effect<never, never, void>
return doneEffect === undefined ? Effect.unit() : doneEffect
})
),
Effect.matchCauseEffect(onFailure, () => read())
).traced(trace)
if (doneEffect === undefined) {
return Effect.suspend(() => read())
}
return pipe(
doneEffect as Effect.Effect<never, never, void>,
Effect.matchCauseEffect(onFailure, () => read())
)
}
case ChannelStateOpCodes.OP_READ: {
readStack.push(current)
readStack.push(state)
return Effect.suspend(() => read()).traced(trace)
}
case ChannelStateOpCodes.OP_FROM_EFFECT: {
readStack.push(current)
return pipe(
current.onEffect(state.effect as Effect.Effect<never, never, void>) as Effect.Effect<never, never, void>,
Effect.catchAllCause((cause) =>
Effect.suspend(() => {
const doneEffect = current.onDone(Exit.failCause(cause)) as Effect.Effect<never, never, void>
return doneEffect === undefined ? Effect.unit() : doneEffect
})
),
Effect.matchCauseEffect(onFailure, () => read())
)
}
case ChannelStateOpCodes.OP_READ: {
readStack.push(current)
readStack.push(state)
return Effect.suspend(() => read())
}
}
return read()
}
)
return read()
}
/** @internal */
export const run = Debug.methodWithTrace((trace) =>
<Env, InErr, InDone, OutErr, OutDone>(
self: Channel.Channel<Env, InErr, unknown, InDone, OutErr, never, OutDone>
): Effect.Effect<Env, OutErr, OutDone> => pipe(runScoped(self), Effect.scoped).traced(trace)
)
export const run = <Env, InErr, InDone, OutErr, OutDone>(
self: Channel.Channel<Env, InErr, unknown, InDone, OutErr, never, OutDone>
): Effect.Effect<Env, OutErr, OutDone> => pipe(runScoped(self), Effect.scoped)

@@ -1209,3 +1131,7 @@ /** @internal */

): Effect.Effect<Env | Scope.Scope, OutErr, OutDone> => {
const run = (deferred: Deferred.Deferred<OutErr, OutDone>, scope: Scope.Scope) =>
const run = (
channelDeferred: Deferred.Deferred<OutErr, OutDone>,
scopeDeferred: Deferred.Deferred<never, void>,
scope: Scope.Scope
) =>
Effect.acquireUseRelease(

@@ -1217,5 +1143,5 @@ Effect.sync(() => new ChannelExecutor(self, void 0, identity)),

runScopedInterpret(exec.run() as ChannelState.ChannelState<Env, OutErr>, exec),
Effect.intoDeferred(deferred),
Effect.zipRight(Deferred.await(deferred)),
Effect.zipLeft(Effect.never())
Effect.intoDeferred(channelDeferred),
Effect.zipRight(Deferred.await(channelDeferred)),
Effect.zipLeft(Deferred.await(scopeDeferred))
)

@@ -1234,10 +1160,24 @@ ),

)
return Effect.flatMap(Effect.scope(), (parent) =>
Effect.flatMap(
Scope.fork(parent, ExecutionStrategy.sequential),
(child) =>
Effect.flatMap(Deferred.make<OutErr, OutDone>(), (deferred) =>
Effect.flatMap(Effect.forkScoped(run(deferred, child)), (fiber) =>
Effect.zipLeft(Deferred.await(deferred), Fiber.inheritAll(fiber))))
))
return Effect.uninterruptibleMask((restore) =>
Effect.flatMap(Effect.scope(), (parent) =>
pipe(
Effect.all(
Scope.fork(parent, ExecutionStrategy.sequential),
Deferred.make<OutErr, OutDone>(),
Deferred.make<never, void>()
),
Effect.flatMap(([child, channelDeferred, scopeDeferred]) =>
pipe(
Effect.forkScoped(restore(run(channelDeferred, scopeDeferred, child))),
Effect.flatMap((fiber) =>
pipe(
Effect.addFinalizer(() => Deferred.succeed(scopeDeferred, void 0)),
Effect.zipRight(restore(Deferred.await(channelDeferred))),
Effect.zipLeft(Fiber.inheritAll(fiber))
)
)
)
)
))
)
}

@@ -1244,0 +1184,0 @@

940

src/internal/core.ts
import * as Chunk from "@effect/data/Chunk"
import type * as Context from "@effect/data/Context"
import * as Debug from "@effect/data/Debug"
import * as Either from "@effect/data/Either"
import { constVoid, identity } from "@effect/data/Function"
import { constVoid, dual, identity } from "@effect/data/Function"
import type { LazyArg } from "@effect/data/Function"

@@ -43,13 +42,3 @@ import * as Option from "@effect/data/Option"

const proto = {
[ChannelTypeId]: channelVariance,
traced(trace: Debug.Trace) {
if (trace) {
return Object.create(proto, {
_tag: { value: OpCodes.OP_TRACED },
channel: { value: this },
trace: { value: trace }
})
}
return this
}
[ChannelTypeId]: channelVariance
}

@@ -81,3 +70,2 @@

| Suspend
| Traced

@@ -200,11 +188,3 @@ /** @internal */

/** @internal */
export interface Traced extends
Op<OpCodes.OP_TRACED, {
readonly channel: ErasedChannel
readonly trace: NonNullable<Debug.Trace>
}>
{}
/** @internal */
export const acquireReleaseOut = Debug.dualWithTrace<
export const acquireReleaseOut = dual<
<R2, Z>(

@@ -217,13 +197,12 @@ release: (z: Z, e: Exit.Exit<unknown, unknown>) => Effect.Effect<R2, never, unknown>

) => Channel.Channel<R | R2, unknown, unknown, unknown, E, Z, void>
>(2, (trace, restore) =>
(self, release) => {
const op = Object.create(proto)
op._tag = OpCodes.OP_BRACKET_OUT
op.acquire = () => self
op.finalizer = restore(release)
return op.traced(trace)
})
>(2, (self, release) => {
const op = Object.create(proto)
op._tag = OpCodes.OP_BRACKET_OUT
op.acquire = () => self
op.finalizer = release
return op
})
/** @internal */
export const catchAllCause = Debug.dualWithTrace<
export const catchAllCause = dual<
<Env1, InErr1, InElem1, InDone1, OutErr, OutErr1, OutElem1, OutDone1>(

@@ -256,43 +235,19 @@ f: (cause: Cause.Cause<OutErr>) => Channel.Channel<Env1, InErr1, InElem1, InDone1, OutErr1, OutElem1, OutDone1>

2,
(trace, restore) =>
<Env, InErr, InElem, InDone, OutElem, OutDone, Env1, InErr1, InElem1, InDone1, OutErr, OutErr1, OutElem1, OutDone1>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
f: (cause: Cause.Cause<OutErr>) => Channel.Channel<Env1, InErr1, InElem1, InDone1, OutErr1, OutElem1, OutDone1>
): Channel.Channel<
Env | Env1,
InErr & InErr1,
InElem & InElem1,
InDone & InDone1,
OutErr1,
OutElem | OutElem1,
OutDone | OutDone1
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_FOLD
op.channel = self
op.k = new ContinuationKImpl(succeed, restore(f))
return op.traced(trace)
}
)
/** @internal */
export const collectElements = Debug.methodWithTrace((trace) =>
<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>
<Env, InErr, InElem, InDone, OutElem, OutDone, Env1, InErr1, InElem1, InDone1, OutErr, OutErr1, OutElem1, OutDone1>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
f: (cause: Cause.Cause<OutErr>) => Channel.Channel<Env1, InErr1, InElem1, InDone1, OutErr1, OutElem1, OutDone1>
): Channel.Channel<
Env,
InErr,
InElem,
InDone,
OutErr,
never,
readonly [Chunk.Chunk<OutElem>, OutDone]
Env | Env1,
InErr & InErr1,
InElem & InElem1,
InDone & InDone1,
OutErr1,
OutElem | OutElem1,
OutDone | OutDone1
> => {
return suspend(() => {
const builder: Array<OutElem> = []
return flatMap(
pipeTo(self, collectElementsReader(builder)),
(value) => sync(() => [Chunk.fromIterable(builder), value] as const)
)
}).traced(trace)
const op = Object.create(proto)
op._tag = OpCodes.OP_FOLD
op.channel = self
op.k = new ContinuationKImpl(succeed, f)
return op
}

@@ -302,2 +257,23 @@ )

/** @internal */
export const collectElements = <Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>
): Channel.Channel<
Env,
InErr,
InElem,
InDone,
OutErr,
never,
readonly [Chunk.Chunk<OutElem>, OutDone]
> => {
return suspend(() => {
const builder: Array<OutElem> = []
return flatMap(
pipeTo(self, collectElementsReader(builder)),
(value) => sync(() => [Chunk.fromIterable(builder), value] as const)
)
})
}
/** @internal */
const collectElementsReader = <OutErr, OutElem, OutDone>(

@@ -320,20 +296,32 @@ builder: Array<OutElem>

/** @internal */
export const concatAll = Debug.methodWithTrace((trace) =>
<Env, InErr, InElem, InDone, OutErr, OutElem>(
channels: Channel.Channel<
Env,
InErr,
InElem,
InDone,
OutErr,
Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, any>,
any
>
): Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, any> =>
concatAllWith(channels, constVoid, constVoid).traced(trace)
)
export const concatAll = <Env, InErr, InElem, InDone, OutErr, OutElem>(
channels: Channel.Channel<
Env,
InErr,
InElem,
InDone,
OutErr,
Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, any>,
any
>
): Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, any> => concatAllWith(channels, constVoid, constVoid)
/** @internal */
export const concatAllWith = Debug.methodWithTrace((trace, restore) =>
<
export const concatAllWith = <
Env,
InErr,
InElem,
InDone,
OutErr,
OutElem,
OutDone,
OutDone2,
OutDone3,
Env2,
InErr2,
InElem2,
InDone2,
OutErr2
>(
channels: Channel.Channel<
Env,

@@ -344,46 +332,29 @@ InErr,

OutErr,
OutElem,
OutDone,
OutDone2,
OutDone3,
Env2,
InErr2,
InElem2,
InDone2,
OutErr2
>(
channels: Channel.Channel<
Env,
InErr,
InElem,
InDone,
OutErr,
Channel.Channel<Env2, InErr2, InElem2, InDone2, OutErr2, OutElem, OutDone>,
OutDone2
>,
f: (o: OutDone, o1: OutDone) => OutDone,
g: (o: OutDone, o2: OutDone2) => OutDone3
): Channel.Channel<
Env | Env2,
InErr & InErr2,
InElem & InElem2,
InDone & InDone2,
OutErr | OutErr2,
OutElem,
OutDone3
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_CONCAT_ALL
op.combineInners = restore(f)
op.combineAll = restore(g)
op.onPull = () => upstreamPullStrategy.PullAfterNext(Option.none())
op.onEmit = () => childExecutorDecision.Continue
op.value = () => channels
op.k = identity
return op.traced(trace)
}
)
Channel.Channel<Env2, InErr2, InElem2, InDone2, OutErr2, OutElem, OutDone>,
OutDone2
>,
f: (o: OutDone, o1: OutDone) => OutDone,
g: (o: OutDone, o2: OutDone2) => OutDone3
): Channel.Channel<
Env | Env2,
InErr & InErr2,
InElem & InElem2,
InDone & InDone2,
OutErr | OutErr2,
OutElem,
OutDone3
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_CONCAT_ALL
op.combineInners = f
op.combineAll = g
op.onPull = () => upstreamPullStrategy.PullAfterNext(Option.none())
op.onEmit = () => childExecutorDecision.Continue
op.value = () => channels
op.k = identity
return op
}
/** @internal */
export const concatMapWith = Debug.dualWithTrace<
export const concatMapWith = dual<
<OutElem, OutElem2, OutDone, OutDone2, OutDone3, Env2, InErr2, InElem2, InDone2, OutErr2>(

@@ -434,51 +405,47 @@ f: (o: OutElem) => Channel.Channel<Env2, InErr2, InElem2, InDone2, OutErr2, OutElem2, OutDone>,

>
>(4, <
Env,
InErr,
InElem,
InDone,
OutErr,
OutElem,
OutElem2,
OutDone,
OutDone2,
OutDone3,
Env2,
InErr2,
InElem2,
InDone2,
OutErr2
>(
4,
(trace, restore) =>
<
Env,
InErr,
InElem,
InDone,
OutErr,
OutElem,
OutElem2,
OutDone,
OutDone2,
OutDone3,
Env2,
InErr2,
InElem2,
InDone2,
OutErr2
>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone2>,
f: (
o: OutElem
) => Channel.Channel<Env2, InErr2, InElem2, InDone2, OutErr2, OutElem2, OutDone>,
g: (o: OutDone, o1: OutDone) => OutDone,
h: (o: OutDone, o2: OutDone2) => OutDone3
): Channel.Channel<
Env | Env2,
InErr & InErr2,
InElem & InElem2,
InDone & InDone2,
OutErr | OutErr2,
OutElem2,
OutDone3
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_CONCAT_ALL
op.combineInners = restore(g)
op.combineAll = restore(h)
op.onPull = () => upstreamPullStrategy.PullAfterNext(Option.none())
op.onEmit = () => childExecutorDecision.Continue
op.value = () => self
op.k = restore(f)
return op.traced(trace)
}
)
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone2>,
f: (
o: OutElem
) => Channel.Channel<Env2, InErr2, InElem2, InDone2, OutErr2, OutElem2, OutDone>,
g: (o: OutDone, o1: OutDone) => OutDone,
h: (o: OutDone, o2: OutDone2) => OutDone3
): Channel.Channel<
Env | Env2,
InErr & InErr2,
InElem & InElem2,
InDone & InDone2,
OutErr | OutErr2,
OutElem2,
OutDone3
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_CONCAT_ALL
op.combineInners = g
op.combineAll = h
op.onPull = () => upstreamPullStrategy.PullAfterNext(Option.none())
op.onEmit = () => childExecutorDecision.Continue
op.value = () => self
op.k = f
return op
})
/** @internal */
export const concatMapWithCustom = Debug.dualWithTrace<
export const concatMapWithCustom = dual<
<OutElem, OutElem2, OutDone, OutDone2, OutDone3, Env2, InErr2, InElem2, InDone2, OutErr2>(

@@ -537,55 +504,51 @@ f: (o: OutElem) => Channel.Channel<Env2, InErr2, InElem2, InDone2, OutErr2, OutElem2, OutDone>,

>
>(6, <
Env,
InErr,
InElem,
InDone,
OutErr,
OutElem,
OutElem2,
OutDone,
OutDone2,
OutDone3,
Env2,
InErr2,
InElem2,
InDone2,
OutErr2
>(
6,
(trace, restore) =>
<
Env,
InErr,
InElem,
InDone,
OutErr,
OutElem,
OutElem2,
OutDone,
OutDone2,
OutDone3,
Env2,
InErr2,
InElem2,
InDone2,
OutErr2
>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone2>,
f: (
o: OutElem
) => Channel.Channel<Env2, InErr2, InElem2, InDone2, OutErr2, OutElem2, OutDone>,
g: (o: OutDone, o1: OutDone) => OutDone,
h: (o: OutDone, o2: OutDone2) => OutDone3,
onPull: (
upstreamPullRequest: UpstreamPullRequest.UpstreamPullRequest<OutElem>
) => UpstreamPullStrategy.UpstreamPullStrategy<OutElem2>,
onEmit: (elem: OutElem2) => ChildExecutorDecision.ChildExecutorDecision
): Channel.Channel<
Env | Env2,
InErr & InErr2,
InElem & InElem2,
InDone & InDone2,
OutErr | OutErr2,
OutElem2,
OutDone3
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_CONCAT_ALL
op.combineInners = restore(g)
op.combineAll = restore(h)
op.onPull = restore(onPull)
op.onEmit = restore(onEmit)
op.value = () => self
op.k = restore(f)
return op.traced(trace)
}
)
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone2>,
f: (
o: OutElem
) => Channel.Channel<Env2, InErr2, InElem2, InDone2, OutErr2, OutElem2, OutDone>,
g: (o: OutDone, o1: OutDone) => OutDone,
h: (o: OutDone, o2: OutDone2) => OutDone3,
onPull: (
upstreamPullRequest: UpstreamPullRequest.UpstreamPullRequest<OutElem>
) => UpstreamPullStrategy.UpstreamPullStrategy<OutElem2>,
onEmit: (elem: OutElem2) => ChildExecutorDecision.ChildExecutorDecision
): Channel.Channel<
Env | Env2,
InErr & InErr2,
InElem & InElem2,
InDone & InDone2,
OutErr | OutErr2,
OutElem2,
OutDone3
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_CONCAT_ALL
op.combineInners = g
op.combineAll = h
op.onPull = onPull
op.onEmit = onEmit
op.value = () => self
op.k = f
return op
})
/** @internal */
export const embedInput = Debug.dualWithTrace<
export const embedInput = dual<
<InErr, InElem, InDone>(

@@ -602,17 +565,16 @@ input: SingleProducerAsyncInput.AsyncInputProducer<InErr, InElem, InDone>

2,
(trace) =>
<Env, OutErr, OutElem, OutDone, InErr, InElem, InDone>(
self: Channel.Channel<Env, unknown, unknown, unknown, OutErr, OutElem, OutDone>,
input: SingleProducerAsyncInput.AsyncInputProducer<InErr, InElem, InDone>
): Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_BRIDGE
op.input = input
op.channel = self
return op.traced(trace)
}
<Env, OutErr, OutElem, OutDone, InErr, InElem, InDone>(
self: Channel.Channel<Env, unknown, unknown, unknown, OutErr, OutElem, OutDone>,
input: SingleProducerAsyncInput.AsyncInputProducer<InErr, InElem, InDone>
): Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_BRIDGE
op.input = input
op.channel = self
return op
}
)
/** @internal */
export const ensuringWith = Debug.dualWithTrace<
export const ensuringWith = dual<
<Env2, OutErr, OutDone>(

@@ -629,50 +591,40 @@ finalizer: (e: Exit.Exit<OutErr, OutDone>) => Effect.Effect<Env2, never, unknown>

2,
(trace, restore) =>
<Env, InErr, InElem, InDone, OutElem, Env2, OutErr, OutDone>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
finalizer: (e: Exit.Exit<OutErr, OutDone>) => Effect.Effect<Env2, never, unknown>
): Channel.Channel<Env | Env2, InErr, InElem, InDone, OutErr, OutElem, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_ENSURING
op.channel = self
op.finalizer = restore(finalizer)
return op.traced(trace)
}
<Env, InErr, InElem, InDone, OutElem, Env2, OutErr, OutDone>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
finalizer: (e: Exit.Exit<OutErr, OutDone>) => Effect.Effect<Env2, never, unknown>
): Channel.Channel<Env | Env2, InErr, InElem, InDone, OutErr, OutElem, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_ENSURING
op.channel = self
op.finalizer = finalizer
return op
}
)
/** @internal */
export const fail = Debug.methodWithTrace((trace) =>
<E>(error: E): Channel.Channel<never, unknown, unknown, unknown, E, never, never> =>
failCause(Cause.fail(error)).traced(trace)
)
export const fail = <E>(error: E): Channel.Channel<never, unknown, unknown, unknown, E, never, never> =>
failCause(Cause.fail(error))
/** @internal */
export const failSync = Debug.methodWithTrace((trace, restore) =>
<E>(
evaluate: LazyArg<E>
): Channel.Channel<never, unknown, unknown, unknown, E, never, never> =>
failCauseSync(() => Cause.fail(restore(evaluate)())).traced(trace)
)
export const failSync = <E>(
evaluate: LazyArg<E>
): Channel.Channel<never, unknown, unknown, unknown, E, never, never> => failCauseSync(() => Cause.fail(evaluate()))
/** @internal */
export const failCause = Debug.methodWithTrace((trace) =>
<E>(
cause: Cause.Cause<E>
): Channel.Channel<never, unknown, unknown, unknown, E, never, never> => failCauseSync(() => cause).traced(trace)
)
export const failCause = <E>(
cause: Cause.Cause<E>
): Channel.Channel<never, unknown, unknown, unknown, E, never, never> => failCauseSync(() => cause)
/** @internal */
export const failCauseSync = Debug.methodWithTrace((trace, restore) =>
<E>(
evaluate: LazyArg<Cause.Cause<E>>
): Channel.Channel<never, unknown, unknown, unknown, E, never, never> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_FAIL
op.error = restore(evaluate)
return op.traced(trace)
}
)
export const failCauseSync = <E>(
evaluate: LazyArg<Cause.Cause<E>>
): Channel.Channel<never, unknown, unknown, unknown, E, never, never> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_FAIL
op.error = evaluate
return op
}
/** @internal */
export const flatMap = Debug.dualWithTrace<
export const flatMap = dual<
<OutDone, Env1, InErr1, InElem1, InDone1, OutErr1, OutElem1, OutDone2>(

@@ -705,25 +657,24 @@ f: (d: OutDone) => Channel.Channel<Env1, InErr1, InElem1, InDone1, OutErr1, OutElem1, OutDone2>

2,
(trace, restore) =>
<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone, Env1, InErr1, InElem1, InDone1, OutErr1, OutElem1, OutDone2>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
f: (d: OutDone) => Channel.Channel<Env1, InErr1, InElem1, InDone1, OutErr1, OutElem1, OutDone2>
): Channel.Channel<
Env | Env1,
InErr & InErr1,
InElem & InElem1,
InDone & InDone1,
OutErr | OutErr1,
OutElem | OutElem1,
OutDone2
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_FOLD
op.channel = self
op.k = new ContinuationKImpl(restore(f), failCause)
return op.traced(trace)
}
<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone, Env1, InErr1, InElem1, InDone1, OutErr1, OutElem1, OutDone2>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
f: (d: OutDone) => Channel.Channel<Env1, InErr1, InElem1, InDone1, OutErr1, OutElem1, OutDone2>
): Channel.Channel<
Env | Env1,
InErr & InErr1,
InElem & InElem1,
InDone & InDone1,
OutErr | OutErr1,
OutElem | OutElem1,
OutDone2
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_FOLD
op.channel = self
op.k = new ContinuationKImpl(f, failCause)
return op
}
)
/** @internal */
export const foldCauseChannel = Debug.dualWithTrace<
export const foldCauseChannel = dual<
<

@@ -797,59 +748,46 @@ Env1,

3,
(trace, restore) =>
<
Env,
InErr,
InElem,
InDone,
OutElem,
Env1,
Env2,
InErr1,
InErr2,
InElem1,
InElem2,
InDone1,
InDone2,
OutErr,
OutErr2,
OutErr3,
OutElem1,
OutElem2,
OutDone,
OutDone2,
OutDone3
>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
onError: (
c: Cause.Cause<OutErr>
) => Channel.Channel<Env1, InErr1, InElem1, InDone1, OutErr2, OutElem1, OutDone2>,
onSuccess: (
o: OutDone
) => Channel.Channel<Env2, InErr2, InElem2, InDone2, OutErr3, OutElem2, OutDone3>
): Channel.Channel<
Env | Env1 | Env2,
InErr & InErr1 & InErr2,
InElem & InElem1 & InElem2,
InDone & InDone1 & InDone2,
OutErr2 | OutErr3,
OutElem | OutElem1 | OutElem2,
OutDone2 | OutDone3
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_FOLD
op.channel = self
op.k = new ContinuationKImpl(restore(onSuccess), restore(onError) as any)
return op.traced(trace)
}
)
/** @internal */
export const fromEffect = Debug.methodWithTrace((trace) =>
<R, E, A>(
effect: Effect.Effect<R, E, A>
): Channel.Channel<R, unknown, unknown, unknown, E, never, A> => {
<
Env,
InErr,
InElem,
InDone,
OutElem,
Env1,
Env2,
InErr1,
InErr2,
InElem1,
InElem2,
InDone1,
InDone2,
OutErr,
OutErr2,
OutErr3,
OutElem1,
OutElem2,
OutDone,
OutDone2,
OutDone3
>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
onError: (
c: Cause.Cause<OutErr>
) => Channel.Channel<Env1, InErr1, InElem1, InDone1, OutErr2, OutElem1, OutDone2>,
onSuccess: (
o: OutDone
) => Channel.Channel<Env2, InErr2, InElem2, InDone2, OutErr3, OutElem2, OutDone3>
): Channel.Channel<
Env | Env1 | Env2,
InErr & InErr1 & InErr2,
InElem & InElem1 & InElem2,
InDone & InDone1 & InDone2,
OutErr2 | OutErr3,
OutElem | OutElem1 | OutElem2,
OutDone2 | OutDone3
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_FROM_EFFECT
op.effect = () => effect
return op.traced(trace)
op._tag = OpCodes.OP_FOLD
op.channel = self
op.k = new ContinuationKImpl(onSuccess, onError as any)
return op
}

@@ -859,3 +797,13 @@ )

/** @internal */
export const pipeTo = Debug.dualWithTrace<
export const fromEffect = <R, E, A>(
effect: Effect.Effect<R, E, A>
): Channel.Channel<R, unknown, unknown, unknown, E, never, A> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_FROM_EFFECT
op.effect = () => effect
return op
}
/** @internal */
export const pipeTo = dual<
<Env2, OutErr, OutElem, OutDone, OutErr2, OutElem2, OutDone2>(

@@ -872,17 +820,16 @@ that: Channel.Channel<Env2, OutErr, OutElem, OutDone, OutErr2, OutElem2, OutDone2>

2,
(trace) =>
<Env, InErr, InElem, InDone, Env2, OutErr, OutElem, OutDone, OutErr2, OutElem2, OutDone2>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
that: Channel.Channel<Env2, OutErr, OutElem, OutDone, OutErr2, OutElem2, OutDone2>
): Channel.Channel<Env | Env2, InErr, InElem, InDone, OutErr2, OutElem2, OutDone2> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_PIPE_TO
op.left = () => self
op.right = () => that
return op.traced(trace)
}
<Env, InErr, InElem, InDone, Env2, OutErr, OutElem, OutDone, OutErr2, OutElem2, OutDone2>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
that: Channel.Channel<Env2, OutErr, OutElem, OutDone, OutErr2, OutElem2, OutDone2>
): Channel.Channel<Env | Env2, InErr, InElem, InDone, OutErr2, OutElem2, OutDone2> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_PIPE_TO
op.left = () => self
op.right = () => that
return op
}
)
/** @internal */
export const provideContext = Debug.dualWithTrace<
export const provideContext = dual<
<Env>(

@@ -899,25 +846,11 @@ env: Context.Context<Env>

2,
(trace) =>
<InErr, InElem, InDone, OutErr, OutElem, OutDone, Env>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
env: Context.Context<Env>
): Channel.Channel<never, InErr, InElem, InDone, OutErr, OutElem, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_PROVIDE
op.context = () => env
op.inner = self
return op.traced(trace)
}
)
/** @internal */
export const readOrFail = Debug.methodWithTrace((trace) =>
<In, E>(
error: E
): Channel.Channel<never, unknown, In, unknown, E, never, In> => {
<InErr, InElem, InDone, OutErr, OutElem, OutDone, Env>(
self: Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
env: Context.Context<Env>
): Channel.Channel<never, InErr, InElem, InDone, OutErr, OutElem, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_READ
op.more = succeed
op.done = new ContinuationKImpl(() => fail(error), () => fail(error))
return op.traced(trace)
op._tag = OpCodes.OP_PROVIDE
op.context = () => env
op.inner = self
return op
}

@@ -927,135 +860,130 @@ )

/** @internal */
export const readWith = Debug.methodWithTrace((trace, restore) =>
<
Env,
InErr,
InElem,
InDone,
OutErr,
OutElem,
OutDone,
Env2,
OutErr2,
OutElem2,
OutDone2,
Env3,
OutErr3,
OutElem3,
OutDone3
>(
input: (input: InElem) => Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
error: (error: InErr) => Channel.Channel<Env2, InErr, InElem, InDone, OutErr2, OutElem2, OutDone2>,
done: (done: InDone) => Channel.Channel<Env3, InErr, InElem, InDone, OutErr3, OutElem3, OutDone3>
): Channel.Channel<
Env | Env2 | Env3,
InErr,
InElem,
InDone,
OutErr | OutErr2 | OutErr3,
OutElem | OutElem2 | OutElem3,
OutDone | OutDone2 | OutDone3
> =>
readWithCause(
restore(input),
(cause) => Either.match(Cause.failureOrCause(cause), restore(error), failCause),
restore(done)
)
)
export const readOrFail = <In, E>(
error: E
): Channel.Channel<never, unknown, In, unknown, E, never, In> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_READ
op.more = succeed
op.done = new ContinuationKImpl(() => fail(error), () => fail(error))
return op
}
/** @internal */
export const readWithCause = Debug.methodWithTrace((trace, restore) =>
<
Env,
InErr,
InElem,
InDone,
OutErr,
OutElem,
OutDone,
Env2,
OutErr2,
OutElem2,
OutDone2,
Env3,
OutErr3,
OutElem3,
OutDone3
>(
input: (input: InElem) => Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
halt: (cause: Cause.Cause<InErr>) => Channel.Channel<Env2, InErr, InElem, InDone, OutErr2, OutElem2, OutDone2>,
done: (done: InDone) => Channel.Channel<Env3, InErr, InElem, InDone, OutErr3, OutElem3, OutDone3>
): Channel.Channel<
Env | Env2 | Env3,
InErr,
InElem,
InDone,
OutErr | OutErr2 | OutErr3,
OutElem | OutElem2 | OutElem3,
OutDone | OutDone2 | OutDone3
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_READ
op.more = restore(input)
op.done = new ContinuationKImpl(restore(done), restore(halt) as any)
return op.traced(trace)
}
)
export const readWith = <
Env,
InErr,
InElem,
InDone,
OutErr,
OutElem,
OutDone,
Env2,
OutErr2,
OutElem2,
OutDone2,
Env3,
OutErr3,
OutElem3,
OutDone3
>(
input: (input: InElem) => Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
error: (error: InErr) => Channel.Channel<Env2, InErr, InElem, InDone, OutErr2, OutElem2, OutDone2>,
done: (done: InDone) => Channel.Channel<Env3, InErr, InElem, InDone, OutErr3, OutElem3, OutDone3>
): Channel.Channel<
Env | Env2 | Env3,
InErr,
InElem,
InDone,
OutErr | OutErr2 | OutErr3,
OutElem | OutElem2 | OutElem3,
OutDone | OutDone2 | OutDone3
> =>
readWithCause(
input,
(cause) => Either.match(Cause.failureOrCause(cause), error, failCause),
done
)
/** @internal */
export const succeed = Debug.methodWithTrace((trace) =>
<A>(
value: A
): Channel.Channel<never, unknown, unknown, unknown, never, never, A> => sync(() => value).traced(trace)
)
export const readWithCause = <
Env,
InErr,
InElem,
InDone,
OutErr,
OutElem,
OutDone,
Env2,
OutErr2,
OutElem2,
OutDone2,
Env3,
OutErr3,
OutElem3,
OutDone3
>(
input: (input: InElem) => Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>,
halt: (cause: Cause.Cause<InErr>) => Channel.Channel<Env2, InErr, InElem, InDone, OutErr2, OutElem2, OutDone2>,
done: (done: InDone) => Channel.Channel<Env3, InErr, InElem, InDone, OutErr3, OutElem3, OutDone3>
): Channel.Channel<
Env | Env2 | Env3,
InErr,
InElem,
InDone,
OutErr | OutErr2 | OutErr3,
OutElem | OutElem2 | OutElem3,
OutDone | OutDone2 | OutDone3
> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_READ
op.more = input
op.done = new ContinuationKImpl(done, halt as any)
return op
}
/** @internal */
export const succeedNow = Debug.methodWithTrace((trace) =>
<OutDone>(
result: OutDone
): Channel.Channel<never, unknown, unknown, unknown, never, never, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_SUCCEED_NOW
op.terminal = result
return op.traced(trace)
}
)
export const succeed = <A>(
value: A
): Channel.Channel<never, unknown, unknown, unknown, never, never, A> => sync(() => value)
/** @internal */
export const suspend = Debug.methodWithTrace((trace, restore) =>
<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>(
evaluate: LazyArg<Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>>
): Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_SUSPEND
op.channel = restore(evaluate)
return op.traced(trace)
}
)
export const succeedNow = <OutDone>(
result: OutDone
): Channel.Channel<never, unknown, unknown, unknown, never, never, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_SUCCEED_NOW
op.terminal = result
return op
}
export const sync = Debug.methodWithTrace((trace, restore) =>
<OutDone>(
evaluate: LazyArg<OutDone>
): Channel.Channel<never, unknown, unknown, unknown, never, never, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_SUCCEED
op.evaluate = restore(evaluate)
return op.traced(trace)
}
)
/** @internal */
export const suspend = <Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>(
evaluate: LazyArg<Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone>>
): Channel.Channel<Env, InErr, InElem, InDone, OutErr, OutElem, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_SUSPEND
op.channel = evaluate
return op
}
export const sync = <OutDone>(
evaluate: LazyArg<OutDone>
): Channel.Channel<never, unknown, unknown, unknown, never, never, OutDone> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_SUCCEED
op.evaluate = evaluate
return op
}
/** @internal */
export const unit = Debug.methodWithTrace((trace) =>
(): Channel.Channel<never, unknown, unknown, unknown, never, never, void> => succeedNow(void 0).traced(trace)
)
export const unit = (): Channel.Channel<never, unknown, unknown, unknown, never, never, void> => succeedNow(void 0)
/** @internal */
export const write = Debug.methodWithTrace((trace) =>
<OutElem>(
out: OutElem
): Channel.Channel<never, unknown, unknown, unknown, never, OutElem, void> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_EMIT
op.out = out
return op.traced(trace)
}
)
export const write = <OutElem>(
out: OutElem
): Channel.Channel<never, unknown, unknown, unknown, never, OutElem, void> => {
const op = Object.create(proto)
op._tag = OpCodes.OP_EMIT
op.out = out
return op
}

6

src/internal/opCodes/channel.ts

@@ -84,7 +84,1 @@ /** @internal */

export type OP_SUSPEND = typeof OP_SUSPEND
/** @internal */
export const OP_TRACED = "Traced" as const
/** @internal */
export type OP_TRACED = typeof OP_TRACED

2

src/internal/stream/sinkEndReason.ts

@@ -27,5 +27,5 @@ /** @internal */

/** @internal */
export const SchedulEnd: SinkEndReason = { _tag: OP_SCHEDULE_END }
export const ScheduleEnd: SinkEndReason = { _tag: OP_SCHEDULE_END }
/** @internal */
export const UpstreamEnd: SinkEndReason = { _tag: OP_UPSTREAM_END }

5

src/Sink.ts

@@ -109,3 +109,6 @@ /**

*/
export const as = internal.as
export const as: {
<Z2>(z: Z2): <R, E, In, L, Z>(self: Sink<R, E, In, L, Z>) => Sink<R, E, In, L, Z2>
<R, E, In, L, Z, Z2>(self: Sink<R, E, In, L, Z>, z: Z2): Sink<R, E, In, L, Z2>
} = internal.as

@@ -112,0 +115,0 @@ /**

3

tsconfig.base.json

@@ -16,3 +16,4 @@ {

"lib": [
"ES2021"
"ES2021",
"DOM"
],

@@ -19,0 +20,0 @@ "sourceMap": true,

Channel.d.ts

Sorry, the diff of this file is too big to display

Channel.d.ts.map

Sorry, the diff of this file is not supported yet

Channel.js.map

Sorry, the diff of this file is not supported yet

internal/channel.d.ts.map

Sorry, the diff of this file is not supported yet

internal/channel.js.map

Sorry, the diff of this file is not supported yet

internal/channel/channelExecutor.d.ts.map

Sorry, the diff of this file is not supported yet

internal/channel/channelExecutor.js.map

Sorry, the diff of this file is not supported yet

internal/core.d.ts.map

Sorry, the diff of this file is not supported yet

internal/core.js.map

Sorry, the diff of this file is not supported yet

internal/opCodes/channel.js.map

Sorry, the diff of this file is not supported yet

internal/sink.js.map

Sorry, the diff of this file is not supported yet

internal/stream.d.ts.map

Sorry, the diff of this file is not supported yet

internal/stream.js

Sorry, the diff of this file is too big to display

internal/stream.js.map

Sorry, the diff of this file is not supported yet

internal/stream/sinkEndReason.js.map

Sorry, the diff of this file is not supported yet

mjs/Channel.mjs.map

Sorry, the diff of this file is not supported yet

mjs/internal/channel.mjs

Sorry, the diff of this file is not supported yet

mjs/internal/channel.mjs.map

Sorry, the diff of this file is not supported yet

mjs/internal/channel/channelExecutor.mjs

Sorry, the diff of this file is not supported yet

mjs/internal/channel/channelExecutor.mjs.map

Sorry, the diff of this file is not supported yet

mjs/internal/core.mjs

Sorry, the diff of this file is not supported yet

mjs/internal/core.mjs.map

Sorry, the diff of this file is not supported yet

mjs/internal/opCodes/channel.mjs

Sorry, the diff of this file is not supported yet

mjs/internal/opCodes/channel.mjs.map

Sorry, the diff of this file is not supported yet

mjs/internal/sink.mjs

Sorry, the diff of this file is not supported yet

mjs/internal/sink.mjs.map

Sorry, the diff of this file is not supported yet

mjs/internal/stream.mjs

Sorry, the diff of this file is not supported yet

mjs/internal/stream.mjs.map

Sorry, the diff of this file is not supported yet

mjs/internal/stream/sinkEndReason.mjs

Sorry, the diff of this file is not supported yet

mjs/internal/stream/sinkEndReason.mjs.map

Sorry, the diff of this file is not supported yet

mjs/Sink.mjs.map

Sorry, the diff of this file is not supported yet

mjs/Stream.mjs

Sorry, the diff of this file is not supported yet

mjs/Stream.mjs.map

Sorry, the diff of this file is not supported yet

Sink.d.ts.map

Sorry, the diff of this file is not supported yet

Sink.js.map

Sorry, the diff of this file is not supported yet

src/Channel.ts

Sorry, the diff of this file is too big to display

src/internal/channel.ts

Sorry, the diff of this file is too big to display

src/internal/sink.ts

Sorry, the diff of this file is too big to display

src/internal/stream.ts

Sorry, the diff of this file is too big to display

src/Stream.ts

Sorry, the diff of this file is too big to display

Stream.d.ts

Sorry, the diff of this file is too big to display

Stream.d.ts.map

Sorry, the diff of this file is not supported yet

Stream.js

Sorry, the diff of this file is too big to display

Stream.js.map

Sorry, the diff of this file is not supported yet

SocketSocket SOC 2 Logo

Product

  • Package Alerts
  • Integrations
  • Docs
  • Pricing
  • FAQ
  • Roadmap
  • Changelog

About

Packages

npm

Go

Maven

PyPI

Rubygems

Stay in touch

Get open source security insights delivered straight into your inbox.

  • Terms
  • Privacy
  • Security

Made with ⚡️ by Socket Inc