TestData: stream via fetch (#16963)

Stream from fetch CSV

package.json

@@ -202,6 +202,7 @@
     "d3-scale-chromatic": "1.3.3",
     "eventemitter3": "2.0.3",
     "file-saver": "1.3.8",
+    "fast-text-encoding": "^1.0.0",
     "immutable": "3.8.2",
     "jquery": "3.4.0",
     "lodash": "4.17.11",

public/app/plugins/datasource/testdata/StreamHandler.ts

@@ -9,6 +9,7 @@ import {
   DataStreamState,
   LoadingState,
   LogLevel,
+  CSVReader,
 } from '@grafana/ui';
 import { TestDataQuery, StreamingQuery } from './types';
 
@@ -57,6 +58,8 @@ export class StreamHandler {
         this.workers[key] = new SignalWorker(key, query, req, observer);
       } else if (type === 'logs') {
         this.workers[key] = new LogsWorker(key, query, req, observer);
+      } else if (type === 'fetch') {
+        this.workers[key] = new FetchWorker(key, query, req, observer);
       } else {
         throw {
           message: 'Unknown Stream type: ' + type,
@@ -72,6 +75,7 @@ export class StreamHandler {
  * Manages a single stream request
  */
 export class StreamWorker {
+  refId: string;
   query: StreamingQuery;
   stream: DataStreamState;
   observer: DataStreamObserver;
@@ -85,6 +89,7 @@ export class StreamWorker {
       request,
       unsubscribe: this.unsubscribe,
     };
+    this.refId = query.refId;
     this.query = query.stream;
     this.last = Date.now();
     this.observer = observer;
@@ -207,6 +212,56 @@ export class SignalWorker extends StreamWorker {
   };
 }
 
+export class FetchWorker extends StreamWorker {
+  csv: CSVReader;
+  reader: ReadableStreamReader<Uint8Array>;
+
+  constructor(key: string, query: TestDataQuery, request: DataQueryRequest, observer: DataStreamObserver) {
+    super(key, query, request, observer);
+    if (!query.stream.url) {
+      throw new Error('Missing Fetch URL');
+    }
+    if (!query.stream.url.startsWith('http')) {
+      throw new Error('Fetch URL must be absolute');
+    }
+
+    this.csv = new CSVReader({ callback: this });
+    fetch(new Request(query.stream.url)).then(response => {
+      this.reader = response.body.getReader();
+      this.reader.read().then(this.processChunk);
+    });
+  }
+
+  processChunk = (value: ReadableStreamReadResult<Uint8Array>): any => {
+    if (this.observer == null) {
+      return; // Nothing more to do
+    }
+
+    if (value.value) {
+      const text = new TextDecoder().decode(value.value);
+      this.csv.readCSV(text);
+    }
+
+    if (value.done) {
+      console.log('Finished stream');
+      this.stream.state = LoadingState.Done;
+      return;
+    }
+
+    return this.reader.read().then(this.processChunk);
+  };
+
+  onHeader = (series: SeriesData) => {
+    series.refId = this.refId;
+    this.stream.series = [series];
+  };
+
+  onRow = (row: any[]) => {
+    // TODO?? this will send an event for each row, even if the chunk passed a bunch of them
+    this.appendRows([row]);
+  };
+}
+
 export class LogsWorker extends StreamWorker {
   index = 0;
 

public/app/plugins/datasource/testdata/partials/query.editor.html

@@ -43,7 +43,7 @@
 				<select 
 				  ng-model="ctrl.target.stream.type" 
 				  class="gf-form-input"
-					ng-options="type for type in ['signal','logs']"
+					ng-options="type for type in ['signal','logs', 'fetch']"
 				  ng-change="ctrl.streamChanged()" />
 				</select>
 			</div>
@@ -78,7 +78,16 @@
 				step="0.1"
 				ng-change="ctrl.streamChanged()" />
 		</div>
-		<div class="gf-form gf-form--grow">
+		<div class="gf-form gf-form--grow" ng-if="ctrl.target.stream.type === 'fetch'">
+			<label class="gf-form-label query-keyword">URL</label>
+			<input type="string" 
+				class="gf-form-input gf-form-label--grow" 
+				placeholder="Fetch URL" 
+				ng-model="ctrl.target.stream.url" 
+				ng-change="ctrl.streamChanged()" 
+				ng-model-onblur />
+		</div>
+		<div class="gf-form gf-form--grow" ng-if="ctrl.target.stream.type !== 'fetch'">
 			<div class="gf-form-label gf-form-label--grow"></div>
 		</div>
 	</div>

public/app/plugins/datasource/testdata/types.ts

@@ -14,9 +14,10 @@ export interface TestDataQuery extends DataQuery {
 }
 
 export interface StreamingQuery {
-  type: 'signal' | 'logs';
+  type: 'signal' | 'logs' | 'fetch';
   speed: number;
   spread: number;
   noise: number; // wiggle around the signal for min/max
   buffer?: number;
+  url?: string; // the Fetch URL
 }

yarn.lock

@@ -7351,6 +7351,11 @@ fast-levenshtein@^2.0.6, fast-levenshtein@~2.0.4:
   resolved "https://registry.yarnpkg.com/fast-levenshtein/-/fast-levenshtein-2.0.6.tgz#3d8a5c66883a16a30ca8643e851f19baa7797917"
   integrity sha1-PYpcZog6FqMMqGQ+hR8Zuqd5eRc=
 
+fast-text-encoding@^1.0.0:
+  version "1.0.0"
+  resolved "https://registry.yarnpkg.com/fast-text-encoding/-/fast-text-encoding-1.0.0.tgz#3e5ce8293409cfaa7177a71b9ca84e1b1e6f25ef"
+  integrity sha512-R9bHCvweUxxwkDwhjav5vxpFvdPGlVngtqmx4pIZfSUhM/Q4NiIUHB456BAf+Q1Nwu3HEZYONtu+Rya+af4jiQ==
+
 fastparse@^1.1.1:
   version "1.1.2"
   resolved "https://registry.yarnpkg.com/fastparse/-/fastparse-1.1.2.tgz#91728c5a5942eced8531283c79441ee4122c35a9"