commit c160e1d75fe240eabcb605bcaa4ee15622a4eb64 · kitten.sh/wonka

.changeset/yellow-hounds-heal.md

···

       1
       +
       ---

     

       2
       +
       'wonka': patch

     

       3
       +
       ---

     

       4
       +
       

     

       5
       +
       Improve compatibility of `fromAsyncIterable` and `toAsyncIterable`. The `toAsyncIterable` will now output an object that's both an `AsyncIterator` and an `AsyncIterable`. Both helpers will now use a polyfill for `Symbol.asyncIterator` to improve compatibility with the Hermes engine and Babel transpilation.

+10 -7

src/__tests__/sinks.test.ts

···

       244
        
           };

     

       245
        
       

     

       246
        
           const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();

     

       0
        
       
     

       247
        
       

     

       0
        
       
     

       0
        
       
     

       248
        
           expect(pulls).toBe(1);

     

       249
       -
           sink!(push(0));

     

       250
       -
           expect(await asyncIterator.next()).toEqual({ value: 0, done: false });

     

       251
       -
           expect(pulls).toBe(2);

     

       252
        
       

     

       253
        
           sink!(push(1));

     

       254
        
           expect(await asyncIterator.next()).toEqual({ value: 1, done: false });

     

       255
       -
           expect(pulls).toBe(3);

     

       256
        
       

     

       257
        
           sink!(SignalKind.End);

     

       258
        
           expect(await asyncIterator.next()).toEqual({ done: true });

     

       259
       -
           expect(pulls).toBe(3);

     

       260
        
         });

     

       261
        
       

     

       262
        
         it('buffers actively pushed values', async () => {

     
···

       273
        
           };

     

       274
        
       

     

       275
        
           const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();

     

       0
        
       
     

       276
        
       

     

       277
        
           sink!(push(0));

     

       278
        
           sink!(push(1));

     

       279
        
           sink!(SignalKind.End);

     

       280
        
       

     

       281
        
           expect(pulls).toBe(1);

     

       282
       -
           expect(await asyncIterator.next()).toEqual({ value: 0, done: false });

     

       283
        
           expect(await asyncIterator.next()).toEqual({ value: 1, done: false });

     

       284
        
           expect(await asyncIterator.next()).toEqual({ done: true });

     

       285
        
         });

     
···

       298
        
           };

     

       299
        
       

     

       300
        
           const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();

     

       0
        
       
     

       301
        
           expect(pulls).toBe(1);

     

       302
        
       

     

       303
        
           let resolved = false;

     
···

       330
        
           };

     

       331
        
       

     

       332
        
           const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();

     

       0
        
       
     

       333
        
       

     

       334
        
           sink!(push(0));

     

       335
       -
           expect(await asyncIterator.next()).toEqual({ value: 0, done: false });

     

       336
        
           expect(await asyncIterator.return!()).toEqual({ done: true });

     

       337
        
       

     

       338
        
           sink!(push(1));

···

       244
        
           };

     

       245
        
       

     

       246
        
           const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();

     

       247
       +
           const next$ = asyncIterator.next();

     

       248
        
       

     

       249
       +
           sink!(push(0));

     

       250
       +
           expect(await next$).toEqual({ value: 0, done: false });

     

       251
        
           expect(pulls).toBe(1);

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       252
        
       

     

       253
        
           sink!(push(1));

     

       254
        
           expect(await asyncIterator.next()).toEqual({ value: 1, done: false });

     

       255
       +
           expect(pulls).toBe(2);

     

       256
        
       

     

       257
        
           sink!(SignalKind.End);

     

       258
        
           expect(await asyncIterator.next()).toEqual({ done: true });

     

       259
       +
           expect(pulls).toBe(2);

     

       260
        
         });

     

       261
        
       

     

       262
        
         it('buffers actively pushed values', async () => {

     
···

       273
        
           };

     

       274
        
       

     

       275
        
           const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();

     

       276
       +
           const next$ = asyncIterator.next();

     

       277
        
       

     

       278
        
           sink!(push(0));

     

       279
        
           sink!(push(1));

     

       280
        
           sink!(SignalKind.End);

     

       281
        
       

     

       282
        
           expect(pulls).toBe(1);

     

       283
       +
           expect(await next$).toEqual({ value: 0, done: false });

     

       284
        
           expect(await asyncIterator.next()).toEqual({ value: 1, done: false });

     

       285
        
           expect(await asyncIterator.next()).toEqual({ done: true });

     

       286
        
         });

     
···

       299
        
           };

     

       300
        
       

     

       301
        
           const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();

     

       302
       +
           asyncIterator.next();

     

       303
        
           expect(pulls).toBe(1);

     

       304
        
       

     

       305
        
           let resolved = false;

     
···

       332
        
           };

     

       333
        
       

     

       334
        
           const asyncIterator = sinks.toAsyncIterable(source)[Symbol.asyncIterator]();

     

       335
       +
           const next$ = asyncIterator.next();

     

       336
        
       

     

       337
        
           sink!(push(0));

     

       338
       +
           expect(await next$).toEqual({ value: 0, done: false });

     

       339
        
           expect(await asyncIterator.return!()).toEqual({ done: true });

     

       340
        
       

     

       341
        
           sink!(push(1));

+27

src/helpers.ts

···

       1
        
       import { TalkbackFn, TeardownFn, Start, Push, SignalKind } from './types';

     

       2
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       3
        
       /** Placeholder {@link TeardownFn | teardown functions} that's a no-op.

     

       4
        
        * @see {@link TeardownFn} for the definition and usage of teardowns.

     

       5
        
        * @internal

     
···

       39
        
           0: value,

     

       40
        
         } as Push<T>;

     

       41
        
       }

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0

···

       1
        
       import { TalkbackFn, TeardownFn, Start, Push, SignalKind } from './types';

     

       2
        
       

     

       3
       +
       declare global {

     

       4
       +
         interface SymbolConstructor {

     

       5
       +
           readonly observable: symbol;

     

       6
       +
         }

     

       7
       +
       }

     

       8
       +
       

     

       9
        
       /** Placeholder {@link TeardownFn | teardown functions} that's a no-op.

     

       10
        
        * @see {@link TeardownFn} for the definition and usage of teardowns.

     

       11
        
        * @internal

     
···

       45
        
           0: value,

     

       46
        
         } as Push<T>;

     

       47
        
       }

     

       48
       +
       

     

       49
       +
       /** Returns the well-known symbol specifying the default AsyncIterator.

     

       50
       +
        * @internal

     

       51
       +
        */

     

       52
       +
       export const asyncIteratorSymbol = (): typeof Symbol.asyncIterator =>

     

       53
       +
         (typeof Symbol === 'function' && Symbol.asyncIterator) || ('@@asyncIterator' as any);

     

       54
       +
       

     

       55
       +
       /** Returns the well-known symbol specifying the default ES Observable.

     

       56
       +
        * @privateRemarks

     

       57
       +
        * This symbol is used to mark an object as a default ES Observable. By the specification, an object

     

       58
       +
        * that abides by the default Observable implementation must carry a method set to this well-known

     

       59
       +
        * symbol that returns the Observable implementation. It's common for this object to be an

     

       60
       +
        * Observable itself and return itself on this method.

     

       61
       +
        *

     

       62
       +
        * @see {@link https://github.com/0no-co/wonka/issues/122} for notes on the intercompatibility

     

       63
       +
        * between Observable implementations.

     

       64
       +
        *

     

       65
       +
        * @internal

     

       66
       +
        */

     

       67
       +
       export const observableSymbol = (): typeof Symbol.observable =>

     

       68
       +
         (typeof Symbol === 'function' && Symbol.observable) || ('@@observable' as any);

+1 -22

src/observable.ts

···

       1
        
       import { Source, SignalKind, TalkbackKind } from './types';

     

       2
       -
       import { push, start, talkbackPlaceholder } from './helpers';

     

       3
       -
       

     

       4
       -
       declare global {

     

       5
       -
         interface SymbolConstructor {

     

       6
       -
           readonly observable: symbol;

     

       7
       -
         }

     

       8
       -
       }

     

       9
        
       

     

       10
        
       /** A definition of the ES Observable Subscription type that is returned by

     

       11
        
        * {@link Observable.subscribe}

     
···

       117
        
         /** The well-known symbol specifying the default ES Observable for an object. */

     

       118
        
         [Symbol.observable](): Observable<T>;

     

       119
        
       }

     

       120
       -
       

     

       121
       -
       /** Returns the well-known symbol specifying the default ES Observable.

     

       122
       -
        * @privateRemarks

     

       123
       -
        * This symbol is used to mark an object as a default ES Observable. By the specification, an object

     

       124
       -
        * that abides by the default Observable implementation must carry a method set to this well-known

     

       125
       -
        * symbol that returns the Observable implementation. It's common for this object to be an

     

       126
       -
        * Observable itself and return itself on this method.

     

       127
       -
        *

     

       128
       -
        * @see {@link https://github.com/0no-co/wonka/issues/122} for notes on the intercompatibility

     

       129
       -
        * between Observable implementations.

     

       130
       -
        *

     

       131
       -
        * @internal

     

       132
       -
        */

     

       133
       -
       const observableSymbol = (): typeof Symbol.observable =>

     

       134
       -
         Symbol.observable || ('@@observable' as any);

     

       135
        
       

     

       136
        
       /** Converts an ES Observable to a {@link Source}.

     

       137
        
        * @param input - The {@link ObservableLike} object that will be converted.

···

       1
        
       import { Source, SignalKind, TalkbackKind } from './types';

     

       2
       +
       import { push, start, talkbackPlaceholder, observableSymbol } from './helpers';

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       3
        
       

     

       4
        
       /** A definition of the ES Observable Subscription type that is returned by

     

       5
        
        * {@link Observable.subscribe}

     
···

       111
        
         /** The well-known symbol specifying the default ES Observable for an object. */

     

       112
        
         [Symbol.observable](): Observable<T>;

     

       113
        
       }

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       114
        
       

     

       115
        
       /** Converts an ES Observable to a {@link Source}.

     

       116
        
        * @param input - The {@link ObservableLike} object that will be converted.

+51 -40

src/sinks.ts

···

       1
       -
       import { Source, Subscription, TalkbackKind, SignalKind } from './types';

     

       2
       -
       import { talkbackPlaceholder } from './helpers';

     

       3
        
       

     

       4
        
       /** Creates a subscription to a given source and invokes a `subscriber` callback for each value.

     

       5
        
        * @param subscriber - A callback function called for each issued value.

     
···

       124
        
        * }

     

       125
        
        * ```

     

       126
        
        */

     

       127
       -
       export const toAsyncIterable = <T>(source: Source<T>): AsyncIterable<T> => ({

     

       128
       -
         [Symbol.asyncIterator](): AsyncIterator<T> {

     

       129
       -
           const buffer: T[] = [];

     

       130
        
       

     

       131
       -
           let ended = false;

     

       132
       -
           let talkback = talkbackPlaceholder;

     

       133
       -
           let next: ((value: IteratorResult<T>) => void) | void;

     

       0
        
       
     

       0
        
       
     

       134
        
       

     

       135
       -
           source(signal => {

     

       136
       -
             if (ended) {

     

       137
       -
               /*noop*/

     

       138
       -
             } else if (signal === SignalKind.End) {

     

       139
       -
               if (next) next = next(doneResult);

     

       140
       -
               ended = true;

     

       141
       -
             } else if (signal.tag === SignalKind.Start) {

     

       142
       -
               (talkback = signal[0])(TalkbackKind.Pull);

     

       143
       -
             } else if (next) {

     

       144
       -
               next = next({ value: signal[0], done: false });

     

       145
       -
             } else {

     

       146
       -
               buffer.push(signal[0]);

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       147
        
             }

     

       148
       -
           });

     

       149
        
       

     

       150
       -
           return {

     

       151
       -
             async next(): Promise<IteratorResult<T>> {

     

       152
       -
               if (ended && !buffer.length) {

     

       153
       -
                 return doneResult;

     

       154
       -
               } else if (!ended && buffer.length <= 1) {

     

       155
       -
                 talkback(TalkbackKind.Pull);

     

       156
       -
               }

     

       157
        
       

     

       158
       -
               return buffer.length

     

       159
       -
                 ? { value: buffer.shift()!, done: false }

     

       160
       -
                 : new Promise(resolve => (next = resolve));

     

       161
       -
             },

     

       162
       -
             async return(): Promise<IteratorReturnResult<void>> {

     

       163
       -
               if (!ended) next = talkback(TalkbackKind.Close);

     

       164
       -
               ended = true;

     

       165
       -
               return doneResult;

     

       166
       -
             },

     

       167
       -
           };

     

       168
       -
         },

     

       169
       -
       });

     

       0
        
       
     

       0
        
       
     

       170
        
       

     

       171
        
       /** Subscribes to a given source and collects all synchronous values into an array.

     

       172
        
        * @param source - A {@link Source}.

···

       1
       +
       import { Source, Subscription, TalkbackKind, SignalKind, SourceIterable } from './types';

     

       2
       +
       import { talkbackPlaceholder, asyncIteratorSymbol } from './helpers';

     

       3
        
       

     

       4
        
       /** Creates a subscription to a given source and invokes a `subscriber` callback for each value.

     

       5
        
        * @param subscriber - A callback function called for each issued value.

     
···

       124
        
        * }

     

       125
        
        * ```

     

       126
        
        */

     

       127
       +
       export const toAsyncIterable = <T>(source: Source<T>): SourceIterable<T> => {

     

       128
       +
         const buffer: T[] = [];

     

       0
        
       
     

       129
        
       

     

       130
       +
         let ended = false;

     

       131
       +
         let started = false;

     

       132
       +
         let pulled = false;

     

       133
       +
         let talkback = talkbackPlaceholder;

     

       134
       +
         let next: ((value: IteratorResult<T>) => void) | void;

     

       135
        
       

     

       136
       +
         return {

     

       137
       +
           async next(): Promise<IteratorResult<T>> {

     

       138
       +
             if (!started) {

     

       139
       +
               started = true;

     

       140
       +
               source(signal => {

     

       141
       +
                 if (ended) {

     

       142
       +
                   /*noop*/

     

       143
       +
                 } else if (signal === SignalKind.End) {

     

       144
       +
                   if (next) next = next(doneResult);

     

       145
       +
                   ended = true;

     

       146
       +
                 } else if (signal.tag === SignalKind.Start) {

     

       147
       +
                   pulled = true;

     

       148
       +
                   (talkback = signal[0])(TalkbackKind.Pull);

     

       149
       +
                 } else {

     

       150
       +
                   pulled = false;

     

       151
       +
                   if (next) {

     

       152
       +
                     next = next({ value: signal[0], done: false });

     

       153
       +
                   } else {

     

       154
       +
                     buffer.push(signal[0]);

     

       155
       +
                   }

     

       156
       +
                 }

     

       157
       +
               });

     

       158
        
             }

     

       0
        
       
     

       159
        
       

     

       160
       +
             if (ended && !buffer.length) {

     

       161
       +
               return doneResult;

     

       162
       +
             } else if (!ended && !pulled && buffer.length <= 1) {

     

       163
       +
               pulled = true;

     

       164
       +
               talkback(TalkbackKind.Pull);

     

       165
       +
             }

     

       0
        
       
     

       166
        
       

     

       167
       +
             return buffer.length

     

       168
       +
               ? { value: buffer.shift()!, done: false }

     

       169
       +
               : new Promise(resolve => (next = resolve));

     

       170
       +
           },

     

       171
       +
           async return(): Promise<IteratorReturnResult<void>> {

     

       172
       +
             if (!ended) next = talkback(TalkbackKind.Close);

     

       173
       +
             ended = true;

     

       174
       +
             return doneResult;

     

       175
       +
           },

     

       176
       +
           [asyncIteratorSymbol()](): SourceIterable<T> {

     

       177
       +
             return this;

     

       178
       +
           },

     

       179
       +
         };

     

       180
       +
       };

     

       181
        
       

     

       182
        
       /** Subscribes to a given source and collects all synchronous values into an array.

     

       183
        
        * @param source - A {@link Source}.

+11 -3

src/sources.ts

···

       1
        
       import { Source, Sink, SignalKind, TalkbackKind, Observer, Subject, TeardownFn } from './types';

     

       2
       -
       import { push, start, talkbackPlaceholder, teardownPlaceholder } from './helpers';

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       3
        
       import { share } from './operators';

     

       4
        
       

     

       5
        
       /** Helper creating a Source from a factory function when it's subscribed to.

     
···

       45
        
        * @see {@link https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#the_async_iterator_and_async_iterable_protocols}

     

       46
        
        * for the JS Iterable protocol.

     

       47
        
        */

     

       48
       -
       export function fromAsyncIterable<T>(iterable: AsyncIterable<T>): Source<T> {

     

       49
        
         return sink => {

     

       50
       -
           const iterator = iterable[Symbol.asyncIterator]();

     

       0
        
       
     

       0
        
       
     

       51
        
           let ended = false;

     

       52
        
           let looping = false;

     

       53
        
           let pulled = false;

···

       1
        
       import { Source, Sink, SignalKind, TalkbackKind, Observer, Subject, TeardownFn } from './types';

     

       2
       +
       import {

     

       3
       +
         push,

     

       4
       +
         start,

     

       5
       +
         talkbackPlaceholder,

     

       6
       +
         teardownPlaceholder,

     

       7
       +
         asyncIteratorSymbol,

     

       8
       +
       } from './helpers';

     

       9
        
       import { share } from './operators';

     

       10
        
       

     

       11
        
       /** Helper creating a Source from a factory function when it's subscribed to.

     
···

       51
        
        * @see {@link https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#the_async_iterator_and_async_iterable_protocols}

     

       52
        
        * for the JS Iterable protocol.

     

       53
        
        */

     

       54
       +
       export function fromAsyncIterable<T>(iterable: AsyncIterable<T> | AsyncIterator<T>): Source<T> {

     

       55
        
         return sink => {

     

       56
       +
           const iterator: AsyncIterator<T> =

     

       57
       +
             (iterable[asyncIteratorSymbol()] && iterable[asyncIteratorSymbol()]()) || iterable;

     

       58
       +
       

     

       59
        
           let ended = false;

     

       60
        
           let looping = false;

     

       61
        
           let pulled = false;

src/types.d.ts

···

       200
        
         /** The {@link Source} that issues the signals as the {@link Observer} methods are called. */

     

       201
        
         source: Source<T>;

     

       202
        
       }

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0

···

       200
        
         /** The {@link Source} that issues the signals as the {@link Observer} methods are called. */

     

       201
        
         source: Source<T>;

     

       202
        
       }

     

       203
       +
       

     

       204
       +
       /** Async Iterable/Iterator after having converted a {@link Source}.

     

       205
       +
        * @see {@link toAsyncIterable} for a helper that creates this structure.

     

       206
       +
        */

     

       207
       +
       export interface SourceIterable<T> extends AsyncIterator<T>, AsyncIterable<T> {}