Implemented retry stream

README.md

@@ -10,16 +10,16 @@ Warden is an outgoing request optimizer for creating fast and scalable applicati
 [![Codacy](https://api.codacy.com/project/badge/Grade/e806d72373414fd9818ab2a403f1b36d)](https://www.codacy.com/app/Acanguven/puzzle-warden?utm_source=github.com&utm_medium=referral&utm_content=puzzle-js/puzzle-warden&utm_campaign=Badge_Grade) 
 
 ## Features
-- 📥  **Smart Caching** Caches requests by converting HTTP requests to smart key strings. ✅
-- 🚧  **Request Holder** Stopping same request to be sent multiple times. ✅
-- 🔌  **Support** Warden can be used with anything but it supports [request](https://github.com/request/request) out of the box. ✅
-- 😎  **Easy Implementation** Warden can be easily implemented with a few lines of codes. ✅
-- 🔁  **Request Retry** Requests will automatically be re-attempted on recoverable errors. 📝
-- 📇  **Schema Parser** Warden uses a schema which can be provided by you for parsing JSON faster. 📝
-- 🚥  **API Queue** Throttles API calls to protect target service. 📝
-- 👻  **Request Shadowing** Copies a fraction of traffic to a new deployment for observation. 📝
-- 🚉  **Reverse Proxy** It can be deployable as an external application which can serve as a reverse proxy. 📝
-- 📛  **Circuit Breaker** Immediately refuses new requests to provide time for the API to become healthy. 📝
+- 📥 **Smart Caching** Caches requests by converting HTTP requests to smart key strings. ✅
+- 🚧 **Request Holder** Stopping same request to be sent multiple times. ✅
+- 🔌 **Support** Warden can be used with anything but it supports [request](https://github.com/request/request) out of the box. ✅
+- 😎 **Easy Implementation** Warden can be easily implemented with a few lines of codes. ✅
+- 🔁 **Request Retry** Requests will automatically be re-attempted on recoverable errors. 📝
+- 📇 **Schema Parser** Warden uses a schema which can be provided by you for parsing JSON faster. 📝
+- 🚥 **API Queue** Throttles API calls to protect target service. 📝
+- 👻 **Request Shadowing** Copies a fraction of traffic to a new deployment for observation. 📝
+- 🚉 **Reverse Proxy** It can be deployable as an external application which can serve as a reverse proxy. 📝
+- 📛 **Circuit Breaker** Immediately refuses new requests to provide time for the API to become healthy. 📝
 
 ![Warden Achitecture](./warden_architecture.svg)
 
@@ -35,11 +35,11 @@ Warden is an outgoing request optimizer for creating fast and scalable applicati
 ### Installing
 
 Yarn
-```
+```bash
 yarn add puzzle-warden
 ```
 Npm
-```
+```bash
 npm i puzzle-warden --save
 ```
 

demo.ts

@@ -0,0 +1,9 @@
+const request = require("request");
+
+
+request({
+    url: 'https://m.trendyol.com',
+    timeout: 5
+}, (err: any, res: any, data: any) => {
+    console.log(err.connect);
+});

jest.config.js

@@ -1,6 +1,6 @@
 module.exports = {
-  preset: 'ts-jest',
-  testEnvironment: 'node',
+  preset: "ts-jest",
+  testEnvironment: "node",
   collectCoverageFrom: [
     "src/**/*.ts",
     "!src/**/*.d.ts",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "puzzle-warden",
-  "version": "1.3.0",
+  "version": "1.4.0",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",
   "license": "MIT",

src/cache-then-network.ts

@@ -4,47 +4,53 @@ import {StreamType} from "./stream-factory";
 import {CachePlugin} from "./cache-factory";
 import request from "request";
 
+interface CacheDecoratedResponse extends ResponseChunk {
+    cacheHit: boolean;
+}
+
+interface CacheDecoratedRequest extends RequestChunk {
+    cacheHit: boolean;
+}
 
 class CacheThenNetwork extends WardenStream {
-  private readonly storage: CachePlugin;
-  private readonly ms?: number;
-
-  constructor(plugin: CachePlugin, ms?: number) {
-    super(StreamType.CACHE);
-
-    this.ms = ms;
-    this.storage = plugin;
-  }
-
-  async onResponse(chunk: ResponseChunk, callback: TransformCallback): Promise<void> {
-    if (!chunk.cacheHit && !chunk.error && chunk.response) {
-      if(chunk.response.headers["set-cookie"]){
-        console.warn('Detected dangerous response with set-cookie header, not caching', chunk.key);
-      }else{
-        await this.storage.set(chunk.key, chunk.response, this.ms);
-      }
+    private readonly storage: CachePlugin;
+    private readonly ms?: number;
+
+    constructor(plugin: CachePlugin, ms?: number) {
+        super(StreamType.CACHE);
+
+        this.ms = ms;
+        this.storage = plugin;
+    }
+
+    async onResponse(chunk: CacheDecoratedResponse, callback: TransformCallback): Promise<void> {
+        if (!chunk.cacheHit && !chunk.error && chunk.response) {
+            if (chunk.response.headers["set-cookie"]) {
+                console.warn('Detected dangerous response with set-cookie header, not caching', chunk.key);
+            } else {
+                await this.storage.set(chunk.key, chunk.response, this.ms);
+            }
+        }
+
+        callback(undefined, chunk);
     }
 
-    callback(undefined, chunk);
-  }
-
-  async onRequest(chunk: RequestChunk, callback: TransformCallback): Promise<void> {
-    const cachedData = await this.storage.get(chunk.key) as request.Response;
-
-    if (cachedData) {
-      this.respond({
-        key: chunk.key,
-        cb: chunk.cb,
-        response: cachedData,
-        cacheHit: true
-      });
-      callback(undefined, null);
-    } else {
-      callback(undefined, chunk);
+    async onRequest(chunk: CacheDecoratedRequest, callback: TransformCallback): Promise<void> {
+        const cachedData = await this.storage.get(chunk.key) as request.Response;
+
+        if (cachedData) {
+            this.respond<CacheDecoratedResponse>({
+                ...chunk,
+                response: cachedData,
+                cacheHit: true
+            });
+            callback(undefined, null);
+        } else {
+            callback(undefined, chunk);
+        }
     }
-  }
 }
 
 export {
-  CacheThenNetwork
+    CacheThenNetwork
 };

src/holder.ts

@@ -2,9 +2,6 @@ import {RequestChunk, ResponseChunk, WardenStream} from "./warden-stream";
 import {TransformCallback} from "stream";
 import {StreamType} from "./stream-factory";
 
-interface HolderConfiguration {
-
-}
 
 class Holder extends WardenStream {
   private holdQueue: { [key: string]: RequestChunk[] | null } = {};
@@ -20,10 +17,8 @@ class Holder extends WardenStream {
     if (holdQueue) {
       holdQueue.forEach(holdChunk => {
         this.respond({
+          ...chunk,
           cb: holdChunk.cb,
-          key: chunk.key,
-          response: chunk.response,
-          error: chunk.error
         });
       });
 
@@ -48,6 +43,5 @@ class Holder extends WardenStream {
 }
 
 export {
-  HolderConfiguration,
   Holder
 };

src/network.ts

@@ -21,8 +21,7 @@ class Network extends WardenStream {
   onRequest(chunk: RequestChunk, callback: TransformCallback): void {
     this.requestWrapper.request[chunk.requestOptions.method](chunk.requestOptions, (error, response) => {
       this.respond({
-        key: chunk.key,
-        cb: chunk.cb,
+        ...chunk,
         response,
         error
       });

src/retry.ts

@@ -0,0 +1,43 @@
+import {RequestChunk, ResponseChunk, WardenStream} from "./warden-stream";
+import {TransformCallback} from "stream";
+
+interface RetryDecoratedResponse extends ResponseChunk {
+    retryCount: number;
+}
+
+interface RetryDecoratedRequest extends RequestChunk {
+    retryCount: number;
+}
+
+class Retry extends WardenStream {
+    retryLimit: number;
+
+    constructor(retryLimit: number) {
+        super('Retry');
+
+        this.retryLimit = retryLimit;
+    }
+
+    async onResponse(chunk: RetryDecoratedResponse, callback: TransformCallback): Promise<void> {
+        if (chunk.error && chunk.retryCount <= this.retryLimit) {
+            this.request<RetryDecoratedRequest>({
+                ...chunk,
+                retryCount: chunk.retryCount + 1
+            });
+            callback(undefined, null);
+        } else {
+            callback(undefined, chunk);
+        }
+    }
+
+    async onRequest(chunk: RetryDecoratedRequest, callback: TransformCallback): Promise<void> {
+        callback(null, {
+            ...chunk,
+            retryCount: 0
+        } as RequestChunk);
+    }
+}
+
+export {
+    Retry
+};

src/stream-factory.ts

@@ -3,19 +3,22 @@ import {CacheFactory} from "./cache-factory";
 import {StreamHead} from "./stream-head";
 import {Holder} from "./holder";
 import {RequestWrapper} from "./request-wrapper";
+import {Retry} from "./retry";
 
 const enum StreamType {
   HOLDER = 'holder',
   CACHE = 'cache',
   NETWORK = 'network',
   QUEUE = 'queue',
   CIRCUIT = 'circuit',
-  HEAD = 'head'
+  HEAD = 'head',
+  RETRY = 'retry'
 }
 
 enum ConfigurableStream {
   HOLDER = StreamType.HOLDER,
-  CACHE = StreamType.CACHE
+  CACHE = StreamType.CACHE,
+  RETRY = StreamType.RETRY
 }
 
 
@@ -41,6 +44,8 @@ class StreamFactory {
       //   throw new Error('Not implemented');
       case StreamType.NETWORK:
         return new Network(this.requestWrapper) as unknown as U;
+      case StreamType.RETRY:
+        return new Retry(3) as unknown as U;
       case StreamType.HEAD:
         return new StreamHead() as unknown as U;
       default: