[Step 1] use Observables on server search API

docs/development/plugins/data/server/kibana-plugin-plugins-data-server.isearchstart.md

@@ -16,6 +16,6 @@ export interface ISearchStart<SearchStrategyRequest extends IKibanaSearchRequest
 |  --- | --- | --- |
 |  [aggs](./kibana-plugin-plugins-data-server.isearchstart.aggs.md) | <code>AggsStart</code> |  |
 |  [getSearchStrategy](./kibana-plugin-plugins-data-server.isearchstart.getsearchstrategy.md) | <code>(name: string) =&gt; ISearchStrategy&lt;SearchStrategyRequest, SearchStrategyResponse&gt;</code> | Get other registered search strategies. For example, if a new strategy needs to use the already-registered ES search strategy, it can use this function to accomplish that. |
-|  [search](./kibana-plugin-plugins-data-server.isearchstart.search.md) | <code>(context: RequestHandlerContext, request: SearchStrategyRequest, options: ISearchOptions) =&gt; Promise&lt;SearchStrategyResponse&gt;</code> |  |
+|  [search](./kibana-plugin-plugins-data-server.isearchstart.search.md) | <code>ISearchStrategy['search']</code> |  |
 |  [searchSource](./kibana-plugin-plugins-data-server.isearchstart.searchsource.md) | <code>{</code><br/><code>        asScoped: (request: KibanaRequest) =&gt; Promise&lt;ISearchStartSearchSource&gt;;</code><br/><code>    }</code> |  |
 

docs/development/plugins/data/server/kibana-plugin-plugins-data-server.isearchstart.search.md

@@ -7,5 +7,5 @@
 <b>Signature:</b>
 
 ```typescript
-search: (context: RequestHandlerContext, request: SearchStrategyRequest, options: ISearchOptions) => Promise<SearchStrategyResponse>;
+search: ISearchStrategy['search'];
 ```

docs/development/plugins/data/server/kibana-plugin-plugins-data-server.isearchstrategy.md

@@ -17,5 +17,5 @@ export interface ISearchStrategy<SearchStrategyRequest extends IKibanaSearchRequ
 |  Property | Type | Description |
 |  --- | --- | --- |
 |  [cancel](./kibana-plugin-plugins-data-server.isearchstrategy.cancel.md) | <code>(context: RequestHandlerContext, id: string) =&gt; Promise&lt;void&gt;</code> |  |
-|  [search](./kibana-plugin-plugins-data-server.isearchstrategy.search.md) | <code>(context: RequestHandlerContext, request: SearchStrategyRequest, options?: ISearchOptions) =&gt; Promise&lt;SearchStrategyResponse&gt;</code> |  |
+|  [search](./kibana-plugin-plugins-data-server.isearchstrategy.search.md) | <code>(request: SearchStrategyRequest, options: ISearchOptions, context: RequestHandlerContext) =&gt; Observable&lt;SearchStrategyResponse&gt;</code> |  |
 

docs/development/plugins/data/server/kibana-plugin-plugins-data-server.isearchstrategy.search.md

@@ -7,5 +7,5 @@
 <b>Signature:</b>
 
 ```typescript
-search: (context: RequestHandlerContext, request: SearchStrategyRequest, options?: ISearchOptions) => Promise<SearchStrategyResponse>;
+search: (request: SearchStrategyRequest, options: ISearchOptions, context: RequestHandlerContext) => Observable<SearchStrategyResponse>;
 ```

examples/search_examples/server/my_strategy.ts

@@ -17,6 +17,7 @@
  * under the License.
  */
 
+import { map } from 'rxjs/operators';
 import { ISearchStrategy, PluginStart } from '../../../src/plugins/data/server';
 import { IMyStrategyResponse, IMyStrategyRequest } from '../common';
 
@@ -25,13 +26,13 @@ export const mySearchStrategyProvider = (
 ): ISearchStrategy<IMyStrategyRequest, IMyStrategyResponse> => {
   const es = data.search.getSearchStrategy('es');
   return {
-    search: async (context, request, options): Promise<IMyStrategyResponse> => {
-      const esSearchRes = await es.search(context, request, options);
-      return {
-        ...esSearchRes,
-        cool: request.get_cool ? 'YES' : 'NOPE',
-      };
-    },
+    search: (request, options, context) =>
+      es.search(request, options, context).pipe(
+        map((esSearchRes) => ({
+          ...esSearchRes,
+          cool: request.get_cool ? 'YES' : 'NOPE',
+        }))
+      ),
     cancel: async (context, id) => {
       if (es.cancel) {
         es.cancel(context, id);

examples/search_examples/server/routes/server_search_route.ts

@@ -39,26 +39,28 @@ export function registerServerSearchRoute(router: IRouter, data: DataPluginStart
       // Run a synchronous search server side, by enforcing a high keepalive and waiting for completion.
       // If you wish to run the search with polling (in basic+), you'd have to poll on the search API.
       // Please reach out to the @app-arch-team if you need this to be implemented.
-      const res = await data.search.search(
-        context,
-        {
-          params: {
-            index,
-            body: {
-              aggs: {
-                '1': {
-                  avg: {
-                    field,
+      const res = await data.search
+        .search(
+          {
+            params: {
+              index,
+              body: {
+                aggs: {
+                  '1': {
+                    avg: {
+                      field,
+                    },
                   },
                 },
               },
+              waitForCompletionTimeout: '5m',
+              keepAlive: '5m',
             },
-            waitForCompletionTimeout: '5m',
-            keepAlive: '5m',
-          },
-        } as IEsSearchRequest,
-        {}
-      );
+          } as IEsSearchRequest,
+          {},
+          context
+        )
+        .toPromise();
 
       return response.ok({
         body: {

src/plugins/data/public/search/search_service.ts

@@ -127,7 +127,7 @@ export class SearchService implements Plugin<ISearchSetup, ISearchStart> {
         request: SearchStrategyRequest,
         options: ISearchOptions
       ) => {
-        return search(request, options).toPromise() as Promise<SearchStrategyResponse>;
+        return search<SearchStrategyRequest, SearchStrategyResponse>(request, options).toPromise();
       },
       onResponse: handleResponse,
       legacy: {

src/plugins/data/server/search/es_search/es_search_strategy.test.ts

@@ -35,7 +35,8 @@ describe('ES search strategy', () => {
       },
     },
   });
-  const mockContext = {
+
+  const mockContext = ({
     core: {
       uiSettings: {
         client: {
@@ -44,7 +45,8 @@ describe('ES search strategy', () => {
       },
       elasticsearch: { client: { asCurrentUser: { search: mockApiCaller } } },
     },
-  };
+  } as unknown) as RequestHandlerContext;
+
   const mockConfig$ = pluginInitializerContextConfigMock<any>({}).legacy.globalConfig$;
 
   beforeEach(() => {
@@ -57,44 +59,51 @@ describe('ES search strategy', () => {
     expect(typeof esSearch.search).toBe('function');
   });
 
-  it('calls the API caller with the params with defaults', async () => {
+  it('calls the API caller with the params with defaults', async (done) => {
     const params = { index: 'logstash-*' };
-    const esSearch = await esSearchStrategyProvider(mockConfig$, mockLogger);
 
-    await esSearch.search((mockContext as unknown) as RequestHandlerContext, { params });
-
-    expect(mockApiCaller).toBeCalled();
-    expect(mockApiCaller.mock.calls[0][0]).toEqual({
-      ...params,
-      ignore_unavailable: true,
-      track_total_hits: true,
-    });
+    await esSearchStrategyProvider(mockConfig$, mockLogger)
+      .search({ params }, {}, mockContext)
+      .subscribe(() => {
+        expect(mockApiCaller).toBeCalled();
+        expect(mockApiCaller.mock.calls[0][0]).toEqual({
+          ...params,
+          ignore_unavailable: true,
+          track_total_hits: true,
+        });
+        done();
+      });
   });
 
-  it('calls the API caller with overridden defaults', async () => {
+  it('calls the API caller with overridden defaults', async (done) => {
     const params = { index: 'logstash-*', ignore_unavailable: false, timeout: '1000ms' };
-    const esSearch = await esSearchStrategyProvider(mockConfig$, mockLogger);
-
-    await esSearch.search((mockContext as unknown) as RequestHandlerContext, { params });
 
-    expect(mockApiCaller).toBeCalled();
-    expect(mockApiCaller.mock.calls[0][0]).toEqual({
-      ...params,
-      track_total_hits: true,
-    });
+    await esSearchStrategyProvider(mockConfig$, mockLogger)
+      .search({ params }, {}, mockContext)
+      .subscribe(() => {
+        expect(mockApiCaller).toBeCalled();
+        expect(mockApiCaller.mock.calls[0][0]).toEqual({
+          ...params,
+          track_total_hits: true,
+        });
+        done();
+      });
   });
 
-  it('has all response parameters', async () => {
-    const params = { index: 'logstash-*' };
-    const esSearch = await esSearchStrategyProvider(mockConfig$, mockLogger);
-
-    const response = await esSearch.search((mockContext as unknown) as RequestHandlerContext, {
-      params,
-    });
-
-    expect(response.isRunning).toBe(false);
-    expect(response.isPartial).toBe(false);
-    expect(response).toHaveProperty('loaded');
-    expect(response).toHaveProperty('rawResponse');
-  });
+  it('has all response parameters', async (done) =>
+    await esSearchStrategyProvider(mockConfig$, mockLogger)
+      .search(
+        {
+          params: { index: 'logstash-*' },
+        },
+        {},
+        mockContext
+      )
+      .subscribe((data) => {
+        expect(data.isRunning).toBe(false);
+        expect(data.isPartial).toBe(false);
+        expect(data).toHaveProperty('loaded');
+        expect(data).toHaveProperty('rawResponse');
+        done();
+      }));
 });

src/plugins/data/server/search/es_search/es_search_strategy.ts

@@ -16,10 +16,10 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+import { Observable, from } from 'rxjs';
 import { first } from 'rxjs/operators';
 import { SharedGlobalConfig, Logger } from 'kibana/server';
 import { SearchResponse } from 'elasticsearch';
-import { Observable } from 'rxjs';
 import { ApiResponse } from '@elastic/elasticsearch';
 import { SearchUsage } from '../collectors/usage';
 import { toSnakeCase } from './to_snake_case';
@@ -29,6 +29,7 @@ import {
   getTotalLoaded,
   getShardTimeout,
   shimAbortSignal,
+  IEsSearchResponse,
 } from '..';
 
 export const esSearchStrategyProvider = (
@@ -37,47 +38,52 @@ export const esSearchStrategyProvider = (
   usage?: SearchUsage
 ): ISearchStrategy => {
   return {
-    search: async (context, request, options) => {
-      logger.debug(`search ${request.params?.index}`);
-      const config = await config$.pipe(first()).toPromise();
-      const uiSettingsClient = await context.core.uiSettings.client;
+    search: (request, options, context) =>
+      from(
+        new Promise<IEsSearchResponse>(async (resolve, reject) => {
+          logger.debug(`search ${request.params?.index}`);
+          const config = await config$.pipe(first()).toPromise();
+          const uiSettingsClient = await context.core.uiSettings.client;
 
-      // Only default index pattern type is supported here.
-      // See data_enhanced for other type support.
-      if (!!request.indexType) {
-        throw new Error(`Unsupported index pattern type ${request.indexType}`);
-      }
+          // Only default index pattern type is supported here.
+          // See data_enhanced for other type support.
+          if (!!request.indexType) {
+            throw new Error(`Unsupported index pattern type ${request.indexType}`);
+          }
 
-      // ignoreThrottled is not supported in OSS
-      const { ignoreThrottled, ...defaultParams } = await getDefaultSearchParams(uiSettingsClient);
+          // ignoreThrottled is not supported in OSS
+          const { ignoreThrottled, ...defaultParams } = await getDefaultSearchParams(
+            uiSettingsClient
+          );
 
-      const params = toSnakeCase({
-        ...defaultParams,
-        ...getShardTimeout(config),
-        ...request.params,
-      });
+          const params = toSnakeCase({
+            ...defaultParams,
+            ...getShardTimeout(config),
+            ...request.params,
+          });
 
-      try {
-        const promise = shimAbortSignal(
-          context.core.elasticsearch.client.asCurrentUser.search(params),
-          options?.abortSignal
-        );
-        const { body: rawResponse } = (await promise) as ApiResponse<SearchResponse<any>>;
+          try {
+            const promise = shimAbortSignal(
+              context.core.elasticsearch.client.asCurrentUser.search(params),
+              options?.abortSignal
+            );
+            const { body: rawResponse } = (await promise) as ApiResponse<SearchResponse<any>>;
 
-        if (usage) usage.trackSuccess(rawResponse.took);
+            if (usage) usage.trackSuccess(rawResponse.took);
 
-        // The above query will either complete or timeout and throw an error.
-        // There is no progress indication on this api.
-        return {
-          isPartial: false,
-          isRunning: false,
-          rawResponse,
-          ...getTotalLoaded(rawResponse._shards),
-        };
-      } catch (e) {
-        if (usage) usage.trackError();
-        throw e;
-      }
-    },
+            // The above query will either complete or timeout and throw an error.
+            // There is no progress indication on this api.
+            resolve({
+              isPartial: false,
+              isRunning: false,
+              rawResponse,
+              ...getTotalLoaded(rawResponse._shards),
+            });
+          } catch (e) {
+            if (usage) usage.trackError();
+            reject(e);
+          }
+        })
+      ),
   };
 };

src/plugins/data/server/search/routes/search.test.ts

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-import { Observable } from 'rxjs';
+import { Observable, from } from 'rxjs';
 
 import {
   CoreSetup,
@@ -66,7 +66,8 @@ describe('Search service', () => {
         },
       },
     };
-    mockDataStart.search.search.mockResolvedValue(response);
+
+    mockDataStart.search.search.mockReturnValue(from(Promise.resolve(response)));
     const mockContext = {};
     const mockBody = { id: undefined, params: {} };
     const mockParams = { strategy: 'foo' };
@@ -83,20 +84,24 @@ describe('Search service', () => {
     await handler((mockContext as unknown) as RequestHandlerContext, mockRequest, mockResponse);
 
     expect(mockDataStart.search.search).toBeCalled();
-    expect(mockDataStart.search.search.mock.calls[0][1]).toStrictEqual(mockBody);
+    expect(mockDataStart.search.search.mock.calls[0][0]).toStrictEqual(mockBody);
     expect(mockResponse.ok).toBeCalled();
     expect(mockResponse.ok.mock.calls[0][0]).toEqual({
       body: response,
     });
   });
 
   it('handler throws an error if the search throws an error', async () => {
-    mockDataStart.search.search.mockRejectedValue({
-      message: 'oh no',
-      body: {
-        error: 'oops',
-      },
-    });
+    const rejectedValue = from(
+      Promise.reject({
+        message: 'oh no',
+        body: {
+          error: 'oops',
+        },
+      })
+    );
+
+    mockDataStart.search.search.mockReturnValue(rejectedValue);
 
     const mockContext = {};
     const mockBody = { id: undefined, params: {} };
@@ -114,7 +119,7 @@ describe('Search service', () => {
     await handler((mockContext as unknown) as RequestHandlerContext, mockRequest, mockResponse);
 
     expect(mockDataStart.search.search).toBeCalled();
-    expect(mockDataStart.search.search.mock.calls[0][1]).toStrictEqual(mockBody);
+    expect(mockDataStart.search.search.mock.calls[0][0]).toStrictEqual(mockBody);
     expect(mockResponse.customError).toBeCalled();
     const error: any = mockResponse.customError.mock.calls[0][0];
     expect(error.body.message).toBe('oh no');