javascript - 在未完成时缓存 Rxjs http 请求的最短代码？

Question

我正在尝试创建一个满足以下要求的可观察流:

在订阅时从存储加载数据

如果数据尚未过期，则返回存储值的 observable

如果数据已过期，则返回一个 HTTP 请求 observable，该请求使用刷新 token 来获取新值并存储它

如果在请求完成之前再次到达此代码，则返回相同的请求可观察

如果在上一个请求完成后或使用不同的刷新 token 到达此代码，则开始新的请求

我知道关于如何执行步骤 (3) 有很多不同的答案，但是当我尝试一起执行这些步骤时，我正在寻找有关我提出的解决方案是否最简洁的指导可以(我怀疑!)。

这是一个演示我当前方法的示例:

var cachedRequestToken;
var cachedRequest;

function getOrUpdateValue() {
    return loadFromStorage()
      .flatMap(data => {
        // data doesn't exist, shortcut out
        if (!data || !data.refreshtoken) 
            return Rx.Observable.empty();

        // data still valid, return the existing value
        if (data.expires > new Date().getTime())
            return Rx.Observable.return(data.value);

        // if the refresh token is different or the previous request is 
        // complete, start a new request, otherwise return the cached request
        if (!cachedRequest || cachedRequestToken !== data.refreshtoken) { 
            cachedRequestToken = data.refreshtoken;

            var pretendHttpBody = {
                value: Math.random(),
                refreshToken: Math.random(),
                expires: new Date().getTime() + (10 * 60 * 1000) // set by server, expires in ten minutes
            };

            cachedRequest = Rx.Observable.create(ob => { 
                // this would really be a http request that exchanges
                // the one use refreshtoken for new data, then saves it
                // to storage for later use before passing on the value

                window.setTimeout(() => { // emulate slow response   
                    saveToStorage(pretendHttpBody);
                    ob.next(pretendHttpBody.value);                 
                    ob.completed(); 
                    cachedRequest = null; // clear the request now we're complete
                }, 2500);
          });
        }

        return cachedRequest;
    });
}

function loadFromStorage() {
    return Rx.Observable.create(ob => {
        var storedData = {  // loading from storage goes here
            value: 15,      // wrapped in observable to delay loading until subscribed
            refreshtoken: 63, // other process may have updated this between requests
            expires: new Date().getTime() - (60 * 1000) // pretend to have already expired
        };

        ob.next(storedData);
        ob.completed();
    })
}

function saveToStorage(data) {
    // save goes here
}

// first request
getOrUpdateValue().subscribe(function(v) { console.log('sub1: ' + v); }); 

// second request, can occur before or after first request finishes
window.setTimeout(
    () => getOrUpdateValue().subscribe(function(v) { console.log('sub2: ' + v); }),
    1500); 

Answer 1

首先，看看一个有效的 jsbin example .

解决方案与您的初始代码略有不同，我想解释一下原因。需要不断返回到您的本地存储，保存它，保存标志(缓存和 token )不适合我使用响应式(Reactive)、功能性方法。我给出的解决方案的核心是:

var data$ = new Rx.BehaviorSubject(storageMock);
var request$ = new Rx.Subject();
request$.flatMapFirst(loadFromServer).share().startWith(storageMock).subscribe(data$);
data$.subscribe(saveToStorage);

function getOrUpdateValue() {
    return data$.take(1)
      .filter(data => (data && data.refreshtoken))
      .switchMap(data => (data.expires > new Date().getTime() 
                    ? data$.take(1)
                    : (console.log('expired ...'), request$.onNext(true) ,data$.skip(1).take(1))));
}

关键是 data$保存您的最新数据并始终保持最新状态，只需执行 data$.take(1) 即可轻松访问。 . take(1)确保您的订阅获得单个值并终止很重要(因为您尝试以程序方式工作，而不是功能方式)。没有 take(1)您的订阅将保持事件状态，并且您将有多个处理程序，也就是说，您还将在仅用于当前更新的代码中处理 future 的更新。

另外，我持有 request$ subject 这是您开始从服务器获取新数据的方式。该函数的工作原理如下:

过滤器确保如果您的数据为空或没有 token ，则没有任何内容通过，类似于 return Rx.Observable.empty()你有过。

如果数据是最新的，则返回 data$.take(1)这是您可以订阅的单个元素序列。

如果没有，则需要刷新。为此，它会触发 request$.onNext(true)并返回 data$.skip(1).take(1) . skip(1)是为了避免当前的、过时的值。

为简洁起见，我使用了 (console.log('expired ...'), request$.onNext(true) ,data$.skip(1).take(1))) .这可能看起来有点神秘。它使用 js 逗号分隔的语法，这在 minifiers/uglifiers 中很常见。它执行所有语句并返回最后一条语句的结果。如果你想要一个更易读的代码，你可以像这样重写它:

.switchMap(data => {
  if(data.expires > new Date().getTime()){ 
    return data$.take(1);
  } else {
    console.log('expired ...');
    request$.onNext(true);
    return data$.skip(1).take(1);
  }
});

最后一部分是 flatMapFirst的用法.这确保一旦请求正在进行中，所有后续请求都会被丢弃。您可以在控制台打印输出中看到它的工作原理。 “从服务器加载”被打印多次，但实际序列只被调用一次，你会得到一个“从服务器加载完成”打印输出。这是对原始 refreshtoken 标志检查的更具 react 性的解决方案。

虽然我不需要保存的数据，但它被保存是因为你提到你可能想在以后的 session 中阅读它。

关于 rxjs 的一些提示:

而不是使用 setTimeout ，这会导致很多问题，你可以简单地做 Rx.Observable.timer(time_out_value).subscribe(...) .

创建一个 observable 很麻烦(你甚至不得不调用 next(...) 和 complete() )。您可以使用 Rx.Subject 以更简洁的方式执行此操作.请注意，您有此类的规范，BehaviorSubject和 ReplaySubject .这些类(class)值得了解并且可以提供很多帮助。

最后一点。这是一个相当大的挑战 :-) 我不熟悉您的服务器端代码和设计注意事项，但我对抑制调用的需要感到不舒服。除非您的后端有很好的理由，否则我的自然方法是使用 flatMap 并让最后一个请求“获胜”，即放弃之前未终止的调用并设置该值。

代码基于 rxjs 4(因此它可以在 jsbin 中运行)，如果您使用的是 angular2(因此是 rxjs 5)，则需要对其进行调整。看看 the migration guide .

============== 对史蒂夫其他问题的回答(在下面的评论中)=======

有 one article我可以推荐。它的标题说明了一切:-)

至于程序与功能方法，我会向服务添加另一个变量:

let token$ = data$.pluck('refreshtoken');

然后在需要时食用。

我的一般方法是首先映射我的数据流和关系，然后像一个优秀的“键盘水管工”(就像我们所有人一样)构建管道。我对服务的顶级草稿是(为了简洁，跳过 angular2 手续和提供者):

class UserService {
  data$: <as above>;
  token$: data$.pluck('refreshtoken');
  private request$: <as above>;

  refresh(){
    request.onNext(true);
  }
}

您可能需要做一些检查，以便 pluck不会失败。

然后，每个需要数据或 token 的组件都可以直接访问它。

现在让我们假设您有一个服务需要对数据或 token 的更改采取行动:

class SomeService {
  constructor(private userSvc: UserService){
    this.userSvc.token$.subscribe(() => this.doMyUpdates());
  }
}

如果您需要合成数据，即使用数据/ token 和一些本地数据:

Rx.Observable.combineLatest(this.userSvc.data$, this.myRelevantData$)
  .subscribe(([data, myData] => this.doMyUpdates(data.someField, myData.someField));

同样，其理念是您构建数据流和管道，将它们连接起来，然后您所要做的就是触发一些东西。

我提出的“迷你模式”是在触发序列后传递给服务并注册结果。让我们以自动完成为例:

class ACService {
   fetch(text: string): Observable<Array<string>> {
     return http.get(text).map(response => response.json().data;
   }
}

然后您必须在每次文本更改时调用它并将结果分配给您的组件:

<div class="suggestions" *ngFor="let suggestion; of suggestions | async;">
  <div>{{suggestion}}</div>
</div>

并在您的组件中:

onTextChange(text) {
  this.suggestions = acSVC.fetch(text);
}

但这也可以这样做:

class ACService {
   createFetcher(textStream: Observable<string>): Observable<Array<string>> {
     return textStream.flatMap(text => http.get(text))
       .map(response => response.json().data;
   }
}

然后在您的组件中:

textStream: Subject<string> = new Subject<string>();
suggestions: Observable<string>;

constructor(private acSVC: ACService){
  this.suggestions = acSVC.createFetcher(textStream);
}

onTextChange(text) {
  this.textStream.next(text);
}

模板代码保持不变。

这在这里似乎是一件小事，但是一旦应用程序变得更大，并且数据流变得复杂，这会更好地工作。你有一个保存数据的序列，你可以在任何需要的地方使用它，你甚至可以进一步转换它。例如，假设您需要知道建议的数量，在第一种方法中，一旦获得结果，您需要进一步查询以获取它，因此:

onTextChange(text) {
  this.suggestions = acSVC.fetch(text);
  this.suggestionsCount = suggestions.pluck('length'); // in a sequence
  // or
  this.suggestions.subscribe(suggestions => this.suggestionsCount = suggestions.length); // in a numeric variable.
}

现在在第二种方法中，您只需定义:

constructor(private acSVC: ACService){
  this.suggestions = acSVC.createFetcher(textStream);
  this.suggestionsCount = this.suggestions.pluck('length');
}

希望这可以帮助 :-)

在写作时，我试图反射(reflection)我使用这样的响应式(Reactive)的途径。毋庸置疑，在进行实验时，大量的 jsbin 和奇怪的失败是其中的重要组成部分。我认为有助于塑造我的方法的另一件事(尽管我目前没有使用它)是学习 redux 和阅读/尝试一些 ngrx(angular 的 redux 端口)。理念和方法甚至不允许您考虑程序化，因此您必须调整基于功能、数据、关系和流程的思维方式。