diff --git a/packages/grafana-runtime/src/services/backendSrv.ts b/packages/grafana-runtime/src/services/backendSrv.ts index 35ba108c0f5..2585cdd1395 100644 --- a/packages/grafana-runtime/src/services/backendSrv.ts +++ b/packages/grafana-runtime/src/services/backendSrv.ts @@ -23,7 +23,7 @@ export type BackendSrvRequest = { }; export interface BackendSrv { - get(url: string, params?: any): Promise; + get(url: string, params?: any, requestId?: string): Promise; delete(url: string): Promise; diff --git a/public/app/core/services/backend_srv.ts b/public/app/core/services/backend_srv.ts index 73f6e24f028..1a181f03d1b 100644 --- a/public/app/core/services/backend_srv.ts +++ b/public/app/core/services/backend_srv.ts @@ -1,5 +1,5 @@ import omitBy from 'lodash/omitBy'; -import { from, merge, MonoTypeOperatorFunction, Observable, Subject, throwError } from 'rxjs'; +import { from, merge, MonoTypeOperatorFunction, Observable, of, Subject, throwError } from 'rxjs'; import { catchError, filter, map, mergeMap, retryWhen, share, takeUntil, tap, throwIfEmpty } from 'rxjs/operators'; import { fromFetch } from 'rxjs/fetch'; import { BackendSrv as BackendService, BackendSrvRequest } from '@grafana/runtime'; @@ -49,6 +49,11 @@ interface ErrorResponse { cancelled?: boolean; } +enum CancellationType { + request, + dataSourceRequest, +} + function serializeParams(data: Record): string { return Object.keys(data) .map(key => { @@ -90,8 +95,8 @@ export class BackendSrv implements BackendService { } } - async get(url: string, params?: any) { - return await this.request({ method: 'GET', url, params }); + async get(url: string, params?: any, requestId?: string) { + return await this.request({ method: 'GET', url, params, requestId }); } async delete(url: string) { @@ -150,6 +155,13 @@ export class BackendSrv implements BackendService { }; async request(options: BackendSrvRequest): Promise { + // A requestId is a unique identifier for a particular query. + // Every observable below has a takeUntil that subscribes to this.inFlightRequests and + // will cancel/unsubscribe that observable when a new datasourceRequest with the same requestId is made + if (options.requestId) { + this.inFlightRequests.next(options.requestId); + } + options = this.parseRequestOptions(options, this.dependencies.contextSrv.user?.orgId); const fromFetchStream = this.getFromFetchStream(options); @@ -173,7 +185,8 @@ export class BackendSrv implements BackendService { // this setTimeout hack enables any caller catching this err to set isHandled to true setTimeout(() => this.requestErrorHandler(err), 50); return throwError(err); - }) + }), + this.handleStreamCancellation(options, CancellationType.request) ) .toPromise(); } @@ -233,28 +246,7 @@ export class BackendSrv implements BackendService { return throwError(err); }), - takeUntil( - this.inFlightRequests.pipe( - filter(requestId => { - let cancelRequest = false; - if (options && options.requestId && options.requestId === requestId) { - // when a new requestId is started it will be published to inFlightRequests - // if a previous long running request that hasn't finished yet has the same requestId - // we need to cancel that request - cancelRequest = true; - } - return cancelRequest; - }) - ) - ), - // when a request is cancelled by takeUntil it will complete without emitting anything - // throwIfEmpty will then throw an error with cancelled set to true - throwIfEmpty(() => ({ - cancelled: true, - status: this.HTTP_REQUEST_CANCELED, - statusText: 'Request was aborted', - request: { url: parseUrlFromOptions(options), ...parseInitFromOptions(options) }, - })) + this.handleStreamCancellation(options, CancellationType.dataSourceRequest) ) .toPromise(); } @@ -540,6 +532,48 @@ export class BackendSrv implements BackendService { ) ) ); + + private handleStreamCancellation = ( + options: BackendSrvRequest, + resultType: CancellationType + ): MonoTypeOperatorFunction => inputStream => + inputStream.pipe( + takeUntil( + this.inFlightRequests.pipe( + filter(requestId => { + let cancelRequest = false; + if (options && options.requestId && options.requestId === requestId) { + // when a new requestId is started it will be published to inFlightRequests + // if a previous long running request that hasn't finished yet has the same requestId + // we need to cancel that request + cancelRequest = true; + } + return cancelRequest; + }) + ) + ), + // when a request is cancelled by takeUntil it will complete without emitting anything so we use throwIfEmpty to identify this case + // in throwIfEmpty we'll then throw an cancelled error and then we'll return the correct result in the catchError or rethrow + throwIfEmpty(() => ({ + cancelled: true, + })), + catchError(err => { + if (!err.cancelled) { + throwError(err); + } + + if (resultType === CancellationType.dataSourceRequest) { + return of({ + data: [], + status: this.HTTP_REQUEST_CANCELED, + statusText: 'Request was aborted', + request: { url: parseUrlFromOptions(options), ...parseInitFromOptions(options) }, + }); + } + + return of([]); + }) + ); } coreModule.factory('backendSrv', () => backendSrv); diff --git a/public/app/core/specs/backend_srv.test.ts b/public/app/core/specs/backend_srv.test.ts index 508556ffd17..1f1f694a4ba 100644 --- a/public/app/core/specs/backend_srv.test.ts +++ b/public/app/core/specs/backend_srv.test.ts @@ -161,6 +161,57 @@ describe('backendSrv', () => { }); }); + describe('when called with the same requestId twice', () => { + it('then it should cancel the first call and the first call should be unsubscribed', async () => { + const url = '/api/dashboard/'; + const { backendSrv, fromFetchMock } = getTestContext({ url }); + const unsubscribe = jest.fn(); + const slowData = { message: 'Slow Request' }; + const slowFetch = new Observable(subscriber => { + subscriber.next({ + ok: true, + status: 200, + statusText: 'Ok', + text: () => Promise.resolve(JSON.stringify(slowData)), + headers: { + 'Content-Type': 'application/json', + }, + redirected: false, + type: 'basic', + url, + }); + return unsubscribe; + }).pipe(delay(10000)); + const fastData = { message: 'Fast Request' }; + const fastFetch = of({ + ok: true, + status: 200, + statusText: 'Ok', + text: () => Promise.resolve(JSON.stringify(fastData)), + headers: { + 'Content-Type': 'application/json', + }, + redirected: false, + type: 'basic', + url, + }); + fromFetchMock.mockImplementationOnce(() => slowFetch); + fromFetchMock.mockImplementation(() => fastFetch); + const options = { + url, + method: 'GET', + requestId: 'A', + }; + const slowRequest = backendSrv.request(options); + const fastResponse = await backendSrv.request(options); + expect(fastResponse).toEqual({ message: 'Fast Request' }); + + const result = await slowRequest; + expect(result).toEqual([]); + expect(unsubscribe).toHaveBeenCalledTimes(1); + }); + }); + describe('when making an unsuccessful call and conditions for retry are favorable and loginPing does not throw', () => { it('then it should retry', async () => { jest.useFakeTimers(); @@ -404,9 +455,9 @@ describe('backendSrv', () => { status: 200, statusText: 'Ok', type: 'basic', - url, + url: '/api/dashboard/', request: { - url, + url: '/api/dashboard/', method: 'GET', body: undefined, headers: { @@ -416,21 +467,20 @@ describe('backendSrv', () => { }, }); - await slowRequest.catch(error => { - expect(error).toEqual({ - cancelled: true, - status: -1, - statusText: 'Request was aborted', - request: { - url, - method: 'GET', - body: undefined, - headers: { - 'Content-Type': 'application/json', - Accept: 'application/json, text/plain, */*', - }, + const result = await slowRequest; + expect(result).toEqual({ + data: [], + status: -1, + statusText: 'Request was aborted', + request: { + url: '/api/dashboard/', + method: 'GET', + body: undefined, + headers: { + 'Content-Type': 'application/json', + Accept: 'application/json, text/plain, */*', }, - }); + }, }); expect(unsubscribe).toHaveBeenCalledTimes(1); });