@fervqz/just-a-stream

A light weight js library for creating, transforming and merging streams.

LIVE DEMO

Installation

Via CDN:

<!-- Source script -->
<script src="https://cdn.jsdelivr.net/npm/@fervqz/[email protected]/dist/index.js"></script>

<!-- TypeScript Types -->
<script src="https://cdn.jsdelivr.net/npm/@fervqz/[email protected]/dist/index.d.ts"></script>

Via package manager:

npm install @fervqz/just-a-stream
# or
yarn install @fervqz/just-a-stream
# or
pnpm install @fervqz/just-a-stream
# or
bun install @fervqz/just-a-stream

Import:

import { JAStream } from '@fervqz/just-a-stream';
// or
const { JAStream } = require('@fervqz/just-a-stream');

Examples:

Move an element with arrow keys:

import { JAStream } from '@fervqz/just-a-stream';

const movementKeys = [
  'ArrowUp',
  'ArrowDown',
  'ArrowLeft',
  'ArrowRight',
];

const keyStrokes = new JAStream((next) => {
  document.addEventListener('keydown', next);
});

const arrowStrokes = keyStrokes
  .filter((eventKey) => movementKeys.includes(eventKey.code))
  .map((eventKey) => eventKey.code);

arrowStrokes.subscribe((arrowCode) => {
  console.log(arrowCode); // ArrowUp | ArrowDown | ArrowLeft | ArrowRight
  // Movement logic...
});

Creating a Stream

You can create a stream by instantiating a new JAStream object or by using the JAStream.from() method.

Using new JAStream()

Creates a stream of values of undefined length that will be received over time.

• params:
  • generator: DataGenerator<T>: A DataGenerator function. This is the value emitter.
  • options?: JAStreamOptions: JAStreamOptions options object for defining stream properties.
• return:
  • JAStream<U>: New JAStream.

constructor(generator: DataGenerator<T>, options?: JAStreamOptions)

import { JAStream } from '@fervqz/just-a-stream';

// Bark every second
const bark = () => console.log('woof!');

const stream = new JAStream((next) => {
    setInterval(next, 1000);
});

stream.subscribe(bark);
// Output:
// woof!      <- after 1s
// woof!      <- after 2s
// woof!      <- after 3s
// ...

Using JAStream.from()

Creates a stream of defined values to be able to transform the data with the JAStream methods.

import { JAStream } from '@fervqz/just-a-stream';

// Times barked
const barked = (event) => console.log(`Barked ${event} times!`);

const stream = JAStream.from([1, 2, 3, 4, 5]);

stream.subscribe(barked);
// Output (all console.log happens at the same time):
// Barked 1 times!
// Barked 2 times!
// Barked 3 times!
// Barked 4 times!
// Barked 5 times!

Strating a Stream with an initial value

To set the initial value of a stream, you can either:

• Set the initial value of accumulation with the reduce function, this will not trigger the listener function for initial value.
• Merge two streams, an initial stream with the offset value and the future values stream, this will trigger the listener function for the initial value.

Initial value with reduce():

import { JAStream } from '@fervqz/just-a-stream';

const INITIAL_VALUE = 'P';

// Will send 'A' every second
const stream = new JAStream((next) => {
    setInterval(() => next('A'), 1000);
});

// Suming the new value with the accumulated.
const accSecondsStream = stream.reduce(
    (acc, curr) => acc + curr,
    INITIAL_VALUE // <= starts at P
);

accSecondsStream.subscribe(event => console.log(`${event}`));

// Output:
// PA     <- First value ignored and processed with second value (first of stream)   
// PAA
// PAAA
// PAAAA
// PAAAAA
// ...

Offset initial value with initial stream merge:

import { JAStream } from '@fervqz/just-a-stream';

// Initial offset, starting from P
const initialValue = JAStream.from(['P']);

const futureValues = new JAStream((next) => {
    setInterval(() => next('A'), 1000);
});

const stream = JAStream.merge(initialValue, futureValues);

const accumulatedSecondsStream = stream.reduce((acc, curr) => {
    return acc + curr;
}, '');

accumulatedSecondsStream.subscribe(event => console.log(`${event}`));

// Output:
// P     <- First value processed by subscribe listener.
// PA
// PAA
// PAAA
// PAAAA
// PAAAAA
// ...

Subscribing to a Stream

Once you have a stream, you can subscribe to it to listen for emitted values.

import { JAStream } from '@fervqz/just-a-stream';

const stream = JAStream.from([1, 2, 3, 4, 5]);

stream.subscribe((event) => {
    console.log(`Dog says: ${"woof ".repeat(event)}`);
});

// Output:
// Dog says: woof
// Dog says: woof woof
// Dog says: woof woof woof
// Dog says: woof woof woof woof
// Dog says: woof woof woof woof woof

import { JAStream } from '@fervqz/just-a-stream';

const randomNumberStream = new JAStream((next) => {
    setInterval(() => {
        // Random number between 0-9
        const randomNumber = Math.floor(Math.random() * 10);
        next(randomNumber);
    }, 1000);
});

randomNumberStream.subscribe((event) => {
    console.log(`Dog says: ${"woof ".repeat(event)}`);
});

// Output:
// Dog says: woof woof woof woof woof
// Dog says: woof
// Dog says: woof woof woof
// Dog says: woof woof woof woof
// Dog says: woof woof woof
// Dog says: woof woof
// Dog says: woof woof woof woof woof woof woof

Transforming Streams

You can transform streams using methods like filter, map, and reduce.

Filtering Stream

Filters values in the stream based on a predicate function.

filter(pred: Predicate<T>): JAStream<T>

• params:
  • pred: Predicate<T>: Predicate function to filter values.
• return:
  • JAStream<T>: New JAStream with filtered values.

import { JAStream } from '@fervqz/just-a-stream';

const stream = JAStream.from([1, 2, 3, 4, 5]);

const isOdd = (value) => (value % 2) > 0;
const isEven = (value) => (value % 2) === 0;
const oddNumbersStream = stream.filter(isOdd);
const evenNumbersStream = stream.filter(isEven);

oddNumbersStream.subscribe((e) => console.log(`odd: ${e}`));
evenNumbersStream.subscribe((e) => console.log(`even: ${e}`));

// Output:
// odd: 1
// odd: 3
// odd: 5
// even: 2
// even: 4

import { JAStream } from '@fervqz/just-a-stream';

const stream = new JAStream((next) => {

    let counter = 0;

    setInterval(() => {
        next(counter);
        counter++;
    }, 1000);

});

const isOdd = (value) => (value % 2) > 0;
const isEven = (value) => (value % 2) === 0;
const oddNumbersStream = stream.filter(isOdd);
const evenNumbersStream = stream.filter(isEven);

oddNumbersStream.subscribe((e) => console.log(`odd: ${e}`));
evenNumbersStream.subscribe((e) => console.log(`even: ${e}`));

// Output:
// odd: 1
// even: 2
// odd: 3
// even: 4
// odd: 5

Mapping Stream

Maps values in the stream to a different type using a mapper function.

map<U>(mapFn: Mapper<T, U>): JAStream<U> 

• params:
  • mapFn: Mapper<T, U>: Mapper function to transform values.
• return:
  • JAStream<U>: New JAStream with mapped values.

import { JAStream } from '@fervqz/just-a-stream';

const disrtyDogsStream = JAStream.from([
    'dirty dog 1',
    'dirty dog 2',
    'dirty dog 3',
    'dirty dog 4',
    'dirty dog 5',
]);

const batheDog = (dog) => dog.replace('dirty', 'clean');
const cleanDogsStream = disrtyDogsStream.map(batheDog);

cleanDogsStream.subscribe((dog) => console.log(dog));

// Output:
// cleaned dog 1
// cleaned dog 2
// cleaned dog 3
// cleaned dog 4
// cleaned dog 5

import { JAStream } from '@fervqz/just-a-stream';

// Random dirty and cleaned dogs
const dirtyCleanDogsStream = new JAStream((next) => {

    setInterval(() => {
        next(`${Math.random() > 0.5 ? 'dirty' : 'cleaned'} dog`);
    }, 1000);

});

const batheDog = (dog) => dog.includes('dirty') ? dog.replace('dirty', 'cleaned') : dog;
const cleanDogsStream = dirtyCleanDogsStream.map(batheDog);

cleanDogsStream.subscribe((dog) => console.log(dog));

// Output:
// cleaned dog 1      <- after 1s
// cleaned dog 2      <- after 2s
// cleaned dog 3      <- after 3s
// cleaned dog 4      <- after 4s
// cleaned dog 5      <- after 5s
// ...

Reducing Stream

Reduces values in the stream to a single value using a reducer function.

reduce<U>(fn: Reducer<T, U>, initialValue: U): JAStream<U>

• params:
  • fn: Reducer<T, U>: Reducer function to accumulate values.
  • initialValue: {U}: Initial value for the accumulator.
• return:
  • Stream<U>: New JAStream with the reduced value.

import { JAStream } from '@fervqz/just-a-stream';

const initialValue = JAStream.from(['frist_woof']);

const stream = new JAStream((next) => {
    setInterval(() => next('woof!'), 1000);
});

const woofsStream = JAStream
    .merge(initialValue, stream)
    .reduce((acc, curr) => {
        return `${acc} ${curr}`;
    }, 'dogs say:');

woofsStream.subscribe(event => console.log(event));

// Output:
// dogs say: first_woof
// dogs say: first_woof woof
// dogs say: first_woof woof woof
// dogs say: first_woof woof woof woof
// dogs say: first_woof woof woof woof woof
// ...

Combining Streams

Merges multiple streams into a single stream.

JAStream.merge = <T>(...streams: JAStream<T>[]): JAStream<T>

• params:
  • streams: {...JAStream<T>[]}: Streams to merge.
• return:
  • JAStream<T>: New JSAtream with values from all input streams.

import { JAStream } from '@fervqz/just-a-stream';

const keysDownStream = new JAStream((next) => {
    document.addEventListener("keydown", next)
});

const clicksStream = new JAStream((next) => {
    document.addEventListener("click", next)
});

const keysClickStream = JAStream.merge(keysDownStream, clicksStream);

keysClickStream.subscribe(event => {
    console.log("You are either clicking or typing");
});
// Output:
// You are either clicking or typing
// You are either clicking or typing
// You are either clicking or typing
// You are either clicking or typing
// You are either clicking or typing

Getting the Last Value

Gets the last emitted value in the stream. IMPORTANT: You must subscribe to the stream before calling .getLast().

getLast(): T | undefined

• return
  • T | undefined: The last emitted value, or undefined if no value has been emitted yet.

import { JAStream } from '@fervqz/just-a-stream';

const stream = JAStream.from([
    'first dog',
    'second dog',
    'third dog',
    'fourth dog',
    'fifth dog',
]);

// You must subscribe to the stream first
stream.subscribe(() => {
    // ...
});
const lastDog = stream.getLast();


console.log(lastDog);
// Output:
// fifth dog

Getting Buffered Values

Gets the buffer values for the last N elements or all values emitted. Usages of buffer must be specified in the options object when creating the stream:

Setting usage of buffer:

import { JAStream } from '@fervqz/just-a-stream';

const generator = (next) => {...}

const options = {
  useBuffer: true,
  bufferSize: 10, // default value = 1
}

const stream = new JAStream(generator, options)

Using buffer:

getBuffer(): T[]

• return
  • T[]: The last emitted values, or [] if no value has been emitted yet.

import { JAStream } from '@fervqz/just-a-stream';

const generator = next => {
    setInterval(() => next((Math.random() * 10).toFixed()), 1000);
};

const options = {
    useBuffer: true,
    bufferSize: 3,
}
const stream = new JAStream(generator, options);

stream.subscribe(event => {
    console.log(event, ' - ', stream.getBuffer());
});

// Output:
// 2 - [2]
// 4 - [2, 4]
// 1 - [2, 4, 1]
// 5 - [4, 1, 5]
// 5 - [1, 5, 5]

TypeScript Types

JAStreamOptions

• useBuffer: Default value false. You can use getBuffer() when true.
• bufferSize: Maximum number of elements to store. If not setted the default value is 1.

/**
 * Represents the options that can be passed to the JAStream class.
 */
export interface JAStreamOptions {
    useBuffer: boolean;
    bufferSize: number;
}

Other Types

/**
 * Represents a function that emits values overt time.
 */
type DataGenerator<T> = (listener: Listener<T>) => void;

/**
 * Represents a function that listens for emitted values.
 */
type Listener<T> = (value: T) => void;

/**
 * Represents a predicate function to filter values.
 */
type Predicate<T> = (value: T) => boolean;

/**
 * Represents a mapper function to transform values.
 */
type Mapper<T, U> = (value: T) => U;

/**
 * Represents a reducer function to reduce values.
 */
type Reducer<T, U> = (accumulator: U, currentValue: T) => U;