Add Streaming capabilities to visualize INFO command (#57)

* Add Streaming query

* Update Query Editor and add Streaming dashboard

* Update README and add screenshot

Author: mikhail-vl

README.md

@@ -8,6 +8,7 @@
 [![Redis Enterprise](https://img.shields.io/badge/Redis%20Enterprise-supported-orange)](https://redislabs.com/redis-enterprise/)
 [![Go Report Card](https://goreportcard.com/badge/github.com/RedisTimeSeries/grafana-redis-datasource)](https://goreportcard.com/report/github.com/RedisTimeSeries/grafana-redis-datasource)
 [![CircleCI](https://circleci.com/gh/RedisTimeSeries/grafana-redis-datasource.svg?style=svg)](https://circleci.com/gh/RedisTimeSeries/grafana-redis-datasource)
+[![Downloads](https://img.shields.io/badge/dynamic/json?color=green&label=downloads&query=%24.downloads&url=https%3A%2F%2Fgrafana.com%2Fapi%2Fplugins%2Fredis-datasource)](https://grafana.com/grafana/plugins/redis-datasource)
 
 ## Summary
 
@@ -73,7 +74,7 @@ Project provides `docker-compose.yml` to start Redis with RedisTimeSeries module
 docker-compose up
 ```
 
-Open Grafana in your browser [http://localhost:3000](http://localhost:3000) and configure Redis Data Source. You can add as many data sources as you want to support multiple Redis databases.
+Open Grafana in your browser and configure Redis Data Source. You can add as many data sources as you want to support multiple Redis databases.
 
 ![Datasource](https://raw.githubusercontent.com/RedisTimeSeries/grafana-redis-datasource/master/src/img/datasource.png)
 
@@ -146,4 +147,4 @@ We love to hear from users, developers and the whole community interested by thi
 
 ## License
 
-- Apache License Version 2.0, see [LICENSE](LICENSE)
+- Apache License Version 2.0, see [LICENSE](https://github.com/RedisTimeSeries/grafana-redis-datasource/blob/master/LICENSE)

src/DataSource.ts

@@ -1,5 +1,16 @@
+import { head } from 'lodash';
+import { Observable } from 'rxjs';
 import { map as map$, switchMap as switchMap$ } from 'rxjs/operators';
-import { DataFrame, DataQueryRequest, DataSourceInstanceSettings, MetricFindValue, ScopedVars } from '@grafana/data';
+import {
+  CircularDataFrame,
+  DataFrame,
+  DataQueryRequest,
+  DataQueryResponse,
+  DataSourceInstanceSettings,
+  FieldType,
+  MetricFindValue,
+  ScopedVars,
+} from '@grafana/data';
 import { DataSourceWithBackend, getTemplateSrv } from '@grafana/runtime';
 import { RedisQuery } from './redis';
 import { RedisDataSourceOptions } from './types';
@@ -69,4 +80,82 @@ export class DataSource extends DataSourceWithBackend<RedisQuery, RedisDataSourc
       filter: query.filter ? templateSrv.replace(query.filter, scopedVars) : '',
     };
   }
+
+  /**
+   * Override query to support streaming
+   */
+  query(request: DataQueryRequest<RedisQuery>): Observable<DataQueryResponse> {
+    const refA = head(request.targets);
+
+    /**
+     * No streaming enabled
+     */
+    if (!refA?.streaming) {
+      return super.query(request);
+    }
+
+    /**
+     * Streaming enabled
+     */
+    return new Observable<DataQueryResponse>((subscriber) => {
+      /**
+       * This dataframe can have values constantly added, and will never exceed the given capacity
+       */
+      const frame = new CircularDataFrame({
+        append: 'tail',
+        capacity: refA?.streamingCapacity || 1000,
+      });
+
+      /**
+       * Set refId and Time field
+       */
+      frame.refId = refA.refId;
+      frame.addField({ name: 'time', type: FieldType.time });
+
+      /**
+       * Interval
+       */
+      const intervalId = setInterval(async () => {
+        let values: { [index: string]: number } = { time: Date.now() };
+
+        /**
+         * Run Query
+         */
+        const fields = await super
+          .query(request)
+          .pipe(
+            switchMap$((response) => response.data),
+            map$((data: DataFrame) => data.fields)
+          )
+          .toPromise();
+
+        if (fields) {
+          /**
+           * Add fields to frame fields and return values
+           */
+          fields.map((field) =>
+            field.values.toArray().map((value) => {
+              if (frame.fields.length < fields.length + 1) {
+                frame.addField({ name: field.name, type: field.type });
+              }
+              return (values[field.name] = value);
+            })
+          );
+        }
+
+        /**
+         * Add frame
+         */
+        frame.add(values);
+        subscriber.next({
+          data: [frame],
+          key: refA.refId,
+        });
+      }, refA.streamingInterval || 1000);
+
+      return () => {
+        clearInterval(intervalId);
+      };
+    });
+  }
 }

src/components/QueryEditor.tsx

@@ -160,6 +160,10 @@ export class QueryEditor extends PureComponent<Props> {
       type,
       section,
       fill,
+      streaming,
+      streamingInterval,
+      streamingCapacity,
+      refId,
     } = this.props.query;
     const { onRunQuery, onChange } = this.props;
 
@@ -314,6 +318,44 @@ export class QueryEditor extends PureComponent<Props> {
           </div>
         )}
 
+        {refId === 'A' && (
+          <div className="gf-form">
+            <Switch
+              label="Streaming"
+              labelClass="width-8"
+              tooltip="If checked, the datasource will stream data. Only Ref A is supported. Command should return only one line of data."
+              checked={streaming || false}
+              onChange={(event) => {
+                onChange({ ...this.props.query, streaming: event.currentTarget.checked });
+              }}
+            />
+            {streaming && (
+              <>
+                <FormField
+                  labelWidth={8}
+                  value={streamingInterval}
+                  type="number"
+                  onChange={(event: ChangeEvent<HTMLInputElement>) => {
+                    onChange({ ...this.props.query, streamingInterval: Number(event.target.value) });
+                  }}
+                  label="Interval"
+                  tooltip="Streaming interval in milliseconds. Default is 1000ms."
+                />
+                <FormField
+                  labelWidth={8}
+                  value={streamingCapacity}
+                  type="number"
+                  onChange={(event: ChangeEvent<HTMLInputElement>) => {
+                    onChange({ ...this.props.query, streamingCapacity: Number(event.target.value) });
+                  }}
+                  label="Capacity"
+                  tooltip="Values will be constantly added and will never exceed the given capacity. Default is 1000."
+                />
+              </>
+            )}
+          </div>
+        )}
+
         <Button onClick={onRunQuery}>Run</Button>
       </div>
     );