feat: Decoder per implemtation

Author: asafgardin

src/APIClient.ts

@@ -85,7 +85,7 @@ export abstract class APIClient {
     };
 
     return this.performRequest(options as FinalRequestOptions).then(
-      (response) => handleAPIResponse<Rsp>(response) as Rsp,
+      (response) => this.fetch.handleResponse<Rsp>(response) as Rsp,
     );
   }
 

src/Streaming/SSEDecoder.ts

@@ -1,13 +1,13 @@
 import { SSE_DATA_PREFIX } from './Consts';
 import { StreamingDecodeError } from '../errors';
-import { getReadableStream } from 'envFetch';
+import { UnifiedResponse } from 'types';
 
 export interface SSEDecoder {
   decode(line: string): string | null;
-  iterLines(response: Response): AsyncIterableIterator<string>;
+  iterLines(response: UnifiedResponse): AsyncIterableIterator<string>;
 }
 
-export class DefaultSSEDecoder implements SSEDecoder {
+abstract class BaseSSEDecoder implements SSEDecoder {
   decode(line: string): string | null {
     if (!line) return null;
 
@@ -18,38 +18,54 @@ export class DefaultSSEDecoder implements SSEDecoder {
     throw new StreamingDecodeError(`Invalid SSE line: ${line}`);
   }
 
-  async *iterLines(response: Response): AsyncIterableIterator<string> {
-    if (!response.body) {
-      throw new Error('Response body is null');
-    }
-
-    const reader = await getReadableStream(response.body);
-    let buffer = '';
-
-
-    try {
-      while (true) {
-        const { done, value } = await reader.read();
+  abstract iterLines(response: UnifiedResponse): AsyncIterableIterator<string>;
 
-        if (done) {
-          if (buffer.length > 0) {
-            const decoded = this.decode(buffer.trim());
+  async *_iterLines(reader: ReadableStreamDefaultReader<Uint8Array>): AsyncIterableIterator<string> {
+    
+      let buffer = '';
+  
+      try {
+        while (true) {
+          const { done, value } = await reader.read();
+  
+          if (done) {
+            if (buffer.length > 0) {
+              const decoded = this.decode(buffer.trim());
+              if (decoded) yield decoded;
+            }
+            break;
+          }
+  
+          buffer += new TextDecoder().decode(value);
+          const lines = buffer.split('\n');
+          buffer = lines.pop() || '';
+  
+          for (const line of lines) {
+            const decoded = this.decode(line.trim());
             if (decoded) yield decoded;
           }
-          break;
-        }
-
-        buffer += new TextDecoder().decode(value);
-        const lines = buffer.split('\n');
-        buffer = lines.pop() || '';
-
-        for (const line of lines) {
-          const decoded = this.decode(line.trim());
-          if (decoded) yield decoded;
         }
+      } finally {
+        reader.releaseLock();
       }
-    } finally {
-      reader.releaseLock();
+  }
+}
+
+export class BrowserSSEDecoder extends BaseSSEDecoder {
+  async *iterLines(response: UnifiedResponse): AsyncIterableIterator<string> {
+    if (!response.body) {
+        throw new Error('Response body is null');
     }
+
+    const body = response.body as ReadableStream<Uint8Array>;
+    yield* this._iterLines(body.getReader());
   }
 }
+
+export class NodeSSEDecoder extends BaseSSEDecoder {
+    async *iterLines(response: UnifiedResponse): AsyncIterableIterator<string> {
+        const readerStream = (await import("stream/web")).ReadableStream as any;
+        const reader = readerStream.from(response.body).getReader();
+        yield* this._iterLines(reader);
+    }
+}

src/Streaming/Stream.ts

@@ -1,4 +1,4 @@
-import { DefaultSSEDecoder } from './SSEDecoder';
+import { BrowserSSEDecoder } from './SSEDecoder';
 import { SSEDecoder } from './SSEDecoder';
 import { SSE_DONE_MSG } from './Consts';
 import { StreamingDecodeError } from '../errors';
@@ -20,7 +20,7 @@ export class Stream<T> implements AsyncIterableIterator<T> {
     private response: UnifiedResponse,
     decoder?: SSEDecoder,
   ) {
-    this.decoder = decoder || new DefaultSSEDecoder();
+    this.decoder = decoder || new BrowserSSEDecoder();
     this.iterator = this.stream();
   }
 

src/envFetch.ts

@@ -3,7 +3,6 @@ import { BrowserFetch, Fetch, NodeFetch } from 'fetch';
 export const isBrowser = typeof window !== "undefined" && typeof window.document !== "undefined";
 export const isWebWorker =
     typeof self === "object" &&
-    // @ts-ignore
     typeof self?.importScripts === "function" &&
     (self.constructor?.name === "DedicatedWorkerGlobalScope" ||
         self.constructor?.name === "ServiceWorkerGlobalScope" ||
@@ -12,16 +11,6 @@ export const isWebWorker =
 
 export const isNode = typeof process !== "undefined" && Boolean(process.version) && Boolean(process.versions?.node);
 
-export const getReadableStream = async (responseBody: ReadableStream<Uint8Array>): Promise<ReadableStreamDefaultReader<Uint8Array>> => {
-    if (isBrowser || isWebWorker) {
-        return responseBody.getReader();
-    } else {
-        // @ts-ignore - ReadableStream.from() is available in Node.js but TypeScript doesn't recognize it
-        return (await import("stream/web")).ReadableStream.from(responseBody).getReader();
-    }
-};
-
-
 export function createFetchInstance(): Fetch {
     if (isBrowser || isWebWorker) {
         return new BrowserFetch();

src/fetch/BaseFetch.ts

@@ -1,6 +1,29 @@
+import { AI21Error } from "errors";
+import { Stream } from "Streaming";
 import { FinalRequestOptions, UnifiedResponse } from "types";
+import { APIResponseProps, UnifiedReadableStream } from "types/API";
 
 
+export type APIResponse<T> = {
+    data?: T;
+    response: UnifiedResponse;
+  };
 export abstract class Fetch {
     abstract call(url: string, options: FinalRequestOptions): Promise<UnifiedResponse>;
+    async handleResponse<T>({
+        response,
+        options
+    }: APIResponseProps) {
+        if (options.stream) {
+            if (!response.body) {
+              throw new AI21Error('Response body is null');
+            }
+            
+            return this.handleStream<T>(response);
+        }
+
+        const contentType = response.headers.get('content-type');
+        return contentType?.includes('application/json') ? await response.json() : null;
+    }
+    abstract handleStream<T>(response: UnifiedResponse): Stream<T>;
 }

src/fetch/BrowserFetch.ts

@@ -1,5 +1,7 @@
 import { FinalRequestOptions, UnifiedResponse } from 'types';
 import { Fetch } from './BaseFetch';
+import { Stream } from 'Streaming';
+import { BrowserSSEDecoder } from 'Streaming/SSEDecoder';
 
 export class BrowserFetch extends Fetch {
     call(url: string, options: FinalRequestOptions): Promise<UnifiedResponse> {
@@ -12,4 +14,8 @@ export class BrowserFetch extends Fetch {
             signal: controller.signal,
         });
     }
+
+    handleStream<T>(response: UnifiedResponse): Stream<T> {
+        return new Stream<T>(response as Response, new BrowserSSEDecoder());
+    }
 }

src/fetch/NodeFetch.ts

@@ -1,5 +1,7 @@
 import { FinalRequestOptions, UnifiedResponse } from "types";
 import { Fetch } from "./BaseFetch";
+import { NodeSSEDecoder } from "Streaming/SSEDecoder";
+import { Stream } from "Streaming";
 
 export class NodeFetch extends Fetch {
     async call(url: string, options: FinalRequestOptions): Promise<UnifiedResponse> {
@@ -12,4 +14,9 @@ export class NodeFetch extends Fetch {
             body: options?.body ? JSON.stringify(options.body) : undefined,
         });
     }
-}
+
+    handleStream<T>(response: UnifiedResponse): Stream<T> {
+        type NodeRespose = import('node-fetch').Response
+        return new Stream<T>(response as NodeRespose, new NodeSSEDecoder());
+    }
+}

src/types/API.ts

@@ -27,4 +27,5 @@ export type FinalRequestOptions = RequestOptions & {
 
 export type DefaultQuery = Record<string, unknown>;
 export type Headers = Record<string, string | null | undefined>;
-export type UnifiedResponse = Response | import('node-fetch').Response;
+export type UnifiedResponse = Response | import('node-fetch').Response;
+export type UnifiedReadableStream = ReadableStream<Uint8Array> | import('stream/web').ReadableStream;