[FFM-1262] polling processor, streaming processor

package.json

@@ -6,6 +6,7 @@
     "node": ">= 12"
   },
   "devDependencies": {
+    "@types/eventsource": "^1.1.6",
     "@types/jest": "~26.0.23",
     "@types/node": "^14.17.11",
     "@typescript-eslint/eslint-plugin": "~4.28.2",
@@ -35,7 +36,9 @@
   "license": "Apache-2.0",
   "dependencies": {
     "@openapitools/openapi-generator-cli": "^2.3.10",
+    "eventsource": "^1.1.0",
     "jwt-decode": "^3.1.2",
+    "node-cache": "^5.1.2",
     "tslib": "~2.3.0"
   },
   "volta": {

src/cache.ts

@@ -0,0 +1,31 @@
+import { FeatureConfig, Segment } from "./openapi";
+
+export type CacheValueType = FeatureConfig | Segment;
+
+export interface Cache {
+  set(key: string, value: CacheValueType): void;
+  get(key: string): CacheValueType;
+  del(key: string): void;
+}
+
+export class SimpleCache implements Cache {
+  private cache = {};
+
+  set(key: string, value: CacheValueType): void {
+    this.cache[key] = value;
+  }
+  get(key: string): CacheValueType {
+    return this.cache[key];
+  }
+  del(key: string): void {
+    delete this.cache[key];
+  }
+}
+
+export const formatFlagKey = (key: string): string => {
+    return `flags/${key}`
+}
+
+export const formatSegmentKey = (key: string): string => {
+    return `segments/${key}`
+}

src/index.ts

@@ -1,7 +1,10 @@
 import jwt_decode from 'jwt-decode';
 import { Claims, defaultOptions, Options } from './types';
 import { Configuration, DefaultApi, Target } from './openapi';
-import { VERSION } from './utils';
+import { log, VERSION } from './utils';
+import { PollingProcessor } from './polling';
+import { StreamProcessor } from './streaming';
+import * as events from 'events';
 
 export class CfClient {
   private api: DefaultApi;
@@ -11,6 +14,9 @@ export class CfClient {
   private configuration: Configuration;
   private options: Options;
   private cluster = '1';
+  private eventBus = new events.EventEmitter();
+  private pollProcessor: PollingProcessor;
+  private streamProcessor: StreamProcessor;
 
   constructor(sdkKey: string, options: Options = {}) {
     this.sdkKey = sdkKey;
@@ -43,45 +49,33 @@ export class CfClient {
     return this.authToken;
   }
 
-  fetchData(): Promise<[void, void]> {
-    const flags = this.api
-      .getFeatureConfig(this.environment, {
-        params: {
-          cluster: this.cluster,
-        },
-      })
-      .then((response) => {
-        // prepare cache for storing flags
-        console.log(response.data);
-      })
-      .catch((error: Error) => {
-        console.error('Error loading flags', error);
-        throw error;
-      });
-    const segments = this.api
-      .getAllSegments(this.environment, {
-        params: {
-          cluster: this.cluster,
-        },
-      })
-      .then((response) => {
-        // prepare cache for storing segments
-        console.log(response.data);
-      })
-      .catch((error: Error) => {
-        console.error('Error loading segments', error);
-        throw error;
-      });
-
-    return Promise.all([flags, segments]);
-  }
-
   async run(): Promise<void> {
     await this.authenticate();
-    await this.fetchData();
+    this.pollProcessor = new PollingProcessor(
+      this.environment,
+      this.cluster,
+      this.api,
+      this.options,
+      this.eventBus,
+    );
+
+    // start processors
+    this.pollProcessor.start();
     if (this.options.enableStream) {
-      // setup SSE client
+      this.streamProcessor = new StreamProcessor(
+        this.api,
+        this.sdkKey,
+        this.environment,
+        this.authToken,
+        this.options,
+        this.cluster,
+        this.eventBus,
+      );
+
+      this.streamProcessor.start();
     }
+
+    log.info('finished setting up processors');
   }
 
   boolVariation(
@@ -92,8 +86,15 @@ export class CfClient {
     console.log(identifier, target);
     return defaultValue;
   }
+
+  close(): void {
+    this.pollProcessor.close();
+    if (this.options.enableStream) {
+      this.streamProcessor.close();
+    }
+  }
 }
 
 // need to setup example project and remove this later
-const client = new CfClient('sdk key');
+const client = new CfClient('1c100d25-4c3f-487b-b198-3b3d01df5794');
 client.boolVariation('test', null, false);

src/log.ts

@@ -0,0 +1,27 @@
+export interface Logger {
+  trace(message?: unknown, ...optionalParams: unknown[]): void;
+  debug(message?: unknown, ...optionalParams: unknown[]): void;
+  info(message?: unknown, ...optionalParams: unknown[]): void;
+  warn(message?: unknown, ...optionalParams: unknown[]): void;
+  error(message?: unknown, ...optionalParams: unknown[]): void;
+}
+
+// Wrapper for console
+export class ConsoleLog implements Logger {
+
+  trace(message?: unknown, ...optionalParams: unknown[]): void {
+    console.trace(message, ...optionalParams);
+  }
+  debug(message?: unknown, ...optionalParams: unknown[]): void {
+    console.debug(message, ...optionalParams);
+  }
+  info(message?: unknown, ...optionalParams: unknown[]): void {
+    console.info(message, ...optionalParams);
+  }
+  warn(message?: unknown, ...optionalParams: unknown[]): void {
+    console.warn(message, ...optionalParams);
+  }
+  error(message?: unknown, ...optionalParams: unknown[]): void {
+    console.error(message, ...optionalParams);
+  }
+}