support non live query

src/StreamingListener.ts

@@ -1,5 +1,5 @@
 import { TimeSeriesMessage } from 'types';
-import { KeyValue, CircularDataFrame, FieldType, LoadingState } from '@grafana/data';
+import { KeyValue, CircularDataFrame, FieldType, LoadingState, DataFrame } from '@grafana/data';
 import { Subject, Observable, ReplaySubject } from 'rxjs';
 import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
 import { DataQueryResponse } from '@grafana/ui';
@@ -17,7 +17,8 @@ export class StreamListener {
   stream?: WebSocketSubject<any>;
 
   constructor(private capacity: number, url?: string) {
-    if (url) {
+    if (url && url.length > 4 && url.startsWith('ws')) {
+      console.log('Connect to:', url);
       this.stream = webSocket({
         url,
         openObserver: {
@@ -65,6 +66,14 @@ export class StreamListener {
     setTimeout(this.dummyValues, 100 + Math.random() * 800); // ~1/sec
   };
 
+  getAllFrames(): DataFrame[] {
+    const all: DataFrame[] = [];
+    for (const v of Object.values(this.byName)) {
+      all.push(v.frame);
+    }
+    return all;
+  }
+
   getAllObservers(): Array<Observable<DataQueryResponse>> {
     const all: Array<Observable<DataQueryResponse>> = [];
     for (const v of Object.values(this.byName)) {
@@ -111,9 +120,8 @@ export class StreamListener {
     df.validate();
     info.subject.next({
       key: info.key,
-      state: LoadingState.Streaming, // ???
+      state: LoadingState.Streaming,
       data: [df],
     });
-    console.log('PROCESS', msg);
   }
 }

src/datasource.ts

@@ -1,8 +1,10 @@
 import { DataQueryRequest, DataQueryResponse, DataSourceApi, DataSourceInstanceSettings } from '@grafana/ui';
 
 import { StreamingQuery, MyDataSourceOptions } from './types';
-import { Observable, of, merge } from 'rxjs';
+import { Observable, of, merge, interval } from 'rxjs';
+import { map } from 'rxjs/operators';
 import { StreamListener } from 'StreamingListener';
+import { LoadingState, DataFrame } from '@grafana/data';
 
 export class DataSource extends DataSourceApi<StreamingQuery, MyDataSourceOptions> {
   listener?: StreamListener;
@@ -13,23 +15,54 @@ export class DataSource extends DataSourceApi<StreamingQuery, MyDataSourceOption
   }
 
   query(options: DataQueryRequest<StreamingQuery>): Observable<DataQueryResponse> {
-    if (!this.listener) {
+    const { listener } = this;
+    if (!listener) {
       throw new Error('missing listener');
     }
 
+    const isLive = options.rangeRaw && options.rangeRaw!.to === 'now';
+    if (!isLive) {
+      let hasStar = false;
+      let data: DataFrame[] = [];
+      options.targets.forEach(t => {
+        if (!t.name || t.name === '*') {
+          hasStar = true;
+        } else {
+          data.push(listener.getOrCreate(t.name).frame);
+        }
+      });
+      if (hasStar) {
+        data = listener.getAllFrames();
+      }
+      return of({ data });
+    }
+
     let hasStar = false;
     let subs: Array<Observable<DataQueryResponse>> = [];
     options.targets.forEach(t => {
       if (!t.name || t.name === '*') {
         hasStar = true;
       } else {
-        subs.push(this.listener!.listen(t.name));
+        subs.push(listener.listen(t.name));
       }
     });
 
     if (hasStar) {
-      subs = this.listener.getAllObservers();
+      subs = listener.getAllObservers();
     }
+
+    // Update every 1/2 sec regardless of results
+    const ping = interval(1000).pipe(
+      map(v => {
+        return {
+          key: 'heartbeat',
+          state: LoadingState.Streaming,
+          data: [],
+        } as DataQueryResponse;
+      })
+    );
+    subs.push(ping);
+
     if (subs.length === 0) {
       return of({ data: [] }); // nothing
     }
@@ -41,6 +74,7 @@ export class DataSource extends DataSourceApi<StreamingQuery, MyDataSourceOption
 
   testDatasource() {
     // Implement a health check for your data source.
+
     return new Promise((resolve, reject) => {
       resolve({
         status: 'success',