Implement/timestamps in log (#11781)

* [utils/streams] add createMapStream()

* [utils/toolingLog] make tooling log operate in object mode

* [ci/ftr] log the time before each log message

(cherry picked from commit 31fed421ff)

src/utils/index.js

@@ -17,4 +17,5 @@ export {
   createPromiseFromStreams,
   createReduceStream,
   createSplitStream,
+  createMapStream,
 } from './streams';

src/utils/streams/__tests__/map_stream.js

@@ -0,0 +1,47 @@
+import sinon from 'sinon';
+import expect from 'expect.js';
+import { delay } from 'bluebird';
+
+import { createPromiseFromStreams } from '../promise_from_streams';
+import { createListStream } from '../list_stream';
+import { createMapStream } from '../map_stream';
+import { createConcatStream } from '../concat_stream';
+
+describe('createMapStream()', () => {
+  it('calls the function with each item in the source stream', async () => {
+    const mapper = sinon.stub();
+
+    await createPromiseFromStreams([
+      createListStream([ 'a', 'b', 'c' ]),
+      createMapStream(mapper),
+    ]);
+
+    sinon.assert.calledThrice(mapper);
+    sinon.assert.calledWith(mapper, 'a', 0);
+    sinon.assert.calledWith(mapper, 'b', 1);
+    sinon.assert.calledWith(mapper, 'c', 2);
+  });
+
+  it('send the return value from the mapper on the output stream', async () => {
+    const result = await createPromiseFromStreams([
+      createListStream([ 1, 2, 3 ]),
+      createMapStream(n => n * 100),
+      createConcatStream([])
+    ]);
+
+    expect(result).to.eql([100, 200, 300]);
+  });
+
+  it('supports async mappers', async () => {
+    const result = await createPromiseFromStreams([
+      createListStream([ 1, 2, 3 ]),
+      createMapStream(async (n, i) => {
+        await delay(n);
+        return n * i;
+      }),
+      createConcatStream([])
+    ]);
+
+    expect(result).to.eql([ 0, 2, 6]);
+  });
+});

src/utils/streams/index.js

@@ -5,3 +5,4 @@ export { createReduceStream } from './reduce_stream';
 export { createJsonParseStream, createJsonStringifyStream } from './json_streams';
 export { createPromiseFromStreams } from './promise_from_streams';
 export { createConcatStream } from './concat_stream';
+export { createMapStream } from './map_stream';

src/utils/streams/map_stream.js

@@ -0,0 +1,17 @@
+import { Transform } from 'stream';
+
+export function createMapStream(fn) {
+  let i = 0;
+
+  return new Transform({
+    objectMode: true,
+    async transform(value, enc, done) {
+      try {
+        this.push(await fn(value, i++));
+        done();
+      } catch (err) {
+        done(err);
+      }
+    }
+  });
+}

src/utils/tooling_log/tooling_log.js

@@ -1,9 +1,10 @@
 import { format } from 'util';
 import { PassThrough } from 'stream';
 
-import { parseLogLevel } from './log_levels';
 import { magenta, yellow, red, blue, brightBlack } from 'ansicolors';
 
+import { parseLogLevel } from './log_levels';
+
 export function createToolingLog(initialLogLevelName = 'silent') {
   // current log level (see logLevel.name and logLevel.flags) changed
   // with ToolingLog#setLevel(newLogLevelName);
@@ -13,6 +14,10 @@ export function createToolingLog(initialLogLevelName = 'silent') {
   let indentString = '';
 
   class ToolingLog extends PassThrough {
+    constructor() {
+      super({ objectMode: true });
+    }
+
     verbose(...args) {
       if (!logLevel.flags.verbose) return;
       this.write(' %s ', magenta('sill'), format(...args));

tasks/functional_test_runner.js

@@ -1,12 +1,19 @@
 import { resolve } from 'path';
 
+import moment from 'moment';
+
 import { createFunctionalTestRunner } from '../src/functional_test_runner';
-import { createToolingLog } from '../src/utils';
+import { createToolingLog, createMapStream } from '../src/utils';
 
 export default function (grunt) {
   grunt.registerTask('functionalTestRunner', function () {
     const log = createToolingLog('debug');
-    log.pipe(process.stdout);
+
+    log
+      .pipe(createMapStream(line => {
+        return `${moment().format('hh:mm:ss.SSS')} ${line}`;
+      }))
+      .pipe(process.stdout);
 
     const functionalTestRunner = createFunctionalTestRunner({
       log,