Streaming: use StreamingDataFrame for testdata client streaming example (#31998)

packages/grafana-data/src/dataframe/DataFrameJSON.ts

@@ -83,7 +83,7 @@ export interface DataFrameSchema {
   /**
    * Field definition without any metadata
    */
-  fields?: FieldSchema[];
+  fields: FieldSchema[];
 }
 
 /**

packages/grafana-data/src/dataframe/StreamingDataFrame.test.ts

@@ -1,5 +1,5 @@
 import { FieldType } from '../types/dataFrame';
-import { DataFrameJSON, dataFrameFromJSON } from './DataFrameJSON';
+import { DataFrameJSON } from './DataFrameJSON';
 import { StreamingDataFrame } from './StreamingDataFrame';
 
 describe('Streaming JSON', () => {
@@ -22,7 +22,7 @@ describe('Streaming JSON', () => {
         },
       };
 
-      const stream = new StreamingDataFrame(dataFrameFromJSON(json));
+      const stream = new StreamingDataFrame(json);
       expect(stream.fields.map((f) => ({ name: f.name, value: f.values.buffer }))).toMatchInlineSnapshot(`
         Array [
           Object {

packages/grafana-data/src/dataframe/StreamingDataFrame.ts

@@ -28,27 +28,12 @@ export class StreamingDataFrame implements DataFrame {
   private lastUpdateTime = 0;
   private timeFieldIndex = -1;
 
-  constructor(frame: DataFrame, opts?: StreamingFrameOptions) {
-    this.name = frame.name;
-    this.refId = frame.refId;
-    this.meta = frame.meta;
+  constructor(frame: DataFrameJSON, opts?: StreamingFrameOptions) {
     this.options = {
       maxLength: 1000,
       ...opts,
     };
-
-    // Keep the existing fields
-    this.fields = frame.fields.map((f) => {
-      if (f.values instanceof ArrayVector) {
-        return f as Field<any, ArrayVector<any>>;
-      }
-      return {
-        ...f,
-        values: new ArrayVector(f.values.toArray()),
-      };
-    });
-
-    this.timeFieldIndex = this.fields.findIndex((f) => f.type === FieldType.time);
+    this.update(frame);
   }
 
   get length() {
@@ -58,14 +43,40 @@ export class StreamingDataFrame implements DataFrame {
     return this.fields[0].values.length;
   }
 
+  /**
+   * apply the new message to the existing data.  This will replace the existing schema
+   * if a new schema is included in the message, or append data matching the current schema
+   */
   update(msg: DataFrameJSON) {
-    if (msg.schema) {
-      // TODO, replace the existing fields
+    const { schema, data } = msg;
+    if (schema) {
+      if (this.fields.length > 0) {
+        // ?? keep existing data?
+      }
+
+      this.name = schema.name;
+      this.refId = schema.refId;
+      this.meta = schema.meta;
+
+      // Create new fields from the schema
+      this.fields = schema.fields.map((f) => {
+        return {
+          config: f.config ?? {},
+          name: f.name,
+          labels: f.labels,
+          type: f.type ?? FieldType.other,
+          values: new ArrayVector(),
+        };
+      });
+
+      this.timeFieldIndex = this.fields.findIndex((f) => f.type === FieldType.time);
     }
 
-    if (msg.data) {
-      const data = msg.data;
+    if (data && data.values.length && data.values[0].length) {
       const { values, entities } = data;
+      if (values.length !== this.fields.length) {
+        throw new Error('update message mismatch');
+      }
 
       if (entities) {
         entities.forEach((ents, i) => {
@@ -81,11 +92,14 @@ export class StreamingDataFrame implements DataFrame {
       });
 
       // Shorten the array less frequently than we append
-      const elapsed = Date.now() - this.lastUpdateTime;
-
+      const now = Date.now();
+      const elapsed = now - this.lastUpdateTime;
       if (elapsed > 5000) {
-        if (this.options.maxSeconds && this.timeFieldIndex >= 0) {
+        if (this.options.maxSeconds && this.timeFieldIndex >= 0 && this.length > 2) {
           // TODO -- check time length
+          const tf = this.fields[this.timeFieldIndex].values.buffer;
+          const elapsed = tf[tf.length - 1] - tf[0];
+          console.log('Check elapsed time: ', elapsed);
         }
         if (this.options.maxLength) {
           const delta = this.length - this.options.maxLength;
@@ -96,9 +110,8 @@ export class StreamingDataFrame implements DataFrame {
             });
           }
         }
+        this.lastUpdateTime = now;
       }
-
-      this.lastUpdateTime = Date.now();
     }
   }
 }

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

@@ -9,6 +9,9 @@ import {
   CSVReader,
   Field,
   LoadingState,
+  StreamingDataFrame,
+  DataFrameSchema,
+  DataFrameData,
 } from '@grafana/data';
 
 import { TestDataQuery, StreamingQuery } from './types';
@@ -45,32 +48,33 @@ export function runSignalStream(
     const streamId = `signal-${req.panelId}-${target.refId}`;
     const maxDataPoints = req.maxDataPoints || 1000;
 
-    const data = new CircularDataFrame({
-      append: 'tail',
-      capacity: maxDataPoints,
-    });
-    data.refId = target.refId;
-    data.name = target.alias || 'Signal ' + target.refId;
-    data.addField({ name: 'time', type: FieldType.time });
-    data.addField({ name: 'value', type: FieldType.number });
+    const schema: DataFrameSchema = {
+      refId: target.refId,
+      name: target.alias || 'Signal ' + target.refId,
+      fields: [
+        { name: 'time', type: FieldType.time },
+        { name: 'value', type: FieldType.number },
+      ],
+    };
 
     const { spread, speed, bands = 0, noise } = query;
-
     for (let i = 0; i < bands; i++) {
       const suffix = bands > 1 ? ` ${i + 1}` : '';
-      data.addField({ name: 'Min' + suffix, type: FieldType.number });
-      data.addField({ name: 'Max' + suffix, type: FieldType.number });
+      schema.fields.push({ name: 'Min' + suffix, type: FieldType.number });
+      schema.fields.push({ name: 'Max' + suffix, type: FieldType.number });
     }
 
+    const frame = new StreamingDataFrame({ schema }, { maxLength: maxDataPoints });
+
     let value = Math.random() * 100;
     let timeoutId: any = null;
 
     const addNextRow = (time: number) => {
       value += (Math.random() - 0.5) * spread;
 
-      let idx = 0;
-      data.fields[idx++].values.add(time);
-      data.fields[idx++].values.add(value);
+      const data: DataFrameData = {
+        values: [[time], [value]],
+      };
 
       let min = value;
       let max = value;
@@ -79,9 +83,12 @@ export function runSignalStream(
         min = min - Math.random() * noise;
         max = max + Math.random() * noise;
 
-        data.fields[idx++].values.add(min);
-        data.fields[idx++].values.add(max);
+        data.values.push([min]);
+        data.values.push([max]);
       }
+
+      const event = { data };
+      return frame.update(event);
     };
 
     // Fill the buffer on init
@@ -96,11 +103,10 @@ export function runSignalStream(
     const pushNextEvent = () => {
       addNextRow(Date.now());
       subscriber.next({
-        data: [data],
+        data: [frame],
         key: streamId,
         state: LoadingState.Streaming,
       });
-
       timeoutId = setTimeout(pushNextEvent, speed);
     };