ElasticRx (RxJS Observables for @elastic/elasticsearch)

elastic-rx.ts

import { ApiResponse } from '@elastic/elasticsearch';
import {
  TransportRequestCallback,
  TransportRequestOptions,
} from '@elastic/elasticsearch/lib/Transport';
import { Observable } from 'rxjs';

/**
 * Use ElasticSearch API with RxJS observables that can be
 * cancelled. Observables are cold -- they must be subscribed to make
 * any request to the server.
 *
 * Usage:
 * ```
 * import { Client } from '@elastic/elasticsearch';
 * import { ElasticRx as er } from './elastic-rx';
 *
 * const client = new Client({
 *   node: 'http://localhost:9200/',
 * });
 *
 * // Get request as Observable<ApiResponse>
 * const request = er.from(client.index, {
 *   index: 'text',
 *   body: { message: 'hello world' },
 * });
 * ```
 */
export const ElasticRx = {
  from,
};

function from<TApiMethod>(
  apiMethod: TApiMethod,
  params?: ParamsOfApiMethod<TApiMethod>,
  options?: TransportRequestOptions
) {
  return new Observable<ApiResponse<BodyOfApiMethod<TApiMethod>>>(observer => {
    const requestCallback: callbackFn<any> = (error, result) => {
      if (error) {
        return observer.error(error);
      }

      observer.next(result);
      observer.complete();
    };

    const requestParams = [];

    if (params) {
      requestParams.push(params);

      if (options) {
        requestParams.push(options);
      }
    }

    const request = (apiMethod as any)(
      ...requestParams,
      requestCallback
    ) as TransportRequestCallback;

    return () => request.abort();
  });
}

// Copied from @elastic/elasticsearch. If exported, would look like:
// import { callbackFn } from '@elastic/elasticsearch';
declare type callbackFn<T> = (
  err: Error | null,
  result: ApiResponse<T>
) => void;

// Copied from @elastic/elasticsearch. If exported, would look like:
// import { ApiMethod } from '@elastic/elasticsearch';
interface ApiMethod<TParams, TBody = any> {
  // Promise API
  (): Promise<ApiResponse<TBody>>;
  (params: TParams): Promise<ApiResponse<TBody>>;
  (params: TParams, options: TransportRequestOptions): Promise<
    ApiResponse<TBody>
  >;

  // Callback API
  (callback: callbackFn<TBody>): TransportRequestCallback;
  (params: TParams, callback: callbackFn<TBody>): TransportRequestCallback;
  (
    params: TParams,
    options: TransportRequestOptions,
    callback: callbackFn<TBody>
  ): TransportRequestCallback;
}

// `infer` feature requires TypeScript 2.8.
// https://www.typescriptlang.org/docs/handbook/release-notes/typescript-2-8.html
type ParamsOfApiMethod<T> = T extends ApiMethod<infer TParams> ? TParams : any;
type BodyOfApiMethod<T> = T extends ApiMethod<any, infer TBody> ? TBody : any;