使用Angular2/RxJS讀取緩存的響應

Question

我正在構建一個從後端讀取數據的網站。該數據即時計算並以緩衝方式發送回客戶端。即一旦第一個塊被計算出來，它被髮送到客戶端，然後計算下一個塊並將其發送到客戶端。整個過程發生在同一個HTTP請求中。客戶端不應該等待完成的響應完成，而應該儘快處理每個塊。通常可以使用XHR進度處理程序消耗這些響應（例如How to get progress from XMLHttpRequest）。

如何使用RxJS和Observables在Angular2中使用HttpModule來響應此類響應？

---

編輯：peeskillet下面給出一個很好的和詳細的解答。另外，我做了一些進一步的挖掘，發現了feature request for the HttpModule of Angular和StackOverflow question with another approach on how to solve it。

Answer 1

注意：以下答案只是一個POC。它旨在教育Http的體系結構，並提供簡單的工作POC實現。一個人應該看看XHRConnection的來源，看看在實現這個時你應該考慮什麼。

當試圖實現這一點時，我沒有看到任何直接進入XHR的方法。看起來也許我們需要提供一些與使用Http有關的組件的自定義實現。我們應該考慮的三個主要組成部分是

• Connection
• ConnectionBackend
• Http

Http需要ConnectionBackend作爲參數傳遞給它的構造。當發出請求時，使用get，Http與ConnectionBackend.createConnection創建連接，並返回ObservableConnection（即從createConnection返回）的屬性。在最精簡（簡化）視圖，它看起來像這樣

class XHRConnection implements Connection { 
    response: Observable<Response>; 
    constructor(request, browserXhr) { 
    this.response = new Observable((observer: Observer<Response>) => { 
     let xhr = browserXhr.create(); 
     let onLoad = (..) => { 
     observer.next(new Response(...)); 
     }; 
     xhr.addEventListener('load', onLoad); 
    }) 
    } 
} 

class XHRBackend implements ConnectionBackend { 
    constructor(private browserXhr) {} 
    createConnection(request): XHRConnection { 
    return new XHRConnection(request, this.broswerXhr).response; 
    } 
} 

class Http { 
    constructor(private backend: ConnectionBackend) {} 

    get(url, options): Observable<Response> { 
    return this.backend.createConnection(createRequest(url, options)).response; 
    } 
} 

因此，瞭解這個架構，我們可以嘗試實行類似的東西。

對於Connection，這裏是POC。爲簡潔起見，省略了進口貨物，但大部分貨物可以從@angular/http導入，Observable/Observer可以從rxjs/{Type}導入。

export class Chunk { 
    data: string; 
} 

export class ChunkedXHRConnection implements Connection { 
    request: Request; 
    response: Observable<Response>; 
    readyState: ReadyState; 

    chunks: Observable<Chunk>; 

    constructor(req: Request, browserXHR: BrowserXhr, baseResponseOptions?: ResponseOptions) { 
    this.request = req; 
    this.chunks = new Observable<Chunk>((chunkObserver: Observer<Chunk>) => { 
     let _xhr: XMLHttpRequest = browserXHR.build(); 
     let previousLen = 0; 
     let onProgress = (progress: ProgressEvent) => { 
     let text = _xhr.responseText; 
     text = text.substring(previousLen); 
     chunkObserver.next({ data: text }); 
     previousLen += text.length; 

     console.log(`chunk data: ${text}`); 
     }; 
     _xhr.addEventListener('progress', onProgress); 
     _xhr.open(RequestMethod[req.method].toUpperCase(), req.url); 
     _xhr.send(this.request.getBody()); 
     return() => { 
     _xhr.removeEventListener('progress', onProgress); 
     _xhr.abort(); 
     }; 
    }); 
    } 
} 

這裏我們只是訂閱XHR progress事件。由於XHR.responseText會將所有連接文字分開，因此我們只需要substring即可獲得塊，並通過Observer發出每個卡盤。

對於XHRBackend，我們有以下（沒有什麼特別的）。再次，一切都可以從@angular/http導入;

@Injectable() 
export class ChunkedXHRBackend implements ConnectionBackend { 
    constructor(
     private _browserXHR: BrowserXhr, private _baseResponseOptions: ResponseOptions, 
     private _xsrfStrategy: XSRFStrategy) {} 

    createConnection(request: Request): ChunkedXHRConnection { 
    this._xsrfStrategy.configureRequest(request); 
    return new ChunkedXHRConnection(request, this._browserXHR, this._baseResponseOptions); 
    } 
} 

對於Http，我們將其擴展，加入getChunks方法。如果你願意，你可以添加更多的方法。

@Injectable() 
export class ChunkedHttp extends Http { 
    constructor(protected backend: ChunkedXHRBackend, protected defaultOptions: RequestOptions) { 
    super(backend, defaultOptions); 
    } 

    getChunks(url, options?: RequestOptionsArgs): Observable<Chunk> { 
    return this.backend.createConnection(
     new Request(mergeOptions(this.defaultOptions, options, RequestMethod.Get, url))).chunks; 
    } 
} 

的mergeOptions方法可以在Http source找到。

現在我們可以爲它創建一個模塊。用戶應該直接使用ChunkedHttp而不是Http。但是因爲不要試圖覆蓋Http令牌，所以如果需要，仍然可以使用Http。

@NgModule({ 
    imports: [ HttpModule ], 
    providers: [ 
    { 
     provide: ChunkedHttp, 
     useFactory: (backend: ChunkedXHRBackend, options: RequestOptions) => { 
     return new ChunkedHttp(backend, options); 
     }, 
     deps: [ ChunkedXHRBackend, RequestOptions ] 
    }, 
    ChunkedXHRBackend 
    ] 
}) 
export class ChunkedHttpModule { 
} 

我們進口的HttpModule因爲它提供了我們需要注入其他服務，但我們不希望有重新實現這些，如果我們不需要。

要測試只需將ChunkedHttpModule導入AppModule。此外，測試我用下面的成分

@Component({ 
    selector: 'app', 
    encapsulation: ViewEncapsulation.None, 
    template: ` 
    <button (click)="onClick()">Click Me!</button> 
    <h4 *ngFor="let chunk of chunks">{{ chunk }}</h4> 
    `, 
    styleUrls: ['./app.style.css'] 
}) 
export class App { 
    chunks: string[] = []; 

    constructor(private http: ChunkedHttp) {} 

    onClick() { 
    this.http.getChunks('http://localhost:8080/api/resource') 
     .subscribe(chunk => this.chunks.push(chunk.data)); 
    } 
} 

我有一個後端端點設置它只是吐出每半秒"Message #x" 10塊。這是結果

似乎是一個錯誤的地方。只有九個 :-)。我認爲這是服務器端相關的。