Stream

Everything in this package that moves data over time, such as standard input and output, a file read or written in pieces, a child program's pipes, a network connection, an HTTP body or a WebSocket message, is handed to the program as a stream. This module is what can be done with one: read from it, write to it, end it, and connect streams together.

A stream has two sides. Values go into its writable side and come out of its readable side, in the order they went in, unless a transformation changes them on the way. The program holds a handle to a side: a Readable to read from, or a Writable to write into. A stream made by fromList or by Stream.Transformation lives in memory. A stream that another module creates over a file, pipe or socket is handed out as one side only, and the resource is on the other side.

Each side of an in-memory stream buffers a limited number of values, so a writer cannot run arbitrarily far ahead of its reader. This is backpressure, and it is why there are two ways to write. write succeeds once the value has been accepted: taken from the writable side's buffer and passed on to the readable side (through the transformation, if there is one), or, for a stream over a resource, written to it. enqueue succeeds as soon as the value is in the writable side's buffer. On an in-memory stream both wait while that buffer is full.

A stream ends in one of three ways, and they are reported differently.

It is closed when its writer says no more values will come (closeWritable), or when the resource behind it has no more to give: the values already written can still be read, and after the last of them a read fails with Closed. A write or a close into a closed stream fails with Closed too.
It is cancelled when the program gives up on it, with a reason: cancelReadable, cancelWritable, or a custom transformation whose step returns Cancel. Buffered values are dropped, except the bytes readLine has kept, and reads, writes and closes fail with Cancelled and the reason, except reads after cancelReadable, which fail as cancelReadable describes.
It has failed when the resource behind it fails, such as a connection the peer reset or a file that cannot be read. Buffered values are dropped as for a cancel, and reads, writes and closes fail with Failed and a Failure.

Once a stream has been handed out, a failure is how a problem with its resource reaches the program. A connection reset, for example, is a read that fails with Failed, not an error of the module that made the connection; Socket.fromStreamFailure reads that failure back as a Socket.Error.

A side is locked while an operation holds it, such as a read waiting for a value, a write waiting for room in the buffer, or a pipe (pipeThrough, pipeTo), which holds the readable side of its source and the writable side of its destination until it finishes. Another operation that needs a locked side fails at once with Locked.

readLine reads a byte stream a line at a time. The bytes it reads past the end of a line are kept with the stream, and the next read, readString, readLine or readUntilClosed hands them out first, so line reads and chunk reads can be mixed. A pipe does not hand them out: bytes kept by readLine are not moved by pipeThrough or pipeTo. At the end of a stream they can be lost, as readLine describes.

Readable Streams

type alias Readable value =
    Readable value

The readable side of a stream, from which values of type value are read. One is made by fromList or pipeThrough, taken from a transformation with Stream.Transformation.readable, or handed out by another module of this package.

fromList : List a -> Task Error (Readable a)

Creates an in-memory Readable stream that delivers the values in the given List, in order, and is then closed.

The stream's buffer is made large enough for the whole list, so the Task does not wait for a reader.

read : Readable value -> Task Error value

Reads the next value off the stream, waiting until there is one.

Once the stream has been closed and every value written before the close has been read, or once the reading side has been cancelled while the stream had not been cancelled or failed in another way, the Task fails with Closed. If the stream has been cancelled from its writing side or by its transformation, it fails with Cancelled and the reason, and if its resource has failed, with Failed and the failure. While another read is waiting on the stream, or something else such as a pipe is consuming it, it fails with Locked.

Killing a task whose read is waiting (Process.kill) withdraws the read: the stream is no longer locked, the wait no longer keeps the program running, and no input is lost, since a value that arrives for the withdrawn read is kept for the next one.

readString : Readable Bytes -> Task Error String

Reads the next chunk of Bytes off the stream, as read does, and converts it to a String. If the chunk is not valid UTF-8, the stream is cancelled, with a reason saying so, and the Task fails with Cancelled; if by then another operation holds the reading side, the Task fails with Locked instead and the stream is not cancelled.

Each chunk is converted on its own, so a multi-byte character split across two chunks does not convert. readLine assembles lines across chunks, and Stream.Transformation.textDecoder decodes a whole byte stream however it is split.

readLine : Readable Bytes -> Task Error String

Reads the next line off a byte stream: everything up to the next newline (\n), decoded as UTF-8, without the newline and without a carriage return (\r) just before it.

countLines : Readable Bytes -> Int -> Task Error Int
countLines stdin n =
    Stream.readLine stdin
        |> Task.andThen (\_ -> countLines stdin (n + 1))
        |> Task.onError
            (\err ->
                case err of
                    Stream.Closed ->
                        Task.succeed n


                    _ ->
                        Task.fail err
            )

A line may be longer than the chunks the stream delivers. On the standard input stream, once it has been closed, a last line without a newline is returned as it is, and after that the Task fails with Closed. On other streams this is not guaranteed: when such a stream ends, the bytes readLine has kept can be lost, including whole lines of the last chunk of an in-memory stream. Bytes read past the newline are kept with the stream and handed out first by the next read, so readLine can be mixed with read and readString. If a line is not valid UTF-8, the stream is cancelled, with a reason saying so, and the Task fails with Cancelled, or with Locked if by then another operation holds the reading side. A cancel does not drop the kept bytes: after one, the next read still hands them out before it fails, and readLine still returns any complete lines among them. Other failures are those of read.

readUntilClosed :
    a -> b -> Result String b
    -> b
    -> Readable a
    -> Task Error b

Reads every value off the stream, folding each into an accumulated value with stepFn, starting from initial, and succeeds with the result once a read fails with Closed.

If stepFn returns an Err, its String is used as the reason to cancel the stream, and the Task fails with Cancelled and that reason; if by then another operation holds the reading side, the Task fails with Locked instead and the stream is not cancelled. Any other failure of a read fails the Task.

cancelReadable : String -> Readable value -> Task Error ()

Cancels the stream from its reading side, because the reader will not take any more values. The String is the reason, a human-readable explanation of what went wrong. Values in the stream's buffer are dropped, except the bytes readLine has kept, and a stream over a resource stops reading from it.

Once any kept bytes have been handed out, later reads fail with Closed, unless the stream had already been cancelled from its writing side or by its transformation, in which case they go on failing with Cancelled and that earlier reason, or its resource had failed, in which case they go on failing with Failed and that failure. If the writing side is still open, or a close of it is still in progress, writes into it that are still waiting, that close, and later writes fail with Cancelled and the given reason. While a read is waiting on the stream, or something else such as a pipe is consuming it, the Task fails with Locked and nothing is cancelled.

Writable Streams

type alias Writable value =
    Writable value

The writable side of a stream, into which values of type value are written. One is taken from a transformation with Stream.Transformation.writable, or handed out by another module of this package.

write : value -> Writable value -> Task Error (Writable value)

Writes a value into the stream, and succeeds once the value has been accepted: passed on to the stream's readable side (through the transformation, if there is one), or, for a stream over a resource, written to it. On an in-memory stream with no room for the value, this waits until a reader makes room.

Writing into a stream that has been closed fails with Closed, into one that has been cancelled with Cancelled and the reason, and into one whose resource has failed with Failed and the failure. While another write is waiting for room, or a pipe is writing into the stream, it fails with Locked.

writeString : String -> Writable Bytes -> Task Error (Writable Bytes)

Converts the String to UTF-8 Bytes and writes them into the stream, as write does.

A lone surrogate (half of a UTF-16 pair with no partner) is converted differently by the two backends: the JavaScript backend writes U+FFFD in its place, and the native backend writes the surrogate's own three bytes, which are not valid UTF-8 and which readString rejects.

writeLine : String -> Writable Bytes -> Task Error (Writable Bytes)

Appends a newline (\n) to the String, converts it to UTF-8 Bytes and writes them into the stream, as write does. A lone surrogate is converted as writeString describes.

enqueue : value -> Writable value -> Task Error (Writable value)

Queues a value to be written into the stream, and succeeds once it is in the writable side's buffer, or, for a stream over a resource, once the write has been started.

The difference from write is when the Task succeeds. It does not wait for a reader to take the value, so it succeeds even if nothing is reading, as long as the buffer has room (on an in-memory stream with a full buffer it waits for room). The price is that it cannot report what happens to the value afterwards: if the stream is cancelled, or the write to the resource fails, before the value is accepted, this Task has already succeeded.

Enqueueing into a stream that has been closed, cancelled or has failed fails as write does, and it fails with Locked in the same cases as write.

closeWritable : Writable value -> Task Error ()

Closes the stream: no more values will be written into it. The Task succeeds once every value written before the close has been accepted, and, for a stream over a resource, once the resource has been closed (the standard streams are never closed; the close only waits for their writes). Readers can still read the values in the buffer, and then get Closed.

Closing a stream that is already closed or being closed fails with Closed, and one that has been cancelled or has failed fails with Cancelled and the reason, or Failed and the failure. While a write is waiting for room, or a pipe is writing into the stream, it fails with Locked. If the stream is cancelled while the close waits, the Task fails with Cancelled and the reason, and if its resource fails, with Failed and the failure.

cancelWritable : String -> Writable value -> Task Error ()

Cancels the stream from its writing side, because something has gone wrong. The String is the reason, a human-readable explanation of what. Values in the stream's buffers are dropped, except the bytes readLine has kept.

On an in-memory stream, writes still in the buffer and a close still in progress fail with Cancelled and the given reason, and so do reads from the readable side, including one already waiting, once any kept bytes have been handed out. A stream over a resource stops writing to it. Cancelling a stream that is already closed or cancelled does nothing. While a write is waiting for room, or a pipe is writing into the stream, the Task fails with Locked and nothing is cancelled.

Logging

The logging functions write to a byte stream and ignore any failure. They are for output where there is nothing sensible to do when a write fails: if an error message cannot be written to the terminal, there is usually nowhere else to report that either.

log : Bytes -> Writable Bytes -> Task x ()

Writes Bytes into a byte stream as write does, and succeeds once the write has succeeded or failed, ignoring any failure.

logString : String -> Writable Bytes -> Task x ()

Writes a String into a byte stream as writeString does, ignoring any failure.

logLine : String -> Writable Bytes -> Task x ()

Writes a String followed by a newline into a byte stream as writeLine does, ignoring any failure.

Stream.logLine "Hello, world!" env.stdout

Error Handling

Stream knows only what any stream can go through. Why the resource behind a stream failed is known to the module that made the stream, and each such module reads a Failure back as its own error type, as Failure describes.

type Error
    = Closed
    | Locked
    | Cancelled String
    | Failed Failure

Why an operation on a stream failed.

Closed means that the stream has ended. A read gets it when the stream has ended normally: it was closed, and every value written before the close has been read. A read also gets Closed after the stream was cancelled from its reading side with cancelReadable, unless it had already been cancelled or failed in another way. A write or a close gets Closed when the stream has been closed, or when its transformation has ended it, as a custom transformation's Close does.

Cancelled means that the program cancelled the stream, and carries the reason it gave, a human-readable explanation: the reason given to cancelReadable or cancelWritable, to a custom transformation's Cancel, or returned as the Err of a readUntilClosed step. readString and readLine cancel a stream themselves, with a reason of their own, when the bytes are not valid UTF-8. A read gets Cancelled when the stream was cancelled from its writing side or by its transformation; a write or a close gets it when the stream has been cancelled; and pipeTo gets it as pipeTo describes.

Failed means that the resource behind the stream failed, for example a connection the peer reset, and carries a Failure describing it.

Locked means that the side of the stream the operation needs is held by another operation, such as a waiting read, a write waiting for room, or a pipe. A single read, write, close, cancel or pipe that fails with Locked has done nothing and can be tried again once the other has finished; a function that reads and then cancels or reads again, such as readString, readLine or readUntilClosed, may already have consumed values.

errorToString : Error -> String

Returns a human-readable description of an error: "Closed", "Locked", "Cancelled: " followed by the reason, or "Failed: " followed by failureToString of the failure.

type alias Failure =
    Failure

Why the resource behind a stream failed: a stream over a file, a pipe, a connection or an HTTP body that could not go on.

A Failure is opaque, because Stream knows nothing about the resources behind streams. The module that made the stream knows, and reads its own failures back as its own error type, giving Nothing for a failure some other module made:

Socket.fromStreamFailure, for the streams of a TCP or Unix domain connection, gives a Socket.Error, such as one whose kind is ConnectionReset;
Socket.Tls.fromStreamFailure, for the streams of a TLS connection, gives a Socket.Tls.Error, covering both TLS failures and failures of the connection under it;
FileSystem.fromStreamFailure, for file streams, gives a FileSystem.Error;
WebSocket.fromStreamFailure, for a WebSocket's message streams, gives a WebSocket.Error;
Http.Stream.fromStreamFailure, for streaming HTTP bodies, gives an Http.Error.

The failures of a child process's pipes, and of a decompressing transformation, have no such module: failureToString describes them.

Only the streams this package makes over a resource fail with a Failure. A stream made in Elm, such as a custom transformation, ends with Closed or is cancelled with a reason.

failureToString : Failure -> String

Returns a human-readable description of a failure, naming the problem, for example "read ECONNRESET" for a connection the peer reset.

Transformation Streams

type alias Transformation input output =
    Transformation input output

An in-memory stream whose two sides are both in the program's hands, and which may change the values on their way through: it can alter them, drop them, or produce several from one.

A Transformation input output accepts input values on its writable side and produces output values on its readable side; textEncoder, for example, makes a Transformation String Bytes.

Transformations are created with the functions in Stream.Transformation, and used with pipeThrough.

pipeThrough :
    Transformation input output
    -> Readable input
    -> Task Error (Readable output)

Starts moving every value of the Readable into the writable side of the Transformation, and succeeds at once with the transformation's readable side.

The pipe locks the source and the transformation's writable side until it finishes. When the source has closed and every value has been moved, the transformation's writable side is closed. If the source is cancelled from its writing side, the transformation is cancelled with the same reason, so reads from the result fail with Cancelled; if its resource fails, the transformation ends with the same failure, so reads from the result fail with Failed. If the transformation's writable side is cancelled, for example by cancelling the result, the source is cancelled with the same reason. These later failures are not reported by this Task, which has already succeeded.

If the source or the transformation's writable side is already locked, the Task fails with Locked.

awaitAndPipeThrough :
    Task Error (Transformation input output)
    -> Readable input
    -> Task Error (Readable output)

Waits for the Task that builds a Transformation, such as Stream.Transformation.gunzip, then pipes the Readable through it as pipeThrough does. A failure of the building Task fails this one.

pipeTo : Writable data -> Readable data -> Task Error ()

Moves every value of the Readable into the Writable, locking both until it finishes.

When the readable side has closed and every value has been moved, the writable side is closed, and the Task succeeds once that close has finished.

If the readable side is cancelled from its writing side, the writable side is cancelled with the same reason, and the Task fails with Cancelled and that reason. If the readable side's resource fails, the writable side ends with the same failure, and the Task fails with Failed and that failure.

If the writable side is cancelled (by its reader or by its transformation) or its resource fails, or it is already closed or being closed when the pipe starts, the readable side is cancelled from its reading side, and the Task fails with the writable side's error: Cancelled and the reason, Failed and the failure, or Closed for a closed one.

If either side is already locked, the Task fails with Locked.