Merge pull request #23 from drone/FFM-1500

[FFM-1500] ability to listen on ff events

Author: enver-bisevac

README.md

@@ -158,6 +158,46 @@ client
   });
 ```
 
+## Listening on events
+
+You can listen on these events:
+
+- `Event.READY` - SDK successfully initialized
+- `Event.FAILED` - SDK throws an error
+- `Event.CHANGED` - any new version of flag or segment triggers this event, if segment is changed then it will find all flags with segment match operator
+
+Methods:
+
+```typescript
+on(Event.READY, () => {
+  console.log('READY');
+});
+
+on(Event.FAILED, () => {
+  console.log('FAILED');
+});
+
+on(Event.CHANGED, (identifier) => {
+  console.log('Changed', identifier);
+});
+```
+
+and if you want to remove the `functionReference` listener for `Event.READY`:
+
+```
+off(Event.READY, functionReference);
+```
+
+or if you want to remove all listeners on `Event.READY`:
+
+```
+off(Event.READY);
+```
+
+or if you call `off()` without params it will close the client.
+
+> All events are applicable to off() function.
+
 ## License
 
 Licensed under the APLv2.

example/cf_reactive.mjs

@@ -0,0 +1,24 @@
+import CfClient, { Event } from '@harnessio/ff-nodejs-server-sdk';
+
+CfClient.init('1c100d25-4c3f-487b-b198-3b3d01df5794');
+
+CfClient.on(Event.READY, () => {
+  console.log('READY');
+});
+
+CfClient.on(Event.FAILED, (error) => {
+  console.log('FAILED with err:', error);
+});
+
+CfClient.on(Event.CHANGED, (identifier) => {
+  console.log('Changed', identifier);
+});
+
+console.log('Starting application');
+
+setInterval(async () => {
+  const value = await CfClient.boolVariation('test', null, false);
+  console.log('Evaluation for flag test and target none: ', value);
+}, 10000);
+
+console.log('Application started');

example/package-lock.json

@@ -4,9 +4,12 @@
   "description": "NodeJS Server SDK sample",
   "main": "index.js",
   "scripts": {
-    "cf": "node index_cf.mjs",
-    "client": "node index.mjs",
-    "start": "node index.js",
+    "cf": "node --enable-source-maps index_cf.mjs",
+    "cf_reactive": "node --enable-source-maps cf_reactive.mjs",
+    "reactive": "node --enable-source-maps reactive.js",
+    "client": "node --enable-source-maps index.mjs",
+    "wait": "node --enable-source-maps wait.js",
+    "start": "node --enable-source-maps index.js",
     "test": "echo \"Error: no test specified\" && exit 1"
   },
   "author": "[email protected]",

example/package.json

@@ -0,0 +1,31 @@
+const { Client, Event } = require('@harnessio/ff-nodejs-server-sdk');
+
+const client = new Client('1c100d25-4c3f-487b-b198-3b3d01df5794', {
+  enableStream: false,
+});
+
+client.on(Event.READY, () => {
+  console.log('READY');
+});
+
+client.on(Event.FAILED, () => {
+  console.log('FAILED');
+});
+
+client.on(Event.CHANGED, (identifier) => {
+  if (identifier === 'test') {
+    console.log('test flag changed');
+  }
+});
+
+console.log('Starting application');
+
+setInterval(async () => {
+  const target = {
+    identifier: 'harness',
+  };
+  const value = await client.boolVariation('test', target, false);
+  console.log('Evaluation for flag test and target: ', value, target);
+}, 10000);
+
+console.log('Application started');

example/reactive.js

@@ -0,0 +1,20 @@
+const { Client } = require('@harnessio/ff-nodejs-server-sdk');
+
+console.log('Starting application');
+const client = new Client('1c100d25-4c3f-487b-b198-3b3d01df5794');
+client
+  .waitForInitialization()
+  .then(() => {
+    setInterval(async () => {
+      const target = {
+        identifier: 'harness',
+      };
+      const value = await client.boolVariation('test', target, false);
+      console.log('Evaluation for flag test and target: ', value, target);
+    }, 10000);
+
+    console.log('Application started');
+  })
+  .catch((error) => {
+    console.log('Error', error);
+  });

example/wait.js

@@ -1,6 +1,6 @@
 {
   "name": "@harnessio/ff-nodejs-server-sdk",
-  "version": "1.1.1",
+  "version": "1.2.0",
   "description": "Feature flags SDK for NodeJS environments",
   "main": "dist/cjs/index.js",
   "module": "dist/esm/index.js",

package.json

@@ -6,10 +6,16 @@ export class SimpleCache implements KeyValueStore {
   set(key: string, value: unknown): void {
     this.cache[key] = value;
   }
+
   get(key: string): unknown {
     return this.cache[key];
   }
+
   del(key: string): void {
     delete this.cache[key];
   }
+
+  keys(): string[] {
+    return Object.keys(this.cache);
+  }
 }

src/cache.ts

@@ -1,4 +1,4 @@
-import * as events from 'events';
+import EventEmitter from 'events';
 import jwt_decode from 'jwt-decode';
 import axios from 'axios';
 import axiosRetry from 'axios-retry';
@@ -9,7 +9,7 @@ import { PollerEvent, PollingProcessor } from './polling';
 import { StreamProcessor } from './streaming';
 import { Evaluator } from './evaluator';
 import { defaultOptions } from './constants';
-import { Repository, StorageRepository } from './repository';
+import { Repository, RepositoryEvent, StorageRepository } from './repository';
 import {
   MetricEvent,
   MetricsProcessor,
@@ -26,7 +26,7 @@ enum Processor {
   METRICS,
 }
 
-export enum ClientEvent {
+export enum Event {
   READY = 'ready',
   FAILED = 'failed',
   CHANGED = 'changed',
@@ -41,7 +41,7 @@ export default class Client {
   private configuration: Configuration;
   private options: Options;
   private cluster = '1';
-  private eventBus = new events.EventEmitter();
+  private eventBus = new EventEmitter();
   private pollProcessor: PollingProcessor;
   private streamProcessor: StreamProcessor;
   private metricsProcessor: MetricsProcessorInterface;
@@ -72,6 +72,7 @@ export default class Client {
     this.repository = new StorageRepository(
       this.options.cache,
       this.options.store,
+      this.eventBus,
     );
     this.evaluator = new Evaluator(this.repository);
     this.api = new ClientApi(this.configuration);
@@ -86,7 +87,7 @@ export default class Client {
 
     this.eventBus.on(PollerEvent.ERROR, () => {
       this.failure = true;
-      this.eventBus.emit(ClientEvent.FAILED);
+      this.eventBus.emit(Event.FAILED);
     });
 
     this.eventBus.on(StreamEvent.READY, () => {
@@ -95,7 +96,7 @@ export default class Client {
 
     this.eventBus.on(StreamEvent.ERROR, () => {
       this.failure = true;
-      this.eventBus.emit(ClientEvent.FAILED);
+      this.eventBus.emit(Event.FAILED);
     });
 
     this.eventBus.on(MetricEvent.READY, () => {
@@ -104,7 +105,7 @@ export default class Client {
 
     this.eventBus.on(MetricEvent.ERROR, () => {
       this.failure = true;
-      this.eventBus.emit(ClientEvent.FAILED);
+      this.eventBus.emit(Event.FAILED);
     });
 
     this.eventBus.on(StreamEvent.CONNECTED, () => {
@@ -114,6 +115,47 @@ export default class Client {
     this.eventBus.on(StreamEvent.DISCONNECTED, () => {
       this.pollProcessor.start();
     });
+
+    for (const event of Object.values(RepositoryEvent)) {
+      this.eventBus.on(event, (identifier) => {
+        switch (event) {
+          case RepositoryEvent.FLAG_STORED:
+          case RepositoryEvent.FLAG_DELETED:
+            this.eventBus.emit(Event.CHANGED, identifier);
+            break;
+          case RepositoryEvent.SEGMENT_STORED:
+          case RepositoryEvent.SEGMENT_DELETED:
+            // find all flags where segment match and emit the event
+            this.repository
+              .findFlagsBySegment(identifier)
+              .then((values: string[]) => {
+                values.forEach((value) =>
+                  this.eventBus.emit(Event.CHANGED, value),
+                );
+              });
+            break;
+        }
+      });
+    }
+  }
+
+  on(event: Event, callback: (...args: unknown[]) => void): void {
+    const arrayObjects = [];
+
+    for (const value of Object.values(Event)) {
+      arrayObjects.push(value);
+    }
+    if (arrayObjects.includes(event)) {
+      this.eventBus.on(event, callback);
+    }
+  }
+
+  off(event?: string, callback?: () => void): void {
+    if (event) {
+      this.eventBus.off(event, callback);
+    } else {
+      this.close();
+    }
   }
 
   private async authenticate(): Promise<void> {
@@ -145,10 +187,10 @@ export default class Client {
       this.waitForInitialize = Promise.reject(this.failure);
     } else {
       this.waitForInitialize = new Promise((resolve, reject) => {
-        this.eventBus.once(ClientEvent.READY, () => {
+        this.eventBus.once(Event.READY, () => {
           resolve(this);
         });
-        this.eventBus.once(ClientEvent.FAILED, reject);
+        this.eventBus.once(Event.FAILED, reject);
       });
     }
     return this.waitForInitialize;
@@ -183,7 +225,7 @@ export default class Client {
     }
 
     this.initialized = true;
-    this.eventBus.emit(ClientEvent.READY);
+    this.eventBus.emit(Event.READY);
   }
 
   private async run(): Promise<void> {
@@ -304,5 +346,6 @@ export default class Client {
     if (this.metricsProcessor) {
       this.metricsProcessor.close();
     }
+    this.eventBus.removeAllListeners();
   }
 }

src/client.ts

@@ -1,4 +1,4 @@
-import Client, { ClientEvent } from './client';
+import Client, { Event } from './client';
 import LRU from 'lru-cache';
 import { Options, Target } from './types';
 import { Logger } from './log';
@@ -7,7 +7,7 @@ import { FileStore } from './store';
 
 export {
   Client,
-  ClientEvent,
+  Event,
   Options,
   Target,
   AsyncKeyValueStore,
@@ -54,6 +54,12 @@ export default {
   ): Promise<Record<string, unknown>> {
     return this.instance.jsonVariation(identifier, target, defaultValue);
   },
+  on: function (event: Event, callback: (...args: unknown[]) => void): void {
+    this.instance.on(event, callback);
+  },
+  off: function (event?: Event, callback?: () => void): void {
+    this.instance.off(event, callback);
+  },
   close: function (): void {
     return this.instance.close();
   },