channel

from pyfect import channel

doubled = channel.from_iterable([1, 2, 3]).map(lambda n: n * 2)
fx.run_sync(doubled.run_collect())  # [2, 4, 6]

catch_if(refinement, f, *, or_else=None) -> Any

def catch_if(
    refinement: Callable[[Any], bool], f: Handler, *, or_else: Handler | None = None
) -> Any:
    """Recover from the typed errors ``refinement`` accepts and remove them
    from ``OutErr``. For a plain ``bool`` predicate, use the ``catch_if``
    method."""
    if or_else is None:
        return _CatchSome[Any, Any, Any, Any, Any, Any, Any, Any](refinement, f, None)
    return _CatchAll[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any](refinement, f, or_else)

catch_tag(tag, f, *, or_else=None) -> Any

def catch_tag(
    tag: type[Any] | tuple[type[Any], ...], f: Handler, *, or_else: Handler | None = None
) -> Any:
    """Recover from the errors that are instances of ``tag`` (a class or a
    tuple of up to three) and remove them from ``OutErr``."""
    predicate: Callable[[Any], bool] = lambda e: isinstance(e, tag)  # noqa: E731
    if or_else is None:
        return _CatchSome[Any, Any, Any, Any, Any, Any, Any, Any](predicate, f, None)
    return _CatchAll[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any](predicate, f, or_else)

catch_reason(tag, reason, f, *, or_else=None) -> Any

def catch_reason(
    tag: type[Any],
    reason: type[Any],
    f: ReasonHandler[AnyChannel],
    *,
    or_else: ReasonHandler[AnyChannel] | None = None,
) -> Any:
    """Switch to ``f(reason, error)`` when the channel fails with a ``tag``
    error whose ``reason`` is an instance of ``reason``; with ``or_else``,
    other ``tag`` errors switch to ``or_else(reason, error)``."""
    run = _reason_catch(tag, ((reason, f),), or_else)
    return (
        _CatchReason[Any, Any, Any, Any, Any, Any, Any](run)
        if or_else is None
        else _CatchReasonAll[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any](run)
    )

catch_reasons(tag, *cases, *, or_else=None) -> Any

def catch_reasons(
    tag: type[Any],
    *cases: tuple[type[Any], ReasonHandler[AnyChannel]],
    or_else: ReasonHandler[AnyChannel] | None = None,
) -> Any:
    """``catch_reason`` with a ``(reason_class, handler)`` pair per reason,
    tried in order."""
    run = _reason_catch(tag, cases, or_else)
    return (
        _CatchReason[Any, Any, Any, Any, Any, Any, Any](run)
        if or_else is None
        else _CatchReasonAll[Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any](run)
    )

unwrap_reason(tag) -> _UnwrapReason[T, T]

def unwrap_reason[T: HasReason[Any]](tag: type[T]) -> _UnwrapReason[T, T]:
    """Fail with the ``reason`` of a ``tag`` error instead of the error."""
    from pyfect.channel._constructors import fail

    return _UnwrapReason(_reason_catch(tag, (), lambda reason, _: fail(reason)))

from_pull(make) -> Channel[Any, Any, Any, Any, Any, Any, Any]

def from_pull(make: Effect[Any, Any, Any]) -> Channel[Any, Any, Any, Any, Any, Any, Any]:
    """A source whose pull ``make`` builds each time the channel runs."""
    require_effect(make, "make")
    return Channel(lambda _upstream, _scope: make)

from_transform(transform) -> Any

def from_transform(transform: Callable[[Any, CloseableScope], Effect[Any, Any, Any]]) -> Any:
    """A channel from ``transform(upstream, scope)``, an effect building
    the channel's pull from the upstream pull. Resources go in ``scope``,
    closed when the run ends. If building the pull fails, the pull fails
    with that cause."""

    channel: Channel[Any, Any, Any, Any, Any, Any, Any] = Channel(transform)
    return channel

to_transform(channel) -> Callable[[Pull[I, IE, ID], CloseableScope], Effect[Pull[O, E, D], Never, R]]

def to_transform[O, E, D, I, IE, ID, R](
    channel: ChannelBase[O, E, D, I, IE, ID, R],
) -> Callable[[Pull[I, IE, ID], CloseableScope], Effect[Pull[O, E, D], Never, R]]:
    """The function a channel is made of (see ``from_transform``)."""
    require_channel(channel, "channel")
    return channel._transform

transform_pull(channel, f) -> Channel[O2, EE, D2, I, IE, ID, R | RR]

def transform_pull[O, E, D, I, IE, ID, R, O2, D2 = None, EE = Never, RR = Never](
    channel: ChannelBase[O, E, D, I, IE, ID, R],
    f: Callable[[Pull[O, E, D], CloseableScope], Effect[Pull[O2, EE, D2, RR], EE, RR]],
) -> Channel[O2, EE, D2, I, IE, ID, R | RR]:
    """A channel whose pull is ``f(pull, scope)`` of ``channel``'s pull."""
    require_channel(channel, "channel")
    inner = channel._transform
    transformed: Channel[O2, EE, D2, I, IE, ID, R | RR] = Channel(
        lambda upstream, scope: inner(upstream, scope).flat_map(lambda pull: f(pull, scope))
    )
    return transformed

acquire_release(acquire, release) -> Channel[A, E, None, object, object, object, RR]

def acquire_release[A, E = Never, RR = Never](
    acquire: Effect[A, E, RR],
    release: Callable[[A, Exit[Any, Any]], Effect[object, Never, RR]],
) -> Channel[A, E, None, object, object, object, RR]:
    """Write the acquired resource, then end; ``release`` runs when the
    channel's scope closes."""
    return unwrap_effect(_acquire_release(acquire, release).map(succeed))

acquire_use_release(acquire, use, release) -> Channel[O, EE, D, I, IE, ID, RR]

def acquire_use_release[A, O, D, I, IE, ID, EE = Never, RR = Never](
    acquire: Effect[A, EE, RR],
    use: Callable[[A], Channel[O, EE, D, I, IE, ID, RR]],
    release: Callable[[A, Exit[D, EE]], Effect[object, Never, RR]],
) -> Channel[O, EE, D, I, IE, ID, RR]:
    """Acquire uninterruptibly, write ``use(resource)``'s items, and
    release with its exit as soon as it fails or ends (or the run ends)."""
    require_effect(acquire, "acquire")

    def build(
        upstream: AnyPull, scope: CloseableScope, forked: CloseableScope
    ) -> Effect[AnyPull, Any, Any]:
        resource: Option[A] = None

        def keep(value: A) -> A:
            nonlocal resource
            resource = Some(value)
            return value

        def finalizer(ex: Exit[Any, Any]) -> Effect[object, Never, Any]:
            return _Succeed(None) if resource is None else release(resource.value, ex)

        return (
            forked.add_finalizer(finalizer)
            .and_then(acquire.map(keep).uninterruptible())
            .flat_map(lambda value: use(value)._transform(upstream, scope))
        )

    return bracket(build)

die(defect) -> Channel[Never, Never, Never]

def die(defect: BaseException) -> Channel[Never, Never, Never]:
    """Fail with the defect ``defect``."""
    return fail_cause(die_cause(defect))

end(value) -> Channel[Never, Never, L]

def end[L](value: L) -> Channel[Never, Never, L]:
    """Write nothing and end with ``value``."""
    pull = end_with(value)
    return Channel(lambda _upstream, _scope: _Succeed(pull))

end_sync(thunk) -> Channel[Never, Never, L]

def end_sync[L](thunk: Callable[[], L]) -> Channel[Never, Never, L]:
    """Write nothing and end with ``thunk()``, computed when the channel
    runs."""
    return Channel(lambda _upstream, _scope: _Sync(lambda: end_with(thunk())))

fail(error) -> Channel[Never, E, Never]

def fail[E](error: E) -> Channel[Never, E, Never]:
    """Fail with ``error``."""
    return fail_cause(fail_cause_of(error))

fail_cause(cause) -> Channel[Never, E, Never]

def fail_cause[E](cause: Cause[E]) -> Channel[Never, E, Never]:
    """Fail with ``cause``."""
    pull = _FailCause(cause)
    return Channel(lambda _upstream, _scope: _Succeed(pull))

fail_cause_sync(thunk) -> Channel[Never, E, Never]

def fail_cause_sync[E](thunk: Callable[[], Cause[E]]) -> Channel[Never, E, Never]:
    """Fail with ``thunk()``, computed when the channel runs."""
    return Channel(lambda _upstream, _scope: _Sync(lambda: _FailCause(thunk())))

fail_sync(thunk) -> Channel[Never, E, Never]

def fail_sync[E](thunk: Callable[[], E]) -> Channel[Never, E, Never]:
    """Fail with ``thunk()``, computed when the channel runs."""
    return Channel(lambda _upstream, _scope: _Sync(lambda: _FailCause(fail_cause_of(thunk()))))

from_effect(eff) -> Channel[A, E, None, object, object, object, R]

def from_effect[A, E = Never, R = Never](
    eff: Effect[A, E, R],
) -> Channel[A, E, None, object, object, object, R]:
    """Write what ``eff`` succeeds with, then end."""
    require_effect(eff, "eff")

    def make() -> AnyPull:
        ran = False

        def pull() -> AnyPull:
            nonlocal ran
            if ran:
                return ENDED
            ran = True
            return eff

        return _Suspend(pull)

    return Channel(lambda _upstream, _scope: _Sync(make))

from_effect_done(eff) -> Channel[Never, E, A, object, object, object, R]

def from_effect_done[A, E = Never, R = Never](
    eff: Effect[A, E, R],
) -> Channel[Never, E, A, object, object, object, R]:
    """Write nothing and end with what ``eff`` succeeds with."""
    require_effect(eff, "eff")
    pull = eff.flat_map(end_with)
    return Channel(lambda _upstream, _scope: _Succeed(pull))

from_effect_drain(eff) -> Channel[Never, E, None, object, object, object, R]

def from_effect_drain[A, E = Never, R = Never](
    eff: Effect[A, E, R],
) -> Channel[Never, E, None, object, object, object, R]:
    """Run ``eff`` when the channel starts, write nothing, and end."""
    require_effect(eff, "eff")
    drained: Channel[Never, E, None, object, object, object, R] = Channel(
        lambda _upstream, _scope: eff.as_(ENDED)
    )
    return drained

from_effect_take(eff) -> Channel[tuple[A, ...], EE, L, object, object, object, RR]

def from_effect_take[A, L = None, EE = Never, RR = Never](
    eff: Effect[Take[A, EE, L], EE, RR],
) -> Channel[tuple[A, ...], EE, L, object, object, object, RR]:
    """Pull by running ``eff`` and replaying the ``Take`` it succeeds with."""
    require_effect(eff, "eff")
    pull = eff.flat_map(take_to_pull)
    return Channel(lambda _upstream, _scope: _Succeed(pull))

from_iterable(iterable) -> Channel[Any, Never, Any]

def from_iterable(iterable: Iterable[Any]) -> Channel[Any, Never, Any]:
    """Write the items of ``iterable``, iterating it anew each run (a
    generator or iterator can only be read once)."""
    return from_iterator(lambda: iter(iterable))

from_iterable_array(iterable, chunk_size=DEFAULT_CHUNK_SIZE) -> Channel[tuple[Any, ...], Never, Any]

def from_iterable_array(
    iterable: Iterable[Any], chunk_size: int = DEFAULT_CHUNK_SIZE
) -> Channel[tuple[Any, ...], Never, Any]:
    """``from_iterable``, writing the items as non-empty tuples of up to
    ``chunk_size`` (at least 1)."""
    return from_iterator_array(lambda: iter(iterable), chunk_size)

from_iterator(make) -> Channel[Any, Never, Any]

def from_iterator(make: Callable[[], Iterator[Any]]) -> Channel[Any, Never, Any]:
    """Write the items of the iterator ``make()`` returns each run. A
    generator's return value is what the channel ends with; an exception
    from the iterator is a defect."""
    return Channel(lambda _upstream, _scope: _Sync(lambda: IteratorPull(make())))

from_iterator_array(make, chunk_size=DEFAULT_CHUNK_SIZE) -> Channel[tuple[Any, ...], Never, Any]

def from_iterator_array(
    make: Callable[[], Iterator[Any]], chunk_size: int = DEFAULT_CHUNK_SIZE
) -> Channel[tuple[Any, ...], Never, Any]:
    """``from_iterator``, writing the items as non-empty tuples of up to
    ``chunk_size`` (at least 1)."""
    size = max(1, chunk_size)
    return Channel(lambda _upstream, _scope: _Sync(lambda: ChunkPull(make(), size)))

from_transform_bracket(f) -> Any

def from_transform_bracket(
    f: Callable[[Any, CloseableScope, CloseableScope], Effect[Any, Any, Any]],
) -> Any:
    """``from_transform`` with a third argument: a child of the channel's
    scope, closed as soon as the channel fails or ends (with the end's
    leftover as a success), so resources can go before the run ends."""
    return bracket(f)

identity() -> Channel[A, E, D, A, E, D]

def identity[A, E = Never, D = None]() -> Channel[A, E, D, A, E, D]:
    """Write what upstream writes; fail and end as it does."""
    return Channel(lambda upstream, _scope: _Succeed(upstream))

succeed(value) -> Channel[A]

def succeed[A](value: A) -> Channel[A]:
    """Write ``value``, then end."""
    return from_effect(_Succeed(value))

suspend(thunk) -> Channel[O, E, D, I, IE, ID, R]

def suspend[O, E, D, I, IE, ID, R](
    thunk: Callable[[], Channel[O, E, D, I, IE, ID, R]],
) -> Channel[O, E, D, I, IE, ID, R]:
    """The channel ``thunk()`` returns, built each time it runs."""
    return Channel(lambda upstream, scope: thunk()._transform(upstream, scope))

sync(thunk) -> Channel[A]

def sync[A](thunk: Callable[[], A]) -> Channel[A]:
    """Write ``thunk()``, computed when the channel runs, then end."""
    return from_effect(_Sync(thunk))

unwrap(eff) -> _AnyChannel

def unwrap(eff: Effect[_AnyChannel, Any, Any]) -> _AnyChannel:
    """The channel ``eff`` succeeds with, built on the first pull. ``eff``
    runs with the channel's ``Scope``, so its resources live as long as
    the channel's."""
    require_effect(eff, "eff")
    return unwrap_effect(eff)

provide(source, /, *, local=False) -> _ProvideLayer[Any, Any, Any] | _ProvideContext[Any]

def provide(
    source: Layer[Any, Any, Any] | Context[Any], /, *, local: bool = False
) -> _ProvideLayer[Any, Any, Any] | _ProvideContext[Any]:
    """Provide a context's services, or build a layer each time the channel
    runs (its resources released when the run ends) and provide what it
    builds. Layers an enclosing ``provide`` built are reused unless
    ``local``."""
    if isinstance(source, Context):
        return _ProvideContext(source)
    require_layer(source, "source", "a Layer or a Context")
    return _ProvideLayer(source, local)

provide_context(ctx) -> _ProvideContext[RO]

def provide_context[RO](ctx: Context[RO]) -> _ProvideContext[RO]:
    """Provide every service of ``ctx`` to a channel and remove them from
    ``Env``."""
    return _ProvideContext(ctx)

provide_service(tag, service) -> _ProvideService[T]

def provide_service[S, T](tag: TagLike[S, T], service: S) -> _ProvideService[T]:
    """Provide one service to a channel and remove it from ``Env``::

    channel.pipe(provide_service(Database, InMemoryDb()))
    """
    return _ProvideService(tag_class(tag), service)

provide_service_effect(tag, acquire) -> _ProvideServiceEffect[T, E2, R2]

def provide_service_effect[S, T, E2 = Never, R2 = Never](
    tag: TagLike[S, T], acquire: Effect[S, E2, R2]
) -> _ProvideServiceEffect[T, E2, R2]:
    """Run ``acquire`` each time the channel runs and provide its value as
    ``tag``'s service, removing ``tag`` from ``Env``."""
    require_effect(acquire, "acquire")
    return _ProvideServiceEffect(tag_class(tag), acquire)

scoped(channel) -> Channel_

def scoped(channel: AnyChannel) -> Channel_:
    """Give the channel a ``Scope`` of its own, closed as soon as it fails
    or ends (or its run ends), and remove ``Scope`` from ``Env``."""

    def build(upstream: AnyPull, scope: CloseableScope, forked: CloseableScope) -> Effect[AnyPull]:
        return _locally(channel, lambda ctx: ctx.add(Scope, forked))._transform(upstream, scope)

    return bracket(build)

update_context(f) -> _UpdateContext[R2, RO]

def update_context[R2, RO](f: Callable[[Context[R2]], Context[RO]]) -> _UpdateContext[R2, RO]:
    """Build the channel with the services ``f(context)`` instead of
    ``context``, so it needs ``R2`` instead of what it needed."""
    return _UpdateContext(f)

callback(f, *, buffer_size=None, strategy='suspend') -> Source[Any, Any, Any]

def callback(
    f: Callable[[Queue[Any, Any]], Effect[object, Any, Any]],
    *,
    buffer_size: int | None = None,
    strategy: Strategy = "suspend",
) -> Source[Any, Any, Any]:
    """A source of what ``f`` offers to a queue, run on a fiber when the
    channel starts, with the channel's ``Scope``. It ends when the queue
    ends and fails with ``f``'s failure or the queue's."""
    channel: Source[Any, Any, Any] = from_transform(
        lambda _upstream, scope: _async_queue(scope, f, buffer_size, strategy).map(
            lambda queue: queue.take()
        )
    )
    return channel

callback_array(f, *, buffer_size=None, strategy='suspend') -> Source[Any, Any, Any]

def callback_array(
    f: Callable[[Queue[Any, Any]], Effect[object, Any, Any]],
    *,
    buffer_size: int | None = None,
    strategy: Strategy = "suspend",
) -> Source[Any, Any, Any]:
    """``callback``, writing every item waiting in the queue at once."""
    channel: Source[Any, Any, Any] = from_transform(
        lambda _upstream, scope: _async_queue(scope, f, buffer_size, strategy).map(
            lambda queue: queue.take_all().map(tuple)
        )
    )
    return channel

from_async_iterable(iterable, on_error) -> Source[A, E]

def from_async_iterable[A, E](
    iterable: AsyncIterable[A], on_error: Callable[[Exception], E]
) -> Source[A, E]:
    """A source of what ``iterable`` yields, failing with
    ``on_error(exception)`` if it raises. The iterator is closed when the
    run ends: awaited if idle, else without waiting for the pending
    ``__anext__``. Needs an asyncio loop (``run_async``)."""

    def transform(_upstream: AnyPull, scope: CloseableScope) -> Effect[AnyPull]:
        iterator: AsyncIterator[A] = aiter(iterable)
        pulling = False

        async def next_item() -> tuple[bool, A | None]:
            nonlocal pulling
            pulling = True
            try:
                return False, await anext(iterator)
            except StopAsyncIteration:
                return True, None
            finally:
                pulling = False

        def close() -> Effect[None, Any, Any]:
            aclose = getattr(iterator, "aclose", None)
            if aclose is None:
                return _Succeed(None)
            if pulling:
                return _Sync(lambda: _discard_task(aclose))
            return try_async(aclose).ignore()

        def written(result: tuple[bool, A | None]) -> AnyPull:
            ended, item = result
            return end_with(None) if ended else _Succeed(item)

        pull = try_async(next_item, on_error).flat_map(written)
        return scope.add_finalizer(lambda _: _Suspend(close)).as_(pull)

    channel: Source[A, E] = from_transform(transform)
    return channel

from_async_iterable_array(iterable, on_error) -> Source[tuple[A, ...], E]

def from_async_iterable_array[A, E](
    iterable: AsyncIterable[A], on_error: Callable[[Exception], E]
) -> Source[tuple[A, ...], E]:
    """``from_async_iterable``, writing each item as a 1-tuple."""
    return from_async_iterable(iterable, on_error).map(lambda item: (item,))

from_pubsub(pubsub) -> Source[A]

def from_pubsub[A](pubsub: PubSub[A]) -> Source[A]:
    """A source of what is published to ``pubsub`` once the channel runs."""
    channel: Source[A] = unwrap_effect(pubsub.subscribe().map(from_subscription))
    return channel

from_pubsub_array(pubsub) -> Source[tuple[A, ...]]

def from_pubsub_array[A](pubsub: PubSub[A]) -> Source[tuple[A, ...]]:
    """``from_pubsub``, writing every message waiting at once."""
    channel: Source[tuple[A, ...]] = unwrap_effect(pubsub.subscribe().map(from_subscription_array))
    return channel

from_pubsub_take(pubsub) -> Channel[tuple[A, ...], E, D, object, object, object]

def from_pubsub_take[A, E, D](
    pubsub: PubSub[Take[A, E, D]],
) -> Channel[tuple[A, ...], E, D, object, object, object]:
    """A source replaying the ``Take`` values published to ``pubsub``."""
    channel: Channel[tuple[A, ...], E, D, object, object, object] = unwrap_effect(
        pubsub.subscribe().map(lambda sub: from_effect_take(sub.take()))
    )
    return channel

from_queue(queue) -> Source[Any, Any]

def from_queue(queue: Dequeue[Any, Any]) -> Source[Any, Any]:
    """A source taking from ``queue`` until it ends (or fails)."""
    return from_pull(_Succeed(queue.take()))

from_queue_array(queue) -> Source[tuple[Any, ...], Any]

def from_queue_array(queue: Dequeue[Any, Any]) -> Source[tuple[Any, ...], Any]:
    """``from_queue``, writing every item waiting at once."""
    return from_pull(_Succeed(queue.take_all().map(tuple)))

from_schedule(schedule) -> Channel[O, E, O, object, object, object, R]

def from_schedule[O, E = Never, R = Never](
    schedule: Schedule[O, Any, E, R],
) -> Channel[O, E, O, object, object, object, R]:
    """Write each output of ``schedule`` after its delay; end with the
    final output when it stops."""

    def emit(step: Metadata[Any, Any] | Stop[Any]) -> AnyPull:
        return end_with(step.output) if isinstance(step, Stop) else _Succeed(step.output)

    def with_step(step: Callable[[Any], Effect[Any, Any, Any]]) -> AnyPull:
        return step(None).flat_map(emit)

    return from_pull(to_step_with_metadata(schedule).map(with_step))

from_subscription(subscription) -> Source[A]

def from_subscription[A](subscription: Subscription[A]) -> Source[A]:
    """A source taking from ``subscription``; it ends when the pubsub shuts
    down."""
    take: Effect[A, Done] = subscription.take().catch_cause(_end_on_interrupt)
    return from_pull(_Succeed(take))

from_subscription_array(subscription) -> Source[tuple[A, ...]]

def from_subscription_array[A](subscription: Subscription[A]) -> Source[tuple[A, ...]]:
    """``from_subscription``, writing every message waiting at once."""
    take: Effect[tuple[A, ...], Done] = (
        subscription.take_all().map(tuple).catch_cause(_end_on_interrupt)
    )
    return from_pull(_Succeed(take))

decode_text(encoding='utf-8', *, fatal=False, ignore_bom=False) -> Channel[tuple[str, ...], E, D, tuple[bytes, ...], E, D]

def decode_text[E = Never, D = None](
    encoding: str = "utf-8", *, fatal: bool = False, ignore_bom: bool = False
) -> Channel[tuple[str, ...], E, D, tuple[bytes, ...], E, D]:
    """Decode chunks of bytes, keeping a character split across chunks
    whole. Bad bytes become U+FFFD, or a defect with ``fatal``. A UTF-8
    byte-order mark is dropped unless ``ignore_bom`` (WHATWG
    ``TextDecoder``)."""

    def transform(upstream: AnyPull, _scope: CloseableScope) -> Effect[AnyPull]:
        def build() -> AnyPull:
            decoder = _decoder(encoding, fatal, ignore_bom)
            pull: AnyPull

            def decoded(chunks: tuple[bytes, ...]) -> tuple[str, ...]:
                return tuple(decoder.decode(chunk) for chunk in chunks)

            def flush(leftover: Any) -> AnyPull:
                nonlocal pull
                pull = end_with(leftover)
                last = decoder.decode(b"", final=True)
                return _Succeed((last,)) if last else pull

            pull = catch_done(upstream.map(decoded), flush)
            return _Suspend(lambda: pull)

        return _Sync(build)

    channel: Channel[tuple[str, ...], E, D, tuple[bytes, ...], E, D] = from_transform(transform)
    return channel

encode_text() -> Channel[tuple[bytes, ...], E, D, tuple[str, ...], E, D]

def encode_text[E = Never, D = None]() -> Channel[tuple[bytes, ...], E, D, tuple[str, ...], E, D]:
    """Encode chunks of text as UTF-8."""

    def encoded(chunks: tuple[str, ...]) -> tuple[bytes, ...]:
        return tuple(chunk.encode() for chunk in chunks)

    channel: Channel[tuple[bytes, ...], E, D, tuple[str, ...], E, D] = from_transform(
        lambda upstream, _scope: _Succeed(upstream.map(encoded))
    )
    return channel

split_lines() -> Channel[tuple[str, ...], E, D, tuple[str, ...], E, D]

def split_lines[E = Never, D = None]() -> Channel[tuple[str, ...], E, D, tuple[str, ...], E, D]:
    """Split chunks of text into lines, without their line breaks; the
    text after the last break is the last line."""

    def transform(upstream: AnyPull, _scope: CloseableScope) -> Effect[AnyPull]:
        def build() -> AnyPull:
            lines = _LineSplitter()
            ended: list[Any] = []

            def written(chunks: tuple[str, ...]) -> AnyPull:
                split = lines.split(chunks)
                return _Succeed(split) if split else pull_or_flush

            def done(leftover: Any) -> AnyPull:
                ended.append(leftover)
                last = lines.flush()
                return _Succeed((last,)) if last else end_with(leftover)

            def failed(cause: Cause[Any]) -> AnyPull:
                return _FailCause(cause)

            def next_pull() -> AnyPull:
                if ended:
                    return end_with(ended[0])
                return match_effect(upstream, on_success=written, on_failure=failed, on_done=done)

            pull_or_flush = _Suspend(next_pull)
            return pull_or_flush

        return _Sync(build)

    channel: Channel[tuple[str, ...], E, D, tuple[str, ...], E, D] = from_transform(transform)
    return channel