refactor(sdk): Use Mapping instead of CachingMap

packages/sdk/src/contracts/StreamRegistry.ts

@@ -40,9 +40,9 @@ import {
     isPublicPermissionQuery,
     streamPermissionToSolidityType
 } from '../permission'
-import { CachingMap } from '../utils/CachingMap'
 import { filter, map } from '../utils/GeneratorUtils'
 import { LoggerFactory } from '../utils/LoggerFactory'
+import { Mapping } from '../utils/Mapping'
 import { ChainEventPoller } from './ChainEventPoller'
 import { ContractFactory } from './ContractFactory'
 import { ObservableContract, initContractEventGateway, waitForTx } from './contract'
@@ -102,13 +102,8 @@ const streamContractErrorProcessor = (err: any, streamId: StreamID, registry: st
     }
 }
 
-const formCacheKeyPrefix = (streamId: StreamID): string => {
-    return `${streamId}|`
-}
-
-const invalidateCache = (cache: CachingMap<string, any, any>, streamId: StreamID): void => {
-    const matchTarget = (s: string) => s.startsWith(formCacheKeyPrefix(streamId))
-    cache.invalidate(matchTarget)
+const invalidateCache = (cache: Mapping<[StreamID, ...any[]], any>, streamId: StreamID): void => {
+    cache.invalidate(([key]) => key === streamId)
 }
 
 @scoped(Lifecycle.ContainerScoped)
@@ -124,10 +119,10 @@ export class StreamRegistry {
     private readonly config: Pick<StrictStreamrClientConfig, 'contracts' | 'cache' | '_timeouts'>
     private readonly authentication: Authentication
     private readonly logger: Logger
-    private readonly metadataCache: CachingMap<string, StreamMetadata, [StreamID]>
-    private readonly publisherCache: CachingMap<string, boolean, [StreamID, UserID]>
-    private readonly subscriberCache: CachingMap<string, boolean, [StreamID, UserID]>
-    private readonly publicSubscribePermissionCache: CachingMap<string, boolean, [StreamID]>
+    private readonly metadataCache: Mapping<[StreamID], StreamMetadata>
+    private readonly publisherCache: Mapping<[StreamID, UserID], boolean>
+    private readonly subscriberCache: Mapping<[StreamID, UserID], boolean>
+    private readonly publicSubscribePermissionCache: Mapping<[StreamID], boolean>
 
     /** @internal */
     constructor(
@@ -168,33 +163,33 @@ export class StreamRegistry {
             }),
             loggerFactory
         })
-        this.metadataCache = new CachingMap((streamId: StreamID) => {
-            return this.getStreamMetadata_nonCached(streamId)
-        }, {
-            ...config.cache,
-            cacheKey: ([streamId]) => formCacheKeyPrefix(streamId)
+        this.metadataCache = new Mapping({
+            valueFactory: (streamId: StreamID) => {
+                return this.getStreamMetadata_nonCached(streamId)
+            },
+            ...config.cache
         })
-        this.publisherCache = new CachingMap((streamId: StreamID, userId: UserID) => {
-            return this.isStreamPublisherOrSubscriber_nonCached(streamId, userId, StreamPermission.PUBLISH)
-        }, {
-            ...config.cache,
-            cacheKey: ([streamId, userId]) =>`${formCacheKeyPrefix(streamId)}${userId}`
+        this.publisherCache = new Mapping({
+            valueFactory: (streamId: StreamID, userId: UserID) => {
+                return this.isStreamPublisherOrSubscriber_nonCached(streamId, userId, StreamPermission.PUBLISH)
+            },
+            ...config.cache
         })
-        this.subscriberCache = new CachingMap((streamId: StreamID, userId: UserID) => {
-            return this.isStreamPublisherOrSubscriber_nonCached(streamId, userId, StreamPermission.SUBSCRIBE)
-        }, {
-            ...config.cache,
-            cacheKey: ([streamId, userId]) =>`${formCacheKeyPrefix(streamId)}${userId}`
+        this.subscriberCache = new Mapping({
+            valueFactory: (streamId: StreamID, userId: UserID) => {
+                return this.isStreamPublisherOrSubscriber_nonCached(streamId, userId, StreamPermission.SUBSCRIBE)
+            }, 
+            ...config.cache
         })
-        this.publicSubscribePermissionCache = new CachingMap((streamId: StreamID) => {
-            return this.hasPermission({
-                streamId,
-                public: true,
-                permission: StreamPermission.SUBSCRIBE
-            })
-        }, {
-            ...config.cache,
-            cacheKey: ([streamId]) => formCacheKeyPrefix(streamId)
+        this.publicSubscribePermissionCache = new Mapping({
+            valueFactory: (streamId: StreamID) => {
+                return this.hasPermission({
+                    streamId,
+                    public: true,
+                    permission: StreamPermission.SUBSCRIBE
+                })
+            },
+            ...config.cache
         })
     }
 

packages/sdk/src/contracts/StreamStorageRegistry.ts

@@ -14,7 +14,7 @@ import { LoggerFactory } from '../utils/LoggerFactory'
 import { ChainEventPoller } from './ChainEventPoller'
 import { ContractFactory } from './ContractFactory'
 import { initContractEventGateway, waitForTx } from './contract'
-import { CachingMap } from '../utils/CachingMap'
+import { Mapping } from '../utils/Mapping'
 
 export interface StorageNodeAssignmentEvent {
     readonly streamId: StreamID
@@ -45,7 +45,7 @@ export class StreamStorageRegistry {
     private readonly config: Pick<StrictStreamrClientConfig, 'contracts' | '_timeouts'>
     private readonly authentication: Authentication
     private readonly logger: Logger
-    private readonly getStorageNodes_cached: CachingMap<string | typeof GET_ALL_STORAGE_NODES, EthereumAddress[], [string | undefined]>
+    private readonly storageNodesCache: Mapping<[StreamID | typeof GET_ALL_STORAGE_NODES], EthereumAddress[]>
 
     constructor(
         streamIdBuilder: StreamIDBuilder,
@@ -78,13 +78,11 @@ export class StreamStorageRegistry {
             )
         }), config.contracts.pollInterval)
         this.initStreamAssignmentEventListeners(eventEmitter, chainEventPoller, loggerFactory)
-        this.getStorageNodes_cached = new CachingMap((streamIdOrPath?: string) => {
-            return this.getStorageNodes_nonCached(streamIdOrPath)
-        }, {
-            ...config.cache,
-            cacheKey: ([streamIdOrPath]): string | typeof GET_ALL_STORAGE_NODES => {
-                return streamIdOrPath ?? GET_ALL_STORAGE_NODES
-            }
+        this.storageNodesCache = new Mapping({
+            valueFactory: (query: StreamID | typeof GET_ALL_STORAGE_NODES) => {
+                return this.getStorageNodes_nonCached(query)
+            },
+            ...config.cache
         })
     }
 
@@ -135,7 +133,7 @@ export class StreamStorageRegistry {
         await this.connectToContract()
         const ethersOverrides = await getEthersOverrides(this.rpcProviderSource, this.config)
         await waitForTx(this.streamStorageRegistryContract!.addStorageNode(streamId, nodeAddress, ethersOverrides))
-        this.getStorageNodes_cached.invalidate((key) => key === streamId)
+        this.storageNodesCache.invalidate((key) => key[0] === streamId)
     }
 
     async removeStreamFromStorageNode(streamIdOrPath: string, nodeAddress: EthereumAddress): Promise<void> {
@@ -144,7 +142,7 @@ export class StreamStorageRegistry {
         await this.connectToContract()
         const ethersOverrides = await getEthersOverrides(this.rpcProviderSource, this.config)
         await waitForTx(this.streamStorageRegistryContract!.removeStorageNode(streamId, nodeAddress, ethersOverrides))
-        this.getStorageNodes_cached.invalidate((key) => key === streamId)
+        this.storageNodesCache.invalidate((key) => key[0] === streamId)
     }
 
     async isStoredStream(streamIdOrPath: string, nodeAddress: EthereumAddress): Promise<boolean> {
@@ -192,13 +190,14 @@ export class StreamStorageRegistry {
     }
 
     async getStorageNodes(streamIdOrPath?: string): Promise<EthereumAddress[]> {
-        return this.getStorageNodes_cached.get(streamIdOrPath)
+        const query = (streamIdOrPath !== undefined) ? await this.streamIdBuilder.toStreamID(streamIdOrPath) : GET_ALL_STORAGE_NODES
+        return this.storageNodesCache.get(query)
     }
 
-    private async getStorageNodes_nonCached(streamIdOrPath?: string): Promise<EthereumAddress[]> {
+    private async getStorageNodes_nonCached(query: StreamID | typeof GET_ALL_STORAGE_NODES): Promise<EthereumAddress[]> {
         let queryResults: NodeQueryResult[]
-        if (streamIdOrPath !== undefined) {
-            const streamId = await this.streamIdBuilder.toStreamID(streamIdOrPath)
+        if (query !== GET_ALL_STORAGE_NODES) {
+            const streamId = query
             this.logger.debug('Get storage nodes of stream', { streamId })
             queryResults = await collect(this.theGraphClient.queryEntities<NodeQueryResult>(
                 (lastId: string, pageSize: number) => {

packages/sdk/src/publish/MessageFactory.ts

@@ -37,7 +37,7 @@ export class MessageFactory {
     private readonly streamId: StreamID
     private readonly authentication: Authentication
     private defaultPartition: number | undefined
-    private readonly defaultMessageChainIds: Mapping<[partition: number], string>
+    private readonly defaultMessageChainIds: Mapping<[number], string>
     private readonly prevMsgRefs: Map<string, MessageRef> = new Map()
     // eslint-disable-next-line max-len
     private readonly streamRegistry: Pick<StreamRegistry, 'getStreamMetadata' | 'hasPublicSubscribePermission' | 'isStreamPublisher' | 'invalidatePermissionCaches'>
@@ -53,7 +53,7 @@ export class MessageFactory {
         this.groupKeyQueue = opts.groupKeyQueue
         this.signatureValidator = opts.signatureValidator
         this.messageSigner = opts.messageSigner
-        this.defaultMessageChainIds = new Mapping({
+        this.defaultMessageChainIds = new Mapping<[number], string>({
             valueFactory: async () => {
                 return createRandomMsgChainId()
             }

packages/sdk/src/publish/Publisher.ts

@@ -1,7 +1,7 @@
 import { StreamID } from '@streamr/utils'
 import isString from 'lodash/isString'
 import pLimit from 'p-limit'
-import { Lifecycle, inject, scoped } from 'tsyringe'
+import { inject, Lifecycle, scoped } from 'tsyringe'
 import { Authentication, AuthenticationInjectionToken } from '../Authentication'
 import { NetworkNodeFacade } from '../NetworkNodeFacade'
 import { StreamIDBuilder } from '../StreamIDBuilder'
@@ -43,8 +43,8 @@ const parseTimestamp = (metadata?: PublishMetadata): number => {
 @scoped(Lifecycle.ContainerScoped)
 export class Publisher {
 
-    private readonly messageFactories: Mapping<[streamId: StreamID], MessageFactory>
-    private readonly groupKeyQueues: Mapping<[streamId: StreamID], GroupKeyQueue>
+    private readonly messageFactories: Mapping<[StreamID], MessageFactory>
+    private readonly groupKeyQueues: Mapping<[StreamID], GroupKeyQueue>
     private readonly concurrencyLimit = pLimit(1)
     private readonly node: NetworkNodeFacade
     private readonly streamRegistry: StreamRegistry

packages/sdk/src/subscribe/ordering/OrderMessages.ts

@@ -102,7 +102,7 @@ export class OrderMessages {
                 const chain = await this.chains.get(msg.getPublisherId(), msg.getMsgChainId())
                 chain.addMessage(msg)
             }
-            await Promise.all(this.chains.values().map((chain) => chain.waitUntilIdle()))
+            await Promise.all([...this.chains.values()].map((chain) => chain.waitUntilIdle()))
             this.outBuffer.endWrite()
         } catch (err) {
             this.outBuffer.endWrite(err)

packages/sdk/src/utils/Mapping.ts

@@ -1,7 +1,7 @@
 import { formLookupKey } from './utils'
 import LRU from '../../vendor/quick-lru'
 
-type KeyType = (string | number)[]
+type KeyType = (string | number | symbol)[]
 
 interface BaseOptions<K extends KeyType, V> {
     valueFactory: (...args: K) => Promise<V>
@@ -14,8 +14,8 @@ interface CacheMapOptions<K extends KeyType, V> extends BaseOptions<K, V> {
 
 type LazyMapOptions<K extends KeyType, V> = BaseOptions<K, V>
 
-// an wrapper object is used so that we can store undefined values
-interface ValueWrapper<V> {
+interface Item<K, V> {
+    key: K
     value: V
 }
 
@@ -26,7 +26,7 @@ interface ValueWrapper<V> {
  */
 export class Mapping<K extends KeyType, V> {
 
-    private readonly delegate: Map<string, ValueWrapper<V>>
+    private readonly delegate: Map<string, Item<K, V>>
     private readonly pendingPromises: Map<string, Promise<V>> = new Map()
     private readonly opts: CacheMapOptions<K, V> | LazyMapOptions<K, V>
 
@@ -37,45 +37,55 @@ export class Mapping<K extends KeyType, V> {
      **/
     constructor(opts: CacheMapOptions<K, V> | LazyMapOptions<K, V>) {
         if ('maxSize' in opts) {
-            this.delegate = new LRU<string, ValueWrapper<V>>({
+            this.delegate = new LRU<string, Item<K, V>>({
                 maxSize: opts.maxSize,
                 maxAge: opts.maxAge
             })
         } else {
-            this.delegate = new Map<string, ValueWrapper<V>>()
+            this.delegate = new Map<string, Item<K, V>>()
         }
         this.opts = opts
     }
 
-    async get(...args: K): Promise<V> {
-        const key = formLookupKey(...args)
-        const pendingPromise = this.pendingPromises.get(key)
+    async get(...key: K): Promise<V> {
+        const lookupKey = formLookupKey(...key)
+        const pendingPromise = this.pendingPromises.get(lookupKey)
         if (pendingPromise !== undefined) {
             return await pendingPromise
         } else {
-            let valueWrapper = this.delegate.get(key)
-            if (valueWrapper === undefined) {
-                const promise = this.opts.valueFactory(...args)
-                this.pendingPromises.set(key, promise)
+            let item = this.delegate.get(lookupKey)
+            if (item === undefined) {
+                const promise = this.opts.valueFactory(...key)
+                this.pendingPromises.set(lookupKey, promise)
                 let value
                 try {
                     value = await promise
                 } finally {
-                    this.pendingPromises.delete(key)
+                    this.pendingPromises.delete(lookupKey)
                 }
-                valueWrapper = { value }
-                this.delegate.set(key, valueWrapper)
+                item = { key, value }
+                this.delegate.set(lookupKey, item)
             }
-            return valueWrapper.value
+            return item.value
         }
     }
 
-    values(): V[] {
-        const result: V[] = []
-        for (const wrapper of this.delegate.values()) {
-            result.push(wrapper.value)
+    set(key: K, value: V): void {
+        this.delegate.set(formLookupKey(...key), { key, value })
+    }
+
+    invalidate(predicate: (key: K) => boolean): void {
+        for (const [lookupKey, item] of this.delegate.entries()) {
+            if (predicate(item.key)) {
+                this.delegate.delete(lookupKey)
+            }
+        }
+    }
+
+    *values(): IterableIterator<V> {
+        for (const item of this.delegate.values()) {
+            yield item.value
         }
-        return result
     }
 }
 

packages/sdk/src/utils/utils.ts

@@ -156,8 +156,8 @@ export function generateClientId(): string {
 
 // A unique internal identifier to some list of primitive values. Useful
 // e.g. as a map key or a cache key.
-export const formLookupKey = <K extends (string | number)[]>(...args: K): string => {
-    return args.join('|')
+export const formLookupKey = <K extends (string | number | symbol)[]>(...args: K): string => {
+    return args.map((a) => a.toString()).join('|')
 }
 
 /** @internal */

packages/sdk/test/unit/Mapping.test.ts

@@ -113,4 +113,17 @@ describe('Mapping', () => {
         await mapping.get('foo', 1)
         expect(valueFactory).toHaveBeenCalledTimes(2)
     })
+
+    it('invalidate', async () => {
+        const mapping = createLazyMap({
+            valueFactory: async (p1: string, p2: number) => `${p1}${p2}`,
+        })
+        mapping.set(['foo', 1], 'foo1')
+        mapping.set(['bar', 1], 'bar1')
+        await mapping.get('foo', 2)
+        await mapping.get('bar', 2)
+        expect([...mapping.values()]).toIncludeSameMembers(['foo1', 'bar1', 'foo2', 'bar2'])
+        mapping.invalidate(([p1]) => (p1 === 'bar'))
+        expect([...mapping.values()]).toIncludeSameMembers(['foo1', 'foo2'])
+    })
 })