Machinat Stream

Reactive programming stream for handling events in back-end.

⚠ This package is still on early experimental. There might be breaking changes in the future for supporting cluster. You can check the future road map here.

Install

npm install @machinat/stream
# or with yarn
yarn add @machinat/stream

Docs

Check the Reactive Programming document and the package reference.

Example

import { makeContainer, IntentRecognizer } from '@machinat/core';
import { fromApp } from '@machinat/stream';
import { map, filter } from '@machinat/stream/operators';
import app from './app';

const event$ = fromApp(app);

const textMsg$ = events$.pipe(
  filter(({ event }) => event.type === 'text'),
  map(
    makeContainer({ deps: [IntentRecognizer] })(
      (recognizer) =>
        async (context) => {
          const { channel, text } = context.event;
          const intent = await recognizer.detectText(channel, text);
          return { ...context, intent };
        }
    )
  )
);

textMsg$.subscribe(async ({ intent, reply }) => {
  const action = intent.type;
  if (action) {
    await reply(`start ${action}...`);
  }
});

Index

Modules

stream/operators

Classes

Stream

Type aliases

Variables

STREAMING_KEY_I

Functions

Type aliases

EventContextOfApp

EventContextOfApp<App>: App extends MachinatApp<any, infer Context> ? Context : never

Type parameters

App: MachinatApp<any, any>

OperatorFunction

OperatorFunction<T, R>: (input: Stream<T>) => Stream<R>

Type parameters

Type declaration

- (input: Stream<T>): Stream<R>
- Parameters
 - input: Stream<T>
 Returns Stream<R>

StreamingFrame

StreamingFrame<T>: { key: undefined | string; scope: ServiceScope; value: T }

Type parameters

Type declaration

key: undefined | string
scope: ServiceScope
value: T

Variables

STREAMING_KEY_I

STREAMING_KEY_I: SingularServiceInterface<unknown> = ...

Functions

conditions

conditions<T, Predicators>(source: Stream<T>, predicators: Predicators): { [ I in keyof Predicators]: Predicators[I] extends MaybeContainer<(v: unknown) => v is infer U> ? Stream : Stream<T> }
conditions<T>(source: Stream<T>, predicators: MaybeContainer<PredicateFn<T>>[]): Stream<T>[]

- Defined in packages/stream/src/conditions.ts:7
Type parameters
- T
- Predicators: readonly MaybeContainer<PredicateFn<T>>[]
Parameters
- source: Stream<T>
- predicators: Predicators
Returns { [ I in keyof Predicators]: Predicators[I] extends MaybeContainer<(v: unknown) => v is infer U> ? Stream : Stream<T> }
- Defined in packages/stream/src/conditions.ts:21
Type parameters
- T
Parameters
- source: Stream<T>
- predicators: MaybeContainer<PredicateFn<T>>[]
Returns Stream<T>[]

Const fromApp

fromApp<App>(app: App): Stream<EventContextOfApp<App>>

- Defined in packages/stream/src/fromApp.ts:6
Type parameters
- App: MachinatApp<any, AnyEventContext, App>
Parameters
- app: App
Returns Stream<EventContextOfApp<App>>

merge

merge<A, B>(a: Stream<A>, b: Stream): Stream<A | B>
merge<A, B, C>(a: Stream<A>, b: Stream, c: Stream<C>): Stream<A | B | C>
merge<A, B, C, D>(a: Stream<A>, b: Stream, c: Stream<C>, d: Stream<D>): Stream<A | B | C | D>
merge<A, B, C, D, E>(a: Stream<A>, b: Stream, c: Stream<C>, d: Stream<D>, e: Stream<E>): Stream<A | B | C | D | E>
merge<A, B, C, D, E, F>(a: Stream<A>, b: Stream, c: Stream<C>, d: Stream<D>, e: Stream<E>, f: Stream<F>): Stream<A | B | C | D | E | F>
merge<A, B, C, D, E, F, G>(a: Stream<A>, b: Stream, c: Stream<C>, d: Stream<D>, e: Stream<E>, f: Stream<F>, g: Stream<G>): Stream<A | B | C | D | E | F | G>
merge<A, B, C, D, E, F, G, H>(a: Stream<A>, b: Stream, c: Stream<C>, d: Stream<D>, e: Stream<E>, f: Stream<F>, g: Stream<G>, h: Stream<H>): Stream<A | B | C | D | E | F | G | H>
merge<A, B, C, D, E, F, G, H, I>(a: Stream<A>, b: Stream, c: Stream<C>, d: Stream<D>, e: Stream<E>, f: Stream<F>, g: Stream<G>, h: Stream<H>, i: Stream): Stream<A | B | C | D | E | F | G | H | I>

- Defined in packages/stream/src/merge.ts:5
Type parameters
- A
- B
Parameters
- a: Stream<A>
- b: Stream
Returns Stream<A | B>
- Defined in packages/stream/src/merge.ts:6
Type parameters
- A
- B
- C
Parameters
- a: Stream<A>
- b: Stream
- c: Stream<C>
Returns Stream<A | B | C>
- Defined in packages/stream/src/merge.ts:7
Type parameters
- A
- B
- C
- D
Parameters
- a: Stream<A>
- b: Stream
- c: Stream<C>
- d: Stream<D>
Returns Stream<A | B | C | D>
- Defined in packages/stream/src/merge.ts:8
Type parameters
- A
- B
- C
- D
- E
Parameters
- a: Stream<A>
- b: Stream
- c: Stream<C>
- d: Stream<D>
- e: Stream<E>
Returns Stream<A | B | C | D | E>
- Defined in packages/stream/src/merge.ts:9
Type parameters
- A
- B
- C
- D
- E
- F
Parameters
- a: Stream<A>
- b: Stream
- c: Stream<C>
- d: Stream<D>
- e: Stream<E>
- f: Stream<F>
Returns Stream<A | B | C | D | E | F>
- Defined in packages/stream/src/merge.ts:10
Type parameters
- A
- B
- C
- D
- E
- F
- G
Parameters
- a: Stream<A>
- b: Stream
- c: Stream<C>
- d: Stream<D>
- e: Stream<E>
- f: Stream<F>
- g: Stream<G>
Returns Stream<A | B | C | D | E | F | G>
- Defined in packages/stream/src/merge.ts:11
Type parameters
- A
- B
- C
- D
- E
- F
- G
- H
Parameters
- a: Stream<A>
- b: Stream
- c: Stream<C>
- d: Stream<D>
- e: Stream<E>
- f: Stream<F>
- g: Stream<G>
- h: Stream<H>
Returns Stream<A | B | C | D | E | F | G | H>
- Defined in packages/stream/src/merge.ts:12
Type parameters
- A
- B
- C
- D
- E
- F
- G
- H
- I
Parameters
- a: Stream<A>
- b: Stream
- c: Stream<C>
- d: Stream<D>
- e: Stream<E>
- f: Stream<F>
- g: Stream<G>
- h: Stream<H>
- i: Stream
Returns Stream<A | B | C | D | E | F | G | H | I>

pipe

pipe<A, B>(fn1: (a: A) => B): (a: A) => B
pipe<A, B, C>(fn1: (a: A) => B, fn2: (b: B) => C): (a: A) => C
pipe<A, B, C, D>(fn1: (a: A) => B, fn2: (b: B) => C, fn3: (c: C) => D): (a: A) => D
pipe<A, B, C, D, E>(fn1: (a: A) => B, fn2: (b: B) => C, fn3: (c: C) => D, fn4: (d: D) => E): (a: A) => E
pipe<A, B, C, D, E, F>(fn1: (a: A) => B, fn2: (b: B) => C, fn3: (c: C) => D, fn4: (d: D) => E, fn5: (e: E) => F): (a: A) => F
pipe<T>(...fns: ((x: T) => T)[]): (x: T) => T

- Defined in packages/stream/src/pipe.ts:2
Type parameters
- A
- B
Parameters
- fn1: (a: A) => B
  - - (a: A): B
    - Parameters
      - a: A
      Returns B
Returns (a: A) => B
- - (a: A): B
  - Parameters
    - a: A
    Returns B
- Defined in packages/stream/src/pipe.ts:3
Type parameters
- A
- B
- C
Parameters
- fn1: (a: A) => B
  - - (a: A): B
    - Parameters
      - a: A
      Returns B
- fn2: (b: B) => C
  - - (b: B): C
    - Parameters
      - b: B
      Returns C
Returns (a: A) => C
- - (a: A): C
  - Parameters
    - a: A
    Returns C
- Defined in packages/stream/src/pipe.ts:4
Type parameters
- A
- B
- C
- D
Parameters
- fn1: (a: A) => B
  - - (a: A): B
    - Parameters
      - a: A
      Returns B
- fn2: (b: B) => C
  - - (b: B): C
    - Parameters
      - b: B
      Returns C
- fn3: (c: C) => D
  - - (c: C): D
    - Parameters
      - c: C
      Returns D
Returns (a: A) => D
- - (a: A): D
  - Parameters
    - a: A
    Returns D
- Defined in packages/stream/src/pipe.ts:9
Type parameters
- A
- B
- C
- D
- E
Parameters
- fn1: (a: A) => B
  - - (a: A): B
    - Parameters
      - a: A
      Returns B
- fn2: (b: B) => C
  - - (b: B): C
    - Parameters
      - b: B
      Returns C
- fn3: (c: C) => D
  - - (c: C): D
    - Parameters
      - c: C
      Returns D
- fn4: (d: D) => E
  - - (d: D): E
    - Parameters
      - d: D
      Returns E
Returns (a: A) => E
- - (a: A): E
  - Parameters
    - a: A
    Returns E
- Defined in packages/stream/src/pipe.ts:15
Type parameters
- A
- B
- C
- D
- E
- F
Parameters
- fn1: (a: A) => B
  - - (a: A): B
    - Parameters
      - a: A
      Returns B
- fn2: (b: B) => C
  - - (b: B): C
    - Parameters
      - b: B
      Returns C
- fn3: (c: C) => D
  - - (c: C): D
    - Parameters
      - c: C
      Returns D
- fn4: (d: D) => E
  - - (d: D): E
    - Parameters
      - d: D
      Returns E
- fn5: (e: E) => F
  - - (e: E): F
    - Parameters
      - e: E
      Returns F
Returns (a: A) => F
- - (a: A): F
  - Parameters
    - a: A
    Returns F
- Defined in packages/stream/src/pipe.ts:22
Type parameters
- T
Parameters
- Rest ...fns: ((x: T) => T)[]
Returns (x: T) => T
- - (x: T): T
  - Parameters
    - x: T
    Returns T

Machinat Stream

Install

Docs

Example

Index

Modules

Classes

Type aliases

Variables

Functions

Type aliases

EventContextOfApp

Type parameters

App: MachinatApp<any, any>

OperatorFunction

Type parameters

T

R

Type declaration

Parameters

input: Stream<T>

Returns Stream<R>

StreamingFrame

Type parameters

T

Type declaration

key: undefined | string

scope: ServiceScope

value: T

Variables

STREAMING_KEY_I

Functions

conditions

Type parameters

T

Predicators: readonly MaybeContainer<PredicateFn<T>>[]

Parameters

source: Stream<T>

predicators: Predicators

Returns { [ I in keyof Predicators]: Predicators[I] extends MaybeContainer<(v: unknown) => v is infer U> ? Stream<U> : Stream<T> }

Type parameters

T

Parameters

source: Stream<T>

predicators: MaybeContainer<PredicateFn<T>>[]

Returns Stream<T>[]

Const fromApp

Type parameters

App: MachinatApp<any, AnyEventContext, App>

Parameters

app: App

Returns Stream<EventContextOfApp<App>>

merge

Type parameters

A

B

Parameters

a: Stream<A>

b: Stream<B>

Returns Stream<A | B>

Type parameters

A

B

C

Parameters

a: Stream<A>

b: Stream<B>

c: Stream<C>

Returns Stream<A | B | C>

Type parameters

A

B

C

D

Parameters

a: Stream<A>

b: Stream<B>

c: Stream<C>

d: Stream<D>

Returns Stream<A | B | C | D>