diff --git a/.betterer.results b/.betterer.results
index 039459ba420..b6152fde655 100644
--- a/.betterer.results
+++ b/.betterer.results
@@ -1328,8 +1328,7 @@ exports[`better eslint`] = {
[0, 0, 0, "Do not use any type assertions.", "2"],
[0, 0, 0, "Do not use any type assertions.", "3"],
[0, 0, 0, "Do not use any type assertions.", "4"],
- [0, 0, 0, "Do not use any type assertions.", "5"],
- [0, 0, 0, "Unexpected any. Specify a different type.", "6"]
+ [0, 0, 0, "Unexpected any. Specify a different type.", "5"]
],
"public/app/core/utils/object.ts:5381": [
[0, 0, 0, "Do not use any type assertions.", "0"],
@@ -3070,9 +3069,6 @@ exports[`better eslint`] = {
[0, 0, 0, "No untranslated strings. Wrap text with ", "6"],
[0, 0, 0, "No untranslated strings. Wrap text with ", "7"]
],
- "public/app/features/apiserver/client.ts:5381": [
- [0, 0, 0, "Do not use any type assertions.", "0"]
- ],
"public/app/features/auth-config/AuthDrawer.tsx:5381": [
[0, 0, 0, "No untranslated strings in text props. Wrap text with or use t()", "0"],
[0, 0, 0, "No untranslated strings in text props. Wrap text with or use t()", "1"],
diff --git a/packages/grafana-runtime/src/services/backendSrv.ts b/packages/grafana-runtime/src/services/backendSrv.ts
index 6d636c8a60b..14352c6aabb 100644
--- a/packages/grafana-runtime/src/services/backendSrv.ts
+++ b/packages/grafana-runtime/src/services/backendSrv.ts
@@ -68,7 +68,7 @@ export type BackendSrvRequest = {
*
* By default values are json parsed from text
*/
- responseType?: 'json' | 'text' | 'arraybuffer' | 'blob' | 'stream';
+ responseType?: 'json' | 'text' | 'arraybuffer' | 'blob';
/**
* Used to cancel an open connection
diff --git a/public/app/core/services/backend_srv.ts b/public/app/core/services/backend_srv.ts
index ce7dab8c838..cc667ac6c56 100644
--- a/public/app/core/services/backend_srv.ts
+++ b/public/app/core/services/backend_srv.ts
@@ -198,13 +198,14 @@ export class BackendSrv implements BackendService {
}
}
process()
- .catch((e) => {
- console.log(requestId, 'catch', e);
- }) // from abort
.then(() => {
console.log(requestId, 'complete');
observer.complete();
- }); // runs in background
+ }) // runs in background
+ .catch((e) => {
+ console.log(requestId, 'catch', e);
+ observer.error(e);
+ }); // from abort
},
error: (e) => {
observer.error(e);
diff --git a/public/app/core/utils/fetch.ts b/public/app/core/utils/fetch.ts
index fb2564146d8..237c5d0eb13 100644
--- a/public/app/core/utils/fetch.ts
+++ b/public/app/core/utils/fetch.ts
@@ -115,7 +115,7 @@ export const parseBody = (options: BackendSrvRequest, isAppJson: boolean) => {
export async function parseResponseBody(
response: Response,
- responseType?: 'json' | 'text' | 'arraybuffer' | 'blob' | 'stream'
+ responseType?: 'json' | 'text' | 'arraybuffer' | 'blob'
): Promise {
if (responseType) {
switch (responseType) {
@@ -138,9 +138,6 @@ export async function parseResponseBody(
}
return await response.json();
- case 'stream':
- return Promise.resolve(response.body) as Promise;
-
case 'text':
// this specifically returns a Promise
// TODO refactor this function to remove the type assertions
diff --git a/public/app/features/apiserver/client.ts b/public/app/features/apiserver/client.ts
index 2a2edb65600..c38c3bf4476 100644
--- a/public/app/features/apiserver/client.ts
+++ b/public/app/features/apiserver/client.ts
@@ -1,4 +1,4 @@
-import { Observable, Subscriber, map, switchMap, from, filter, retry, catchError } from 'rxjs';
+import { Observable, switchMap, from, retry, catchError } from 'rxjs';
import { config, getBackendSrv } from '@grafana/runtime';
import { contextSrv } from 'app/core/core';
@@ -38,104 +38,40 @@ export class ScopedResourceClient implements
}
public watch(name?: string, resourceVersion?: string): Observable> {
- return this.createWatchRequest(name, resourceVersion).pipe(
- switchMap((response) => this.handleWatchStream(response)),
- retry({ count: 3, delay: 1000 }),
- catchError((error) => {
- console.error('Watch stream error:', error);
- throw error;
+ return getBackendSrv()
+ .chunked({
+ url: name ? `${this.url}/${name}` : this.url,
+ params: {
+ watch: true,
+ resourceVersion,
+ },
})
- );
- }
-
- private createWatchRequest(name?: string, resourceVersion?: string) {
- return getBackendSrv().fetch>({
- url: name ? `${this.url}/${name}` : this.url,
- params: {
- watch: true,
- resourceVersion,
- },
- responseType: 'stream',
- });
- }
-
- private handleWatchStream(result: { data: ReadableStream }): Observable> {
- const decoder = new TextDecoder();
- let buffer = '';
-
- return fromReadableStream(result.data).pipe(
- map((chunk) => decoder.decode(chunk, { stream: true })),
- map((text) => {
- buffer += text;
- const events: Array> = [];
-
- // Find complete JSON objects in the buffer
- let startIndex = buffer.indexOf('{');
- while (startIndex !== -1) {
- try {
- const endIndex = this.findJsonEnd(buffer, startIndex);
- if (endIndex === -1) {
- break;
- }
-
- const jsonStr = buffer.substring(startIndex, endIndex + 1);
- const event = JSON.parse(jsonStr) as ResourceEvent;
- events.push(event);
-
- // Remove processed data from buffer
- buffer = buffer.substring(endIndex + 1);
- startIndex = buffer.indexOf('{');
- } catch (e) {
- // If JSON.parse fails, move to next potential object
- buffer = buffer.substring(startIndex + 1);
- startIndex = buffer.indexOf('{');
+ .pipe(
+ switchMap((response) => {
+ const events: Array> = [];
+ if (response.ok && response.data) {
+ const decoder = new TextDecoder();
+ decoder
+ .decode(response.data)
+ .split('\n')
+ .forEach((v) => {
+ if (v?.length) {
+ try {
+ events.push(JSON.parse(v));
+ } catch (e) {
+ console.warn('invalid response', e);
+ }
+ }
+ });
}
- }
-
- return events;
- }),
- filter((events) => events.length > 0),
- switchMap((events) => from(events))
- );
- }
-
- private findJsonEnd(str: string, startIndex: number): number {
- let brackets = 0;
- let inString = false;
- let escapeNext = false;
-
- for (let i = startIndex; i < str.length; i++) {
- const char = str[i];
-
- if (escapeNext) {
- escapeNext = false;
- continue;
- }
-
- if (char === '\\' && inString) {
- escapeNext = true;
- continue;
- }
-
- if (char === '"' && !escapeNext) {
- inString = !inString;
- continue;
- }
-
- if (!inString) {
- if (char === '{') {
- brackets++;
- }
- if (char === '}') {
- brackets--;
- if (brackets === 0) {
- return i;
- }
- }
- }
- }
-
- return -1;
+ return from(events);
+ }),
+ retry({ count: 3, delay: 1000 }),
+ catchError((error) => {
+ console.error('Watch stream error:', error);
+ throw error;
+ })
+ );
}
public async subresource(name: string, path: string): Promise {
@@ -242,48 +178,3 @@ export const parseListOptionsSelector = (selector: ListOptionsLabelSelector | Li
.filter(Boolean)
.join(',');
};
-
-/**
- * Creates an Observable source from a ReadableStream source that will emit any
- * values emitted by the stream.
- *
- * https://github.com/rxjs-ninja/rxjs-ninja/blob/main/libs/rxjs/utility/src/lib/from-readable-stream.ts
- */
-function fromReadableStream(
- stream: ReadableStream,
- signal?: AbortSignal,
- queueStrategy?: QueuingStrategy,
- throwEndAsError = false
-): Observable {
- function createStream(subscriber: Subscriber) {
- return new WritableStream(
- {
- write: (value) => subscriber.next(value),
- abort: (error) => {
- if (throwEndAsError) {
- subscriber.error(error);
- } else if (!subscriber.closed) {
- subscriber.complete();
- }
- },
- close: () => {
- if (!subscriber.closed) {
- subscriber.complete();
- }
- },
- },
- queueStrategy
- );
- }
-
- return new Observable((subscriber) => {
- stream
- .pipeTo(createStream(subscriber), { signal })
- .then(() => {
- return !subscriber.closed && subscriber.complete();
- })
- .catch((error) => subscriber.error(error));
-
- return () => !stream.locked && stream.cancel();
- });
-}