Angular 2 Http、Observables 和递归请求

Question

我有一个 REST 端点，它返回一个项目列表，一次最多 1000 个项目。如果项目超过 1000 个，则响应的 HTTP 状态为 206，并且有一个 Next-Range 标头，我可以在下一个请求中使用它来获取更多项目。

我正在开发Angular 2 应用程序，并尝试使用Http 和Observable 来实现它。我的问题是我不知道如何合并多个Observables，具体取决于有多少页项目，最终返回一个Observable，我的组件可以订阅。

以下是我当前的 TypeScript 实现：

// 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();
}

但是，这不起作用。如果我理解正确，Observable 将作为初始 Observable 的值返回，而不是项目数组。

Answer 1

您可以使用扩展运算符来实现这一点。您真正想要做的是创建一个递归平面图。这正是运算符 expand 的用途。

这是它的工作原理的代码 sn-p：

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));

所做的是模拟第一个 http 请求，它的“下一个”属性为 true。这与您的 206 标头匹配。然后我们进行第二次调用，它的“下一个”属性为 false。

结果是一个包含两个请求结果的数组。借助扩展运算符，它也适用于更多请求。

可以在这里找到工作的 jsbin 示例：http://jsbin.com/wowituluqu/edit?js,console

编辑：更新为使用从数组返回数组的 http 调用，最终结果是包含数组中所有元素的单个数组。

如果您希望得到一个数组，其中包含来自请求的单独数组仍在其中，只需删除平面图并直接返回项目。在此处更新 codepen： http://codepen.io/anon/pen/xRZyaZ?editors=0010#0

Answer 2

我对 KwintenP 的示例进行了细微的调整：

// 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());
}

在订阅Observable 的组件中，我必须添加一个已完成的处理程序：

// 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;
  }
);

Answer 3

上面的答案很有用。我不得不以递归方式使用分页 API 获取数据，并创建了 code snippet 计算阶乘。

Answer 4

在最新版本中，angular 6+（响应本身返回 JSON），RxJs 6+（以管道方式使用运算符）。

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 5

以防万一其他人遇到这种情况。我正在使用的模式使用相同的扩展概念。然而，当您需要将来自服务器的响应转换为不同类型的Observable 时，这确实是一个“完整”的示例，就像上面 Visa Kopu 的示例一样。

我打破了每个“步骤”，以便在方法中捕获流程（而不是编写最紧凑的版本）。我觉得这样比较好学。

import {Injectable} from '@angular/core';
import {HttpClient, HttpParams, HttpResponse} from '@angular/common/http';
import {EMPTY, Observable} from 'rxjs';
import {expand, map} from 'rxjs/operators';

// this service is consuming a backend api that is calling/proxying a Salesforce query that is paginated
@Injectable({providedIn: 'root'})
export class ExampleAccountService {

    constructor(protected http: HttpClient) {
    }

    // this method maps the 'pages' of AccountsResponse objects to a single Observable array of Account objects
    allAccounts(): Observable<Account[]> {
        const accounts: Account[] = [];
        return this.aPageOfAccounts(null).pipe(
            map((ret: HttpResponse<AccountsResponse>) => {
                for (const account of ret.body.accounts) {
                    accounts.push(account);
                }
                return accounts;
            })
        );
    }

    // recursively fetch pages of accounts until there are no more pages
    private aPageOfAccounts(page): Observable<HttpResponse<AccountsResponse>> {
        return this.fetchAccountsFromServer(page).pipe(
            expand((res: HttpResponse<AccountsResponse>) => {
                if (res.body.nextRecordsUrl) {
                    return this.aPageOfAccounts(res.body.nextRecordsUrl);
                } else {
                    return EMPTY;
                }
            }));
    }

    // this one does the actual fetch to the server
    private fetchAccountsFromServer(page: string): Observable<HttpResponse<AccountsResponse>> {
        const options = createRequestOption({page});
        return this.http.get<AccountsResponse>(`https://wherever.com/accounts/page`,
            {params: options, observe: 'response'});
    }
}

export class AccountsResponse {
    constructor(public totalSize?: number,
                public done?: boolean,
                public nextRecordsUrl?: string,
                public accounts?: Account[]) {
    }
}

export class Account {
    constructor(public id?: string,
                public name?: string
    ) {

    }
}

export const createRequestOption = (req?: any): HttpParams => {
    let options: HttpParams = new HttpParams();
    if (req) {
        Object.keys(req).forEach((key) => {
            if (key !== 'sort') {
                options = options.set(key, req[key]);
            }
        });
        if (req.sort) {
            req.sort.forEach((val) => {
                options = options.append('sort', val);
            });
        }
    }
    return options;
};