Angular 2 Http, Observables and recursive requests

Question

I have a REST endpoint that returns a list of items, max 1000 items at a time. If there are more than 1000 items, the response has HTTP status 206 and there's a Next-Range header that I can use in my next request for getting more items.

I'm working on an Angular 2 application and trying to implement this with Http and Observable. My problem is that I don't know how to merge multiple Observables depending on how many pages of items there are and finally return one Observable that my component can subscribe to.

Here's where I've got with my current TypeScript implementation:

// NOTE: Non-working example!

getAllItems(): Observable<any[]> {
  // array of all items, possibly received with multiple requests
  const allItems: any[] = [];

  // inner function for getting a range of items
  const getRange = (range?: string) => {
    const headers: Headers = new Headers();
    if (range) {
      headers.set('Range', range);
    }

    return this.http.get('http://api/endpoint', { headers })
      .map((res: Response) => {
        // add all to received items
        // (maybe not needed if the responses can be merged some other way?)
        allItems.push.apply(allItems, res.json());

        // partial content
        if (res.status === 206) {
          const nextRange = res.headers.get('Next-Range');

          // get next range of items
          return getRange(nextRange);
        }

        return allItems;
      });
  };

  // get first range
  return getRange();
}

However, this doesn't work. If I understood it correctly, an Observable is returned as the value of the initial Observable and not the array of items.

Answer 1

You can implement this using the expand operator. What you actually want to do is create a recursive flatmap. That's exactly what the operator expand was created for.

Here is the code snippet of how this works:

let times = true;
// This is a mock method for your http.get call
const httpMock = () => {
  if(times) {
    times = false;
    return Rx.Observable.of({items: ["1", "2", "3"], next: true});
  } else {
    return Rx.Observable.of({items: ["4", "5", "6"], next: false});
  }
}

httpMock()
  .expand(obj => {
    // In your case, the obj will be the response
    // implement your logic here if the 206 http header is found
    if(obj.next) {
      // If you have next values, just call the http.get method again
      // In my example it's the httpMock
      return httpMock();
    } else {
      return Rx.Observable.empty();
    }
  })
  .map(obj => obj.items.flatMap(array => array)) 
  .reduce((acc, x) => acc.concat(x), []);
  .subscribe((val) => console.log(val));

What is does is mock a first http request, that has a 'next' property to true. This matches your 206 header. We then make a second call which has the 'next' property to false.

The result is an array containing the results from both the requests. It's applicable for more requests as well thanks to the expand operator.

Working jsbin example can be found here: http://jsbin.com/wowituluqu/edit?js,console

EDIT: updated to work with an http call that returns an array from arrays and the end result is a single array that contains all the elements form the arrays.

If you wish to have as a result an array with the separate arrays from the request still inside, just remove the flatmap and return the items directly. Update codepen here: http://codepen.io/anon/pen/xRZyaZ?editors=0010#0

Answer 2

On latest version, angular 6+ (response by itself returns JSON), RxJs 6+ (Uses operators in pipeable fashion).

getAllItems(): Observable<any[]> {
  const getRange = (range?: string): Observable<any> => {
    const headers: Headers = new Headers();
    if (range) {
      headers.set('Range', range);
    }
    return this.http.get('http://api/endpoint', { headers });
  };

  return getRange().pipe(expand((res: Response) => {
    if (res['status'] === 206) {
      const nextRange = res['headers'].get('Next-Range');
      return getRange(nextRange);
    } else {
      return EMPTY;
    }
  }));
}

Answer 3

I got it working with minor tweaks to KwintenP's example:

// service.ts

getAllItems(): Observable<any[]> {
  const getRange = (range?: string): Observable<any> => {
    const headers: Headers = new Headers();
    if (range) {
      headers.set('Range', range);
    }

    return this.http.get('http://api/endpoint', { headers });
  };

  return getRange().expand((res: Response) => {
    if (res.status === 206) {
      const nextRange = res.headers.get('Next-Range');

      return getRange(nextRange);
    } else {
      return Observable.empty();
    }
  }).map((res: Response) => res.json());
}

In the component that subscribes to the Observable, I had to add a completed handler:

// component.ts

const temp = [];

service.getAllItems().subscribe(
  items => {
    // page received, push items to temp
    temp.push.apply(temp, items);
  },
  err => {
    // handle error
  },
  () => {
    // completed, expose temp to component
    this.items = temp;
  }
);