Segments cache: replace addToSegment, removeFromSegment and setChangeNumber with update method for simplicity

Author: EmilianoSanchez

src/storages/AbstractSegmentsCacheSync.ts

@@ -39,6 +39,7 @@ export abstract class AbstractSegmentsCacheSync implements ISegmentsCacheSync {
    * For client-side synchronizer: the method is not used.
    */
   registerSegments(names: string[]): boolean { return false; }
+  update(name: string, addedKeys: string[], removedKeys: string[], changeNumber: number) { return false; }
 
   /**
    * For server-side synchronizer: get the list of segments to fetch changes.

src/storages/inMemory/SegmentsCacheInMemory.ts

@@ -11,6 +11,18 @@ export class SegmentsCacheInMemory extends AbstractSegmentsCacheSync {
   private segmentCache: Record<string, ISet<string>> = {};
   private segmentChangeNumber: Record<string, number> = {};
 
+  update(name: string, addedKeys: string[], removedKeys: string[], changeNumber: number) {
+    const keySet = this.segmentCache[name] || new _Set<string>();
+
+    addedKeys.forEach(k => keySet.add(k));
+    removedKeys.forEach(k => keySet.delete(k));
+
+    this.segmentCache[name] = keySet;
+    this.segmentChangeNumber[name] = changeNumber;
+
+    return addedKeys.length > 0 || removedKeys.length > 0;
+  }
+
   addToSegment(name: string, segmentKeys: string[]): boolean {
     const values = this.segmentCache[name];
     const keySet = values ? values : new _Set<string>();

src/storages/inRedis/SegmentsCacheInRedis.ts

@@ -17,6 +17,23 @@ export class SegmentsCacheInRedis implements ISegmentsCacheAsync {
     this.keys = keys;
   }
 
+  /**
+   * Update the given segment `name` with the lists of `addedKeys`, `removedKeys` and `changeNumber`.
+   * The returned promise is resolved if the operation success, with `true` if the segment was updated (i.e., some key was added or removed),
+   * or rejected if it fails (e.g., Redis operation fails).
+   */
+  update(name: string, addedKeys: string[], removedKeys: string[], changeNumber: number) {
+    const segmentKey = this.keys.buildSegmentNameKey(name);
+
+    return Promise.all([
+      addedKeys.length && this.redis.sadd(segmentKey, addedKeys),
+      removedKeys.length && this.redis.srem(segmentKey, removedKeys),
+      this.redis.set(this.keys.buildSegmentTillKey(name), changeNumber + '')
+    ]).then(() => {
+      return addedKeys.length > 0 || removedKeys.length > 0;
+    });
+  }
+
   addToSegment(name: string, segmentKeys: string[]) {
     const segmentKey = this.keys.buildSegmentNameKey(name);
 

src/storages/pluggable/SegmentsCachePluggable.ts

@@ -22,6 +22,23 @@ export class SegmentsCachePluggable implements ISegmentsCacheAsync {
     this.wrapper = wrapper;
   }
 
+  /**
+   * Update the given segment `name` with the lists of `addedKeys`, `removedKeys` and `changeNumber`.
+   * The returned promise is resolved if the operation success, with `true` if the segment was updated (i.e., some key was added or removed),
+   * or rejected if it fails (e.g., wrapper operation fails).
+   */
+  update(name: string, addedKeys: string[], removedKeys: string[], changeNumber: number) {
+    const segmentKey = this.keys.buildSegmentNameKey(name);
+
+    return Promise.all<any>([
+      addedKeys.length && this.wrapper.addItems(segmentKey, addedKeys),
+      removedKeys.length && this.wrapper.removeItems(segmentKey, removedKeys),
+      this.wrapper.set(this.keys.buildSegmentTillKey(name), changeNumber + '')
+    ]).then(() => {
+      return addedKeys.length > 0 || removedKeys.length > 0;
+    });
+  }
+
   /**
    * Add a list of `segmentKeys` to the given segment `name`.
    * The returned promise is resolved when the operation success

src/storages/types.ts

@@ -258,6 +258,7 @@ export interface ISegmentsCacheBase {
   getRegisteredSegments(): MaybeThenable<string[]> // only for Server-Side
   setChangeNumber(name: string, changeNumber: number): MaybeThenable<boolean | void> // only for Server-Side
   getChangeNumber(name: string): MaybeThenable<number> // only for Server-Side
+  update(name: string, addedKeys: string[], removedKeys: string[], changeNumber: number): MaybeThenable<boolean> // only for Server-Side
   clear(): MaybeThenable<boolean | void>
 }
 
@@ -271,6 +272,7 @@ export interface ISegmentsCacheSync extends ISegmentsCacheBase {
   getKeysCount(): number // only used for telemetry
   setChangeNumber(name: string, changeNumber: number): boolean | void
   getChangeNumber(name?: string): number
+  update(name: string, addedKeys: string[], removedKeys: string[], changeNumber: number): boolean // only for Server-Side
   resetSegments(segmentsData: MySegmentsData | IMySegmentsResponse): boolean // only for Sync Client-Side
   clear(): void
 }
@@ -283,6 +285,7 @@ export interface ISegmentsCacheAsync extends ISegmentsCacheBase {
   getRegisteredSegments(): Promise<string[]>
   setChangeNumber(name: string, changeNumber: number): Promise<boolean | void>
   getChangeNumber(name: string): Promise<number>
+  update(name: string, addedKeys: string[], removedKeys: string[], changeNumber: number): Promise<boolean>
   clear(): Promise<boolean | void>
 }
 

src/sync/polling/updaters/segmentChangesUpdater.ts

@@ -1,12 +1,9 @@
 import { ISegmentChangesFetcher } from '../fetchers/types';
 import { ISegmentsCacheBase } from '../../../storages/types';
 import { IReadinessManager } from '../../../readiness/types';
-import { MaybeThenable } from '../../../dtos/types';
-import { findIndex } from '../../../utils/lang';
 import { SDK_SEGMENTS_ARRIVED } from '../../../readiness/constants';
 import { ILogger } from '../../../logger/types';
 import { LOG_PREFIX_INSTANTIATION, LOG_PREFIX_SYNC_SEGMENTS } from '../../../logger/constants';
-import { thenable } from '../../../utils/promise/thenable';
 
 type ISegmentChangesUpdater = (fetchOnlyNew?: boolean, segmentName?: string, noCache?: boolean, till?: number) => Promise<boolean>
 
@@ -30,31 +27,22 @@ export function segmentChangesUpdaterFactory(
 
   let readyOnAlreadyExistentState = true;
 
-  function updateSegment(segmentName: string, noCache?: boolean, till?: number, fetchOnlyNew?: boolean) {
+  function updateSegment(segmentName: string, noCache?: boolean, till?: number, fetchOnlyNew?: boolean): Promise<boolean> {
     log.debug(`${LOG_PREFIX_SYNC_SEGMENTS}Processing segment ${segmentName}`);
     let sincePromise = Promise.resolve(segments.getChangeNumber(segmentName));
 
     return sincePromise.then(since => {
       // if fetchOnlyNew flag, avoid processing already fetched segments
-      if (fetchOnlyNew && since !== -1) return -1;
-
-      return segmentChangesFetcher(since, segmentName, noCache, till).then(function (changes) {
-        let changeNumber = -1;
-        const results: MaybeThenable<boolean | void>[] = [];
-        changes.forEach(x => {
-          if (x.added.length > 0) results.push(segments.addToSegment(segmentName, x.added));
-          if (x.removed.length > 0) results.push(segments.removeFromSegment(segmentName, x.removed));
-          if (x.added.length > 0 || x.removed.length > 0) {
-            results.push(segments.setChangeNumber(segmentName, x.till));
-            changeNumber = x.till;
-          }
-
-          log.debug(`${LOG_PREFIX_SYNC_SEGMENTS}Processed ${segmentName} with till = ${x.till}. Added: ${x.added.length}. Removed: ${x.removed.length}`);
+      return fetchOnlyNew && since !== -1 ?
+        false :
+        segmentChangesFetcher(since, segmentName, noCache, till).then((changes) => {
+          return Promise.all(changes.map(x => {
+            log.debug(`${LOG_PREFIX_SYNC_SEGMENTS}Processing ${segmentName} with till = ${x.till}. Added: ${x.added.length}. Removed: ${x.removed.length}`);
+            return segments.update(x.name, x.added, x.removed, x.till);
+          })).then((updates) => {
+            return updates.some(update => update);
+          });
         });
-        // If at least one storage operation result is a promise, join all in a single promise.
-        if (results.some(result => thenable(result))) return Promise.all(results).then(() => changeNumber);
-        return changeNumber;
-      });
     });
   }
   /**
@@ -75,16 +63,12 @@ export function segmentChangesUpdaterFactory(
     let segmentsPromise = Promise.resolve(segmentName ? [segmentName] : segments.getRegisteredSegments());
 
     return segmentsPromise.then(segmentNames => {
-      // Async fetchers are collected here.
-      const updaters: Promise<number>[] = [];
-
-      for (let index = 0; index < segmentNames.length; index++) {
-        updaters.push(updateSegment(segmentNames[index], noCache, till, fetchOnlyNew));
-      }
+      // Async fetchers
+      const updaters = segmentNames.map(segmentName => updateSegment(segmentName, noCache, till, fetchOnlyNew));
 
       return Promise.all(updaters).then(shouldUpdateFlags => {
         // if at least one segment fetch succeeded, mark segments ready
-        if (findIndex(shouldUpdateFlags, v => v !== -1) !== -1 || readyOnAlreadyExistentState) {
+        if (shouldUpdateFlags.some(update => update) || readyOnAlreadyExistentState) {
           readyOnAlreadyExistentState = false;
           if (readiness) readiness.segments.emit(SDK_SEGMENTS_ARRIVED);
         }