import { createQueue , Subscription } from '@effection/subscription' ; import { Operation , Task , Resource , spawn } from '@effection/core' ; import { DeepPartial , matcher } from './match' ; import { createBuffer } from './buffer' ; /**
 * A callback operation to  createStream . This is an operation which takes
 * a `publish` function as an argument. Each time the publish function is
 * invoked with a value, the stream emits that value.
 *
 *  typeParam typeParam export type StreamCallback < T , TReturn > = ( publish : ( value : T ) => void ) => Operation < TReturn > ; typeParam typeParam export interface Stream < T , TReturn = undefined > extends Resource < Subscription < T , TReturn > > { param predicate typeParam filter < R extends T > ( predicate ) : Stream < R , TReturn > ; filter ( predicate ) : Stream < T , TReturn > ; filter ( predicate ) : Stream < T , TReturn > ; param reference match ( reference ) : Stream < T , TReturn > ; param search grep ( search ) : Stream < T , TReturn > ; typeParam param mapper map < R > ( mapper ) : Stream < R , TReturn > ; /**
     * Returns the first item in the stream, or `undefined` if the stream
     * closes before another item is emitted.
     */ first ( ) : Operation < T | undefined > ; /**
     * Returns the first item in the stream, or throws an error if the stream
     * closes before before another item is emitted.
     */ expect ( ) : Operation < T > ; /**
     * Iterates all items of the stream, returns the value the stream closes
     * with. The given callback is invoked for each item in the stream. The
     * callback can be either a regular function which returns `undefined`, or an
     * operation.
     *
     * ### Example
     *
     * Iterating with a regular function:
     *
     * ```typescript
     * let comments = createStream(function*(publish) {
     *   publish({ text: "hello", state: "published" });
     *   publish({ text: "world", state: "draft" });
     *   return 2;
     * });
     *
     * let result = yield comments.forEach((item) => {
     *   console.log(item);
     * });
     *
     * console.log(result) // => 2
     * ```
     *
     * Iterating with an operation:
     *
     * ```typescript
     * let comments = createStream(function*(publish) {
     *   publish({ text: "hello", state: "published" });
     *   publish({ text: "world", state: "draft" });
     * });
     *
     * yield comments.forEach(function*(value) {
     *   yield createComment(value);
     * });
     * ```
     *
     */ forEach ( visit ) : Operation < TReturn > ; /**
     * Wait for the stream to close and return the value it closes with.
     *
     * ### Example
     *
     * ```typescript
     * let comments = createStream(function*(publish) {
     *   publish({ text: "hello", state: "published" });
     *   publish({ text: "world", state: "draft" });
     *   return 2;
     * });
     *
     * let result = yield comments.join();
     *
     * console.log(result) // => 2
     * ```
     */ join ( ) : Operation < TReturn > ; /**
     * Block until the stream finishes, and return a synchronous iterator over
     * all of the emitted values.
     *
     * ### Example
     *
     * ```typescript
     * let comments = createStream(function*(publish) {
     *   publish("hello");
     *   publish("world");
     *   return 2;
     * });
     *
     * let result = yield comments.collect();
     *
     * console.log(result.next()); // => { done: false, value: "hello" }
     * console.log(result.next()); // => { done: false, value: "world" }
     * console.log(result.next()); // => { done: true, value: 2 }
     * ```
     */ collect ( ) : Operation < Iterator < T , TReturn > > ; /**
     * Block until the stream finishes and return an array of all of the items
     * that were emitted.
     *
     * ### Example
     *
     * ```typescript
     * let comments = createStream(function*(publish) {
     *   publish("hello");
     *   publish("world");
     *   return 2;
     * });
     *
     * let result = yield comments.toArray();
     *
     * console.log(result); // => ["hello", "world"]
     * ```
     */ toArray ( ) : Operation < T [ ] > ; hidden subscribe ( scope ) : Subscription < T , TReturn > ; param limit toBuffer ( limit ? ) : Resource < Iterable < T > > ; param limit buffered ( limit ? ) : Resource < Stream < T , TReturn > > ; } /**
 * Create a new stream of values. The given callback operation is run each time
 * someone subscribes to the stream, so it may be invoked multiple times. The
 * stream emits a value each time the `publish` function passed as an argument
 * to the callback is invoked. The stream is closed when the callback function
 * returns, and the return value of the callback is the value the stream closes
 * with.
 *
 * ### Example
 *
 * ```typescript
 * import { main, createStream, sleep, all } from 'effection';
 *
 * main(function*() {
 *   let stream = createStream<number, string>(function*(publish) {
 *     yield sleep(100);
 *     publish(1);
 *     yield sleep(100);
 *     publish(2);
 *     return "I'm done";
 *   });
 *
 *   // we can subscribe multiple times
 *   let result = yield all([
 *     stream.forEach((value) => { console.log(value) }); // => '1', '2'
 *     stream.forEach((value) => { console.log(value) }); // => '1', '2'
 *   ]);
 *   console.log(result) // => ["I'm done", "I'm done"]
 * });
 * ```
 *
 * See [the guide on Streams and Subscriptions](https://frontside.com/effection/docs/guides/collections) for more details.
 */ export function createStream < T , TReturn = undefined > ( callback , name = 'stream' ) { let subscribe = ( task ) => { let queue = createQueue < T , TReturn > ( name ) ; task . run ( function * ( ) { let result = yield callback ( queue . send ) ; queue . closeWith ( result ) ; } , { labels : { name : 'publisher' , expand : false } } ) ; return queue . subscription ; } ; function filter < R extends T > ( predicate ) ; function filter ( predicate ) ; function filter ( predicate ) { return createStream ( ( publish ) => { return stream . forEach ( ( value ) => function * ( ) { if ( predicate ( value ) ) { publish ( value ) ; } } ) ; } , `${ name }.filter()` ) ; } let stream = { subscribe , filter , * init ( task ) { return subscribe ( task ) ; } , match ( reference ) { return stream . filter ( matcher ( reference ) ) ; } , grep ( search ) { if ( typeof ( search ) === 'string' ) { return stream . filter ( ( value ) => String ( value ) . includes ( search ) ) ; } else { return stream . filter ( ( value ) => ! ! String ( value ) . match ( search ) ) ; } } , map < R > ( mapper ) { return createStream ( ( publish ) => { return stream . forEach ( ( value ) => function * ( ) { publish ( mapper ( value ) ) ; } ) ; } , `${ name }.map()` ) ; } , first ( ) { return task => subscribe ( task ) . first ( ) ; } , expect ( ) { return task => subscribe ( task ) . expect ( ) ; } , forEach ( visit ) { return task => subscribe ( task ) . forEach ( visit ) ; } , join ( ) { return task => subscribe ( task ) . join ( ) ; } , collect ( ) { return task => subscribe ( task ) . collect ( ) ; } , toArray ( ) { return task => subscribe ( task ) . toArray ( ) ; } , toBuffer ( limit = Infinity ) { return { name : `${ name }.buffer(${ limit })` , * init ( ) { let buffer = createBuffer < T > ( limit ) ; yield spawn ( stream . forEach ( ( value ) => { buffer . push ( value ) ; } ) ) ; return buffer ; } } ; } , buffered ( limit = Infinity ) { return { * init ( ) { let buffer = yield stream . toBuffer ( limit ) ; return createStream < T , TReturn > ( ( publish ) => function * ( ) { for ( let value of buffer ) { publish ( value ) ; } return yield stream . forEach ( publish ) ; } , `${ name }.buffered(${ limit })` ) ; } } ; } } ; return stream ; }