Прочитать буферизованный ответ с помощью Angular2/RxJS

Question

Я создаю веб-сайт, который считывает данные из бэкэнд. Эти данные вычисляются «на лету» и отправляются обратно клиенту буферизованным образом. То есть как только первый кусок вычисляется, он отправляется клиенту, затем он вычисляет следующий фрагмент и отправляет его клиенту. Весь этот процесс выполняется в том же HTTP-запросе. Клиент не должен ждать завершения полного ответа, но обрабатывать каждый кусок самостоятельно, как только он был отправлен. Такие ответы обычно могут потребляться с использованием обработчика прогресса XHR (например, How to get progress from XMLHttpRequest).

Как я могу использовать такой ответ с помощью HttpModule в Angular2 с использованием RxJS и Observables?

---

Edit: 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 и возвращает Observable свойство Connection (это возвращается с 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 кусков каждый полсекунды. И это результат

Там, кажется, ошибка где-то. Там только девять :-). Я думаю, что это связано с сервером.