diff --git a/package.json b/package.json index cd564bd3393..69309e2e341 100644 --- a/package.json +++ b/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", diff --git a/public/app/plugins/datasource/testdata/StreamHandler.ts b/public/app/plugins/datasource/testdata/StreamHandler.ts index 11ede1346bd..389c20a8bf3 100644 --- a/public/app/plugins/datasource/testdata/StreamHandler.ts +++ b/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; + + 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): 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; diff --git a/public/app/plugins/datasource/testdata/partials/query.editor.html b/public/app/plugins/datasource/testdata/partials/query.editor.html index 768c30152be..e982ce9fb61 100644 --- a/public/app/plugins/datasource/testdata/partials/query.editor.html +++ b/public/app/plugins/datasource/testdata/partials/query.editor.html @@ -43,7 +43,7 @@ @@ -78,7 +78,16 @@ step="0.1" ng-change="ctrl.streamChanged()" /> -
+
+ + +
+
diff --git a/public/app/plugins/datasource/testdata/types.ts b/public/app/plugins/datasource/testdata/types.ts index c17097cd01f..73d562add92 100644 --- a/public/app/plugins/datasource/testdata/types.ts +++ b/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 } diff --git a/yarn.lock b/yarn.lock index b676eb5a8b4..624f08d556b 100644 --- a/yarn.lock +++ b/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"