RxJ를 사용하여 간격 당 API 호출 수를 제한하는 방법은 무엇입니까?

Question

각도 응용 프로그램을 개발 중입니다. 그리고 소셜 네트워크 중 하나에서 제공하는 API를 사용하며 초당 5 회의 API 호출 만 허용됩니다.

가장 직접적인 해결책은 요청을 계산하고 제한에 대기시키는 사용자 지정 논리를 작성하는 것입니다. 따라서 API에 6 번째 요청을 1 초 이내에 보내면 첫 번째 요청이 전송 된 후 2 번째로 전송됩니다.

그러나 RxJ를 사용하여 가능한 경우 우아한 해결책을 찾고 싶습니다.

예를 들어, 나는 다음과 같은 예처럼 관찰 가능한에 대한 debounseTime을 설정할 수 있습니다. 하지만 실제로 얻을 수있는 것은 그 사이에 200ms보다 작은 간격으로 몇 줄의 요청을 할 수 없다는 것입니다.

this.searchControl.valueChanges 
    .debounceTime(200) // 200ms ~ 5 requests per second 
    .switchMap(search => this.api.searchPeople(search)) 

RxJs는 간격 당 방출 수를 제한하고 요청이 너무 자주 전송되는 경우 대기열에 넣을 수있는 기법을 갖고 있습니까?

Answer 1

최근에 API를 호출 한 횟수를 계속 추적 할 수 있습니다. 따라서 초당 5 회의 통화를 할 수 있다면 5 개의 토큰이 있다는 것을 의미하며, 토큰이 소비되면 1 초 후에 갱신됩니다. 나는 당신이 필요하지 그 다음 연산자를했습니다 :

Observable.prototype.rateLimit = function (count: number, slidingWindowTime: number, scheduler = async) { 
 
    let tokens = count; 
 
    const tokenChanged = new BehaviorSubject(tokens); 
 
    const consumeToken =() => tokenChanged.next(--tokens); 
 
    const renewToken =() => tokenChanged.next(++tokens); 
 
    const availableTokens = tokenChanged.filter(() => tokens > 0); 
 

 
    return this.mergeMap(value => 
 
    availableTokens 
 
    .take(1) 
 
    .map(() => { 
 
     consumeToken(); 
 
     Observable.timer(slidingWindowTime, scheduler).subscribe(renewToken); 
 
     return value; 
 
    })); 
 
} 
 

 
declare module 'rxjs/Observable' { 
 
    interface Observable <T> { 
 
    rateLimit(count: number, slidingWindowTime: number, scheduler ? : Scheduler): Observable <T> 
 
    } 
 
}