asyncstreams

Unstable API.

Imports

asyncfutures, deques

Types

FutureStream[T] = ref object
  queue: Deque[T]
  finished: bool
  cb: proc () {...}{.closure, gcsafe.}
  error*: ref Exception

Special future that acts as a queue. Its API is still experimental and so is subject to change. Source Edit

Procs

proc newFutureStream[T](fromProc = "unspecified"): FutureStream[T]

Create a new FutureStream. This future's callback is activated when two events occur:

• New data is written into the future stream.
• The future stream is completed (this means that no more data will be written).

Specifying fromProc, which is a string specifying the name of the proc that this future belongs to, is a good habit as it helps with debugging.

Note: The API of FutureStream is still new and so has a higher likelihood of changing in the future.

Source Edit

proc complete[T](future: FutureStream[T])

Completes a FutureStream signalling the end of data. Source Edit

proc fail[T](future: FutureStream[T]; error: ref Exception)

Completes future with error. Source Edit

proc callback=[T](future: FutureStream[T];
                  cb: proc (future: FutureStream[T]) {...}{.closure, gcsafe.})

Sets the callback proc to be called when data was placed inside the future stream.

The callback is also called when the future is completed. So you should use finished to check whether data is available.

If the future stream already has data or is finished then cb will be called immediately.

Source Edit

proc finished[T](future: FutureStream[T]): bool

Check if a FutureStream is finished. true value means that no more data will be placed inside the stream and that there is no data waiting to be retrieved. Source Edit

proc failed[T](future: FutureStream[T]): bool

Determines whether future completed with an error. Source Edit

proc write[T](future: FutureStream[T]; value: T): Future[void]

Writes the specified value inside the specified future stream.

This will raise ValueError if future is finished.

Source Edit

proc read[T](future: FutureStream[T]): owned(Future[(bool, T)])

Returns a future that will complete when the FutureStream has data placed into it. The future will be completed with the oldest value stored inside the stream. The return value will also determine whether data was retrieved, false means that the future stream was completed and no data was retrieved.

This function will remove the data that was returned from the underlying FutureStream.

Source Edit

proc len[T](future: FutureStream[T]): int

Returns the amount of data pieces inside the stream. Source Edit