Refactor content stream factory to get rid of currying

x-pack/plugins/reporting/server/lib/content_stream.ts

@@ -7,12 +7,13 @@
 
 import { Duplex } from 'stream';
 import type { ElasticsearchClient } from 'src/core/server';
+import { ReportingCore } from '..';
 import { ReportDocument } from '../../common/types';
 
 type Callback = (error?: Error) => void;
 type SearchRequest = Required<Parameters<ElasticsearchClient['search']>>[0];
 
-export interface ContentStreamDocument {
+interface ContentStreamDocument {
   id: string;
   index: string;
   if_primary_term?: number;
@@ -103,3 +104,9 @@ export class ContentStream extends Duplex {
     return this.primaryTerm;
   }
 }
+
+export async function getContentStream(reporting: ReportingCore, document: ContentStreamDocument) {
+  const { asInternalUser: client } = await reporting.getEsClient();
+
+  return new ContentStream(client, document);
+}

x-pack/plugins/reporting/server/lib/index.ts

@@ -7,8 +7,7 @@
 
 export { checkLicense } from './check_license';
 export { checkParamsVersion } from './check_params_version';
-export { ContentStream } from './content_stream';
-export { getContentStreamFactory } from './content_stream_factory';
+export { ContentStream, getContentStream } from './content_stream';
 export { cryptoFactory } from './crypto';
 export { ExportTypesRegistry, getExportTypesRegistry } from './export_types_registry';
 export { LevelLogger } from './level_logger';

x-pack/plugins/reporting/server/lib/tasks/execute_report.ts

@@ -11,7 +11,7 @@ import { UpdateResponse } from '@elastic/elasticsearch/api/types';
 import moment from 'moment';
 import * as Rx from 'rxjs';
 import { timeout } from 'rxjs/operators';
-import { LevelLogger, getContentStreamFactory } from '../';
+import { LevelLogger, getContentStream } from '../';
 import { ReportingCore } from '../../';
 import {
   RunContext,
@@ -57,15 +57,13 @@ export class ExecuteReportTask implements ReportingTask {
   private kibanaId?: string;
   private kibanaName?: string;
   private store?: ReportingStore;
-  private getContentStream: ReturnType<typeof getContentStreamFactory>;
 
   constructor(
     private reporting: ReportingCore,
     private config: ReportingConfigType,
     logger: LevelLogger
   ) {
     this.logger = logger.clone(['runTask']);
-    this.getContentStream = getContentStreamFactory(reporting);
   }
 
   /*
@@ -256,7 +254,7 @@ export class ExecuteReportTask implements ReportingTask {
 
     const completedTime = moment().toISOString();
     const { content, ...docOutput } = this._formatOutput(output);
-    const stream = await this.getContentStream({
+    const stream = await getContentStream(this.reporting, {
       id: report._id,
       index: report._index!,
       if_primary_term: report._primary_term,

x-pack/plugins/reporting/server/routes/jobs.test.ts

@@ -5,8 +5,8 @@
  * 2.0.
  */
 
-jest.mock('../lib/content_stream_factory', () => ({
-  getContentStreamFactory: jest.fn(),
+jest.mock('../lib/content_stream', () => ({
+  getContentStream: jest.fn(),
 }));
 
 import { UnwrapPromise } from '@kbn/utility-types';
@@ -17,7 +17,7 @@ import { setupServer } from 'src/core/server/test_utils';
 import supertest from 'supertest';
 import { ReportingCore } from '..';
 import { ReportingInternalSetup } from '../core';
-import { ContentStream, ExportTypesRegistry, getContentStreamFactory } from '../lib';
+import { ContentStream, ExportTypesRegistry, getContentStream } from '../lib';
 import {
   createMockConfigSchema,
   createMockPluginSetup,
@@ -100,9 +100,7 @@ describe('GET /api/reporting/jobs/download', () => {
     mockEsClient = (await core.getEsClient()).asInternalUser as typeof mockEsClient;
     stream = ({ toString: jest.fn(() => 'test') } as unknown) as typeof stream;
 
-    (getContentStreamFactory as jest.MockedFunction<
-      typeof getContentStreamFactory
-    >).mockReturnValue(async () => stream);
+    (getContentStream as jest.MockedFunction<typeof getContentStream>).mockResolvedValue(stream);
   });
 
   afterEach(async () => {

x-pack/plugins/reporting/server/routes/lib/get_document_payload.ts

@@ -10,7 +10,7 @@ import contentDisposition from 'content-disposition';
 import { CSV_JOB_TYPE, CSV_JOB_TYPE_DEPRECATED } from '../../../common/constants';
 import { ReportApiJSON } from '../../../common/types';
 import { ReportingCore } from '../../';
-import { getContentStreamFactory, statuses } from '../../lib';
+import { getContentStream, statuses } from '../../lib';
 import { ExportTypeDefinition } from '../../types';
 
 export interface ErrorFromPayload {
@@ -48,7 +48,6 @@ const getReportingHeaders = (output: TaskRunResult, exportType: ExportTypeDefini
 
 export function getDocumentPayloadFactory(reporting: ReportingCore) {
   const exportTypesRegistry = reporting.getExportTypesRegistry();
-  const getContentStream = getContentStreamFactory(reporting);
 
   function encodeContent(
     content: string | null,
@@ -116,7 +115,7 @@ export function getDocumentPayloadFactory(reporting: ReportingCore) {
     payload: { title },
   }: ReportApiJSON): Promise<Payload> {
     if (output) {
-      const stream = await getContentStream({ id, index });
+      const stream = await getContentStream(reporting, { id, index });
       const content = await stream.toString();
 
       if (status === statuses.JOB_STATUS_COMPLETED || status === statuses.JOB_STATUS_WARNINGS) {