Postgres: Support request cancellation properly (Uses new backendSrv.fetch…

Postgres: Support request cancellation properly (Uses new backendSrv.fetch Observable request API) (#27478)

* Postgres: Replaces dataSourceRequest with fetch

* Postgres: Replaces dataSourceRequest with fetch

* Tests: removes unnecessary import

public/app/plugins/datasource/postgres/datasource.ts

 import _ from 'lodash';
 import _ from 'lodash';
+import { Observable, of } from 'rxjs';
+import { map } from 'rxjs/operators';
+import { getBackendSrv } from '@grafana/runtime';
+import { DataQueryResponse, ScopedVars } from '@grafana/data';
+
 import ResponseParser from './response_parser';
 import ResponseParser from './response_parser';
 import PostgresQuery from 'app/plugins/datasource/postgres/postgres_query';
 import PostgresQuery from 'app/plugins/datasource/postgres/postgres_query';
-import { getBackendSrv } from '@grafana/runtime';
-import { ScopedVars } from '@grafana/data';
 import { TemplateSrv } from 'app/features/templating/template_srv';
 import { TemplateSrv } from 'app/features/templating/template_srv';
 import { TimeSrv } from 'app/features/dashboard/services/TimeSrv';
 import { TimeSrv } from 'app/features/dashboard/services/TimeSrv';
 //Types
 //Types
-import { PostgresQueryForInterpolation } from './types';
+import { PostgresMetricFindValue, PostgresQueryForInterpolation } from './types';
 import { getSearchFilterScopedVar } from '../../../features/variables/utils';
 import { getSearchFilterScopedVar } from '../../../features/variables/utils';
 
 
 export class PostgresDatasource {
 export class PostgresDatasource {
@@ -31,7 +34,7 @@ export class PostgresDatasource {
     this.interval = (instanceSettings.jsonData || {}).timeInterval || '1m';
     this.interval = (instanceSettings.jsonData || {}).timeInterval || '1m';
   }
   }
 
 
-  interpolateVariable = (value: string, variable: { multi: any; includeAll: any }) => {
+  interpolateVariable = (value: string | string[], variable: { multi: any; includeAll: any }) => {
     if (typeof value === 'string') {
     if (typeof value === 'string') {
       if (variable.multi || variable.includeAll) {
       if (variable.multi || variable.includeAll) {
         return this.queryModel.quoteLiteral(value);
         return this.queryModel.quoteLiteral(value);
@@ -69,7 +72,7 @@ export class PostgresDatasource {
     return expandedQueries;
     return expandedQueries;
   }
   }
 
 
-  query(options: any) {
+  query(options: any): Observable<DataQueryResponse> {
     const queries = _.filter(options.targets, target => {
     const queries = _.filter(options.targets, target => {
       return target.hide !== true;
       return target.hide !== true;
     }).map(target => {
     }).map(target => {
@@ -86,11 +89,11 @@ export class PostgresDatasource {
     });
     });
 
 
     if (queries.length === 0) {
     if (queries.length === 0) {
-      return Promise.resolve({ data: [] });
+      return of({ data: [] });
     }
     }
 
 
     return getBackendSrv()
     return getBackendSrv()
-      .datasourceRequest({
+      .fetch({
         url: '/api/tsdb/query',
         url: '/api/tsdb/query',
         method: 'POST',
         method: 'POST',
         data: {
         data: {
@@ -99,7 +102,7 @@ export class PostgresDatasource {
           queries: queries,
           queries: queries,
         },
         },
       })
       })
-      .then(this.responseParser.processQueryResult);
+      .pipe(map(this.responseParser.processQueryResult));
   }
   }
 
 
   annotationQuery(options: any) {
   annotationQuery(options: any) {
@@ -117,7 +120,7 @@ export class PostgresDatasource {
     };
     };
 
 
     return getBackendSrv()
     return getBackendSrv()
-      .datasourceRequest({
+      .fetch({
         url: '/api/tsdb/query',
         url: '/api/tsdb/query',
         method: 'POST',
         method: 'POST',
         data: {
         data: {
@@ -126,10 +129,14 @@ export class PostgresDatasource {
           queries: [query],
           queries: [query],
         },
         },
       })
       })
-      .then((data: any) => this.responseParser.transformAnnotationResponse(options, data));
+      .pipe(map((data: any) => this.responseParser.transformAnnotationResponse(options, data)))
+      .toPromise();
   }
   }
 
 
-  metricFindQuery(query: string, optionalOptions: { variable?: any; searchFilter?: string }) {
+  metricFindQuery(
+    query: string,
+    optionalOptions: { variable?: any; searchFilter?: string }
+  ): Promise<PostgresMetricFindValue[]> {
     let refId = 'tempvar';
     let refId = 'tempvar';
     if (optionalOptions && optionalOptions.variable && optionalOptions.variable.name) {
     if (optionalOptions && optionalOptions.variable && optionalOptions.variable.name) {
       refId = optionalOptions.variable.name;
       refId = optionalOptions.variable.name;
@@ -156,12 +163,13 @@ export class PostgresDatasource {
     };
     };
 
 
     return getBackendSrv()
     return getBackendSrv()
-      .datasourceRequest({
+      .fetch({
         url: '/api/tsdb/query',
         url: '/api/tsdb/query',
         method: 'POST',
         method: 'POST',
         data: data,
         data: data,
       })
       })
-      .then((data: any) => this.responseParser.parseMetricFindQueryResult(refId, data));
+      .pipe(map((data: any) => this.responseParser.parseMetricFindQueryResult(refId, data)))
+      .toPromise();
   }
   }
 
 
   getVersion() {
   getVersion() {

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

+import { MetricFindValue } from '@grafana/data';
+
 export interface PostgresQueryForInterpolation {
 export interface PostgresQueryForInterpolation {
   alias?: any;
   alias?: any;
   format?: any;
   format?: any;
@@ -5,3 +7,7 @@ export interface PostgresQueryForInterpolation {
   refId?: any;
   refId?: any;
   hide?: any;
   hide?: any;
 }
 }
+
+export interface PostgresMetricFindValue extends MetricFindValue {
+  value?: string;
+}

public/test/helpers/observableTester.ts

@@ -9,6 +9,11 @@ interface SubscribeAndExpectOnNext<T> extends ObservableTester<T> {
   expect: (value: T) => void;
   expect: (value: T) => void;
 }
 }
 
 
+interface SubscribeAndExpectOnNextAndComplete<T> extends ObservableTester<T> {
+  expectOnNext: (value: T) => void;
+  expectOnComplete: () => void;
+}
+
 interface SubscribeAndExpectOnComplete<T> extends ObservableTester<T> {
 interface SubscribeAndExpectOnComplete<T> extends ObservableTester<T> {
   expect: () => void;
   expect: () => void;
 }
 }
@@ -47,6 +52,33 @@ export const observableTester = () => {
     });
     });
   };
   };
 
 
+  const subscribeAndExpectOnNextAndComplete = <T>({
+    observable,
+    expectOnComplete,
+    expectOnNext,
+    done,
+  }: SubscribeAndExpectOnNextAndComplete<T>): void => {
+    observable.subscribe({
+      next: (value: T) => {
+        try {
+          expectOnNext(value);
+          done();
+        } catch (err) {
+          done.fail(err);
+        }
+      },
+      error: err => done.fail(err),
+      complete: () => {
+        try {
+          expectOnComplete();
+          done();
+        } catch (err) {
+          done.fail(err);
+        }
+      },
+    });
+  };
+
   const subscribeAndExpectOnError = <T>({ observable, expect, done }: SubscribeAndExpectOnError<T>): void => {
   const subscribeAndExpectOnError = <T>({ observable, expect, done }: SubscribeAndExpectOnError<T>): void => {
     observable.subscribe({
     observable.subscribe({
       next: () => {},
       next: () => {},
@@ -64,5 +96,10 @@ export const observableTester = () => {
     });
     });
   };
   };
 
 
-  return { subscribeAndExpectOnNext, subscribeAndExpectOnComplete, subscribeAndExpectOnError };
+  return {
+    subscribeAndExpectOnNext,
+    subscribeAndExpectOnComplete,
+    subscribeAndExpectOnNextAndComplete,
+    subscribeAndExpectOnError,
+  };
 };
 };