v0

Useful Rxjs Operators & Utils.

你的关注是我最大的动力。 Your Star is the best gift.

• Install
• Usage
  • Operators
    • delayRetry
    • tapAsync
  • Decorators
    • @Cacheable(timeout = 0, mode = ReturnType.Detect)
  • Utils
    • cacheable(FN, timeout = 0, isPromise = false)
    • RxPromise
• Contribute
• License

Install

npm i --save rxjs v0
# or
yarn add rxjs v0

Usage

中文文档参考： rx.js.cool 中的【进阶（Advanced）】章节系列文章

Operators

delayRetry

{
  maxAttempts?: number;
  duration?: number;
}

Defaults:

{
  maxAttempts = 3,
  duration = 1000
}

Usage:

import { delayRetry } from 'v0';
// import { delayRetry } from 'v0/operators';
source$.pipe(
  // ...
  // Retry all options in current pipe
  delayRetry({
    maxAttempts: 2,
    duration: 200
  }),
  // catchError is needed
  catchError((error) => of(error))
);

tapAsync

Just like tap, support async/await (promise) function.

Usage:

import { tapAsync } from 'v0';
// import { tapAsync } from 'v0/operators';
source$.pipe(
  tapAsync(async (val) => {
    await SomeFn(val);
  })
);

Decorators

@Cacheable(timeout = 0, mode = ReturnType.Detect)

Return Type:

enum ResultType {
  Promise = 'Promise',
  Observable = 'Observable',
  Detect = 'Detect'
}

Usage:

import { interval } from 'rxjs';
import { Cacheable, ReturnType } from 'v0';
// import { Cacheable } from 'v0/decorators';
class Demo {
  @Cacheable(300, ReturnType.Promise)
  async save(n: number) {
    // await something...
    console.log(`${n} saved`);
    return `${n} succeed`;
  }
}
const demo = new Demo();
demo.save(1).then(console.log);
demo.save(1).then(console.log);
demo.save(3).then(console.log);
/** logs:
 *  1 saved         <---- save only called once, the second call resued before if last call is pending
 *  1 succeed
 *  1 succeed
 *  3 saved
 *  3 succeed
 */

Utils

cacheable(FN, timeout = 0, isPromise = false)

import { cacheable } from 'v0';
// import { cacheable } from 'v0/utils';
const get = () => {
  return from(
    new Promise((resolve) => {
      setTimeout(() => resolve(new Date()), 2000);
    })
  );
};
const cachedGet = cacheable(get, false /*if this function returns a PromiseLike result*/);

RxPromise

import { RxPromise } from 'v0';
// import { RxPromise } from 'v0/utils';
// RxPromise<T, R = Error> extends Observable<T>
const mockedPromise = new RxPromise(resolver);

Type of resolver:

(resolve: (r: T) => void, reject: (r: R) => void) => void

Example，transform mongoose exec to observable:

import mongoose from 'mongoose';
import { RxPromise } from 'v0';
mongoose.Promise = RxPromise;
mongoose.connect('mongodb://localhost/test', { useNewUrlParser: true, useUnifiedTopology: true });
const kittySchema = new mongoose.Schema({
  name: String
});
const Kitten = mongoose.model('Kitten', kittySchema);
const s$ = <Observable<Record<string, unknown>[]>>(<any>Kitten.find().exec());
// or
// const s$ = (Kitten.find().exec()) as any) as Observable<Record<string, unknown>[]>;
s$.subscribe({
  next(v) {
    console.log(v);
  },
  complete() {
    console.log('ended');
  }
});

Contribute

1. Add your own operator
2. Update Readme Doc (List / Usage)
3. Make a PR

License

Apache 2.0

通过支付宝捐赠：