92green-rxjs 中文文档教程

Question

92green-rxjs

yarn add 92green-rxjs

Rxjs 算法。

• multiCache
• memoryCache
• zipDiff
• operators/bufferDistinct
• operators/complete
• dynamodb/batchGetWithRetry
• dynamodb/batchWriteWithRetry
• dynamodb/queryAll
• dynamodb/scanAll

multiCache

接受一组源（旨在成为内存缓存、数据库缓存和实际数据源之类的东西），并针对每个项目尝试按顺序从每个源检索项目，直到找到为止。 一旦找到，它还会将项目保存在缓存中。 它不保证商品顺序。

import {multiCache} from '92green-rxjs';

let myMemoryCacheOperator = {
    name: 'memory-cache',
    load, // operator that loads from cache
    save, // operator that saves to cache (not needed on data source)
    clear, // operator that clears item from cache (not needed on data source)
};

// ^ all of the above operators must accept a payload of {id, item} and return {id, item}
//   item may be undefined if not known or not found

let cache = multiCache([
    myMemoryCacheOperator, // first cache
    myDatabaseCacheOperator, // deeper cache
    myDataSource // data source
])

from([
    {id: 'a'},
    {id: 'b'},
    {id: 'c'}
])
    .pipe(cache.load);

// Example output observable:

// {id: 'a', item: {id: 'a', name: 'Hi'}, source: 'data-source'}
// {id: 'b', item: {id: 'b', name: 'Hello'}, source: 'memory-cache'}
// {id: 'c', item: undefined, source: undefined}

memoryCache

一个 rx 缓存，其工作方式很像 graphql/dataloader。

与 multiCache 一起使用：

import {from} from 'rxjs'
import {flatMap} from 'rxjs/operators'
import {multiCache} from '92green-rxjs';
import {memoryCache} from '92green-rxjs';

let dataSource = {
    name: 'data-source',
    load: flatMap(async (payload) => ({
        ...payload,
        item: await fetch(payload.id)
    }))
};

let cache = multiCache([
    memoryCache(),
    dataSource
]);

from([
    {id: 'a'},
    {id: 'b'},
    {id: 'a'}
])
    .pipe(cache.load);


// Example output observable:

// {id: 'a', item: {id: 'a', name: 'Hi'}, source: 'data-source'}
// {id: 'b', item: {id: 'b', name: 'Hello'}, source: 'data-source'}
// {id: 'a', item: {id: 'a', name: 'Hi'}, source: 'memory-cache'}
// a is only requested once

zipDiff

获取两个可观察对象，并根据键查找其中一个或两个对象中存在哪些项目。 它发出的项目与在一个对象上压缩在一起的匹配项目：{a: itemA, b: itemB}, 并在具有单个键的对象上发出不匹配的项，如 {a: itemA}、{b: itemB}。

keyBy 函数在每个项目上被调用。 返回的数据用作标识符 当两个项目匹配时计算出来。

如果您需要为第二个可观察对象使用不同的 keyBy 函数，则可以添加第二个 keyBy 函数。

项目会尽早发出，因此内部缓冲区可以尽可能小。 成对的匹配项在匹配时立即发出，不匹配的项在知道不可能匹配时立即发出。 （例如，如果缓冲区包含来自 A 的项目，并且输入 observable B 完成，则已知永远不会匹配任何缓冲的 A，因此它们都被发射）

import {zipDiff} from '92green-rxjs';

zipDiff(
    obsA: Observable,
    obsB: Observable,
    keyBy: (item) => any,
    keyByB?: (item) => any
): Observable

zipDiff(obsA, obsB, itemA => itemA.id)

// obsA adds {id: "1", name: "One"}
// obsA adds {id: "2", name: "Two"}
// obsB adds {id: "3", name: "Three"}
// obsB adds {id: "2", name: "Too"}

// output:

{
    a: {id: "2", name: "Two"},
    b: {id: "2", name: "Too"}
}

{
    a: {id: "1", name: "One"}
}

{
    b: {id: "3", name: "Three"}
}

operators/bufferDistinct

缓冲项目，同时给定的谓词函数返回相同的东西，并发射当给定的谓词函数返回其他内容或流关闭时缓冲区内容。 使用 Object.is() 执行值比较。

如果传递了 flushObs，则每次 flushObs 发出时都会刷新缓冲区。

import {bufferDistinct} from '92green-rxjs/operators';

bufferDistinct(item => comparisonValue, flushObs: ?Observable)

obs.pipe(
    bufferDistinct(item => item.id)
);

// Obs adds {id: 'a', value: 1}
// Obs adds {id: 'a', value: 2}
// Obs adds {id: 'b', value: 3}
// Obs adds {id: 'c', value: 4}
// Obs adds {id: 'c', value: 5}

// output:

[
    {id: 'a', value: 1},
    {id: 'a', value: 2}
]

[
    {id: 'b', value: 3}
]

[
    {id: 'c', value: 4},
    {id: 'c', value: 5}
]

operators/complete

在可观察对象完成时发出单个项目。

import {complete} from '92green-rxjs/operators';

complete()

obs.pipe(
    complete(),
    tap(() => {
        console.log("I'm done mate");
    })
);

dynamodb/batchGetWithRetry

将 AWS DocClient.batchGet() 变成一个管道可观察对象，它接受 id 的可观察对象并调用 batchGet()，一次将项目批处理到 100 个并重试 dynamo 的 UnprocessedKeys 自动。

import {batchGetWithRetry} from '92green-rxjs/dynamodb';

batchGetWithRetry({
    docClient: DocClient,
    tableName: string,
    returnItems?: boolean
}): Observable => Observable

let keys = [
    {id: 1},
    {id: 2},
    {id: 3}
];

from(keys)
    .pipe(batchGetWithRetry(...))
    .toPromise();

dynamodb/batchWriteWithRetry

将 AWS DocClient.batchWrite() 变成一个管道可观察对象，它接受参数的可观察对象并调用 batchWrite()，一次将项目批处理到 25 个并重试 dynamo 的 UnprocessedItems 自动。

import {batchWriteWithRetry} from '92green-rxjs/dynamodb';

batchWriteWithRetry({
    docClient: DocClient,
    tableName: string,
    returnItems?: boolean
}): Observable => Observable

from([{
    {
        PutRequest: {
            Item: {
                foo: 200
            }
        }
    }
    ...
}])
    .pipe(batchWriteWithRetry(...))
    .toPromise();

dynamodb/queryAll

将 AWS DocClient.query() 变成一个可观察对象，默认情况下，只要有更多数据需要分页，它就会继续请求。

import {queryAll} from '92green-rxjs/dynamodb';

queryAll(
    docClient: DocClient,
    params: Params,
    feedbackObservable: ?Observable
): Observable

dynamodb/scanAll

将 AWS DocClient.scan() 变成一个可观察对象，默认情况下，只要有更多数据需要分页，它就会继续请求。

import {scanAll} from '92green-rxjs/dynamodb';

scanAll(
    docClient: DocClient,
    params: Params,
    feedbackObservable: ?Observable
): Observable