remove metadata field, call populateMetadataCache()

Author: teogeb

packages/sdk/src/Stream.ts

@@ -31,7 +31,6 @@ export class Stream {
     /** @internal */
     constructor(
         id: StreamID,
-        _metadata: StreamMetadata,
         client: StreamrClient
     ) {
         this.id = id

packages/sdk/src/StreamrClient.ts

@@ -3,7 +3,7 @@ import './utils/PatchTsyringe'
 
 import { DhtAddress } from '@streamr/dht'
 import { ProxyDirection } from '@streamr/trackerless-network'
-import { DEFAULT_PARTITION_COUNT, EthereumAddress, HexString, StreamID, TheGraphClient, toEthereumAddress, toUserId } from '@streamr/utils'
+import { DEFAULT_PARTITION_COUNT, EthereumAddress, HexString, Logger, StreamID, TheGraphClient, toEthereumAddress, toUserId } from '@streamr/utils'
 import type { Overrides } from 'ethers'
 import EventEmitter from 'eventemitter3'
 import merge from 'lodash/merge'
@@ -52,6 +52,7 @@ import { LoggerFactory } from './utils/LoggerFactory'
 import { pOnce } from './utils/promises'
 import { convertPeerDescriptorToNetworkPeerDescriptor, createTheGraphClient } from './utils/utils'
 import { addStreamToStorageNode } from './utils/addStreamToStorageNode'
+import { map } from './utils/GeneratorUtils'
 
 // TODO: this type only exists to enable tsdoc to generate proper documentation
 export type SubscribeOptions = StreamDefinition & ExtraSubscribeOptions
@@ -73,6 +74,8 @@ export interface ExtraSubscribeOptions {
     erc1271Contract?: HexString
 }
 
+const logger = new Logger(module)
+
 /**
  * The main API used to interact with Streamr.
  *
@@ -357,8 +360,11 @@ export class StreamrClient {
      */
     async getStream(streamIdOrPath: string): Promise<Stream> {
         const streamId = await this.streamIdBuilder.toStreamID(streamIdOrPath)
-        const metadata = await this.streamRegistry.getStreamMetadata(streamId, false)
-        return new Stream(streamId, metadata, this)
+        // Check if the stream exists by querying its metadata. Throws an error if no metadata is found,
+        // indicating the stream doesn't exist. As a side-effect this populates StreamRegistry's metadata
+        // cache for future use, such as stream.getPartitionCount() calls.
+        await this.streamRegistry.getStreamMetadata(streamId, false)
+        return new Stream(streamId, this)
     }
 
     /**
@@ -376,7 +382,7 @@ export class StreamrClient {
         const streamId = await this.streamIdBuilder.toStreamID(props.id)
         const metadata = merge({ partitions: DEFAULT_PARTITION_COUNT }, omit(props, 'id') )
         await this.streamRegistry.createStream(streamId, metadata)
-        return new Stream(streamId, metadata, this)
+        return new Stream(streamId, this)
     }
 
     /**
@@ -434,12 +440,16 @@ export class StreamrClient {
         permissionFilter: SearchStreamsPermissionFilter | undefined,
         orderBy: SearchStreamsOrderBy = { field: 'id', direction: 'asc' }
     ): AsyncIterable<Stream> {
-        return this.streamRegistry.searchStreams(
+        logger.debug('Search for streams', { term, permissionFilter })
+        if ((term === undefined) && (permissionFilter === undefined)) {
+            throw new Error('Requires a search term or a permission filter')
+        }
+        const streamIds = this.streamRegistry.searchStreams(
             term,
             (permissionFilter !== undefined) ? toInternalSearchStreamsPermissionFilter(permissionFilter) : undefined,
-            orderBy,
-            this
+            orderBy
         )
+        return map(streamIds, (id) => new Stream(id, this))
     }
 
     // --------------------------------------------------------------------------------------------
@@ -572,8 +582,11 @@ export class StreamrClient {
      */
     async getStoredStreams(storageNodeAddress: HexString): Promise<{ streams: Stream[], blockNumber: number }> {
         const queryResult = await this.streamStorageRegistry.getStoredStreams(toEthereumAddress(storageNodeAddress))
+        for (const stream of queryResult.streams) {
+            this.streamRegistry.populateMetadataCache(stream.id, stream.metadata)
+        }
         return {
-            streams: queryResult.streams.map((item) => new Stream(item.id, item.metadata, this)),
+            streams: queryResult.streams.map((item) => new Stream(item.id, this)),
             blockNumber: queryResult.blockNumber
         }
     }

packages/sdk/src/contracts/StreamRegistry.ts

@@ -20,10 +20,8 @@ import { Lifecycle, inject, scoped } from 'tsyringe'
 import { Authentication, AuthenticationInjectionToken } from '../Authentication'
 import { ConfigInjectionToken, StrictStreamrClientConfig } from '../Config'
 import { RpcProviderSource } from '../RpcProviderSource'
-import { Stream } from '../Stream'
 import { StreamIDBuilder } from '../StreamIDBuilder'
 import { StreamMetadata, parseMetadata } from '../StreamMetadata'
-import { StreamrClient } from '../StreamrClient'
 import { StreamrClientError } from '../StreamrClientError'
 import type { StreamRegistryV5 as StreamRegistryContract } from '../ethereumArtifacts/StreamRegistryV5'
 import StreamRegistryArtifact from '../ethereumArtifacts/StreamRegistryV5Abi.json'
@@ -241,6 +239,7 @@ export class StreamRegistry {
             await this.ensureStreamIdInNamespaceOfAuthenticatedUser(domain, streamId)
             await waitForTx(this.streamRegistryContract!.createStream(path, JSON.stringify(metadata), ethersOverrides))
         }
+        this.populateMetadataCache(streamId, metadata)
     }
 
     private async ensureStreamIdInNamespaceOfAuthenticatedUser(address: EthereumAddress, streamId: StreamID): Promise<void> {
@@ -289,19 +288,21 @@ export class StreamRegistry {
         return parseMetadata(metadata)
     }
 
-    searchStreams(
+    async* searchStreams(
         term: string | undefined,
         permissionFilter: InternalSearchStreamsPermissionFilter | undefined,
-        orderBy: SearchStreamsOrderBy,
-        client: StreamrClient
-    ): AsyncIterable<Stream> {
-        return _searchStreams(
+        orderBy: SearchStreamsOrderBy
+    ): AsyncGenerator<StreamID> {
+        const queryResult = _searchStreams(
             term,
             permissionFilter,
             orderBy,
-            this.theGraphClient,
-            this.logger,
-            client)
+            this.theGraphClient)
+        for await (const item of queryResult) {
+            const id = toStreamID(item.stream.id)
+            this.populateMetadataCache(id, parseMetadata(item.stream.metadata))
+            yield id
+        }
     }
 
     getStreamPublishers(streamIdOrPath: string): AsyncIterable<UserID> {
@@ -539,6 +540,10 @@ export class StreamRegistry {
         return this.hasPublicSubscribePermission_cached.get(streamId)
     }
 
+    populateMetadataCache(streamId: StreamID, metadata: StreamMetadata): void {
+        this.getStreamMetadata_cached.set([streamId], metadata)
+    }
+
     invalidateMetadataCache(streamId: StreamID): void {
         this.logger.trace('Clear metadata cache for stream', { streamId })
         this.getStreamMetadata_cached.invalidate((s) => s.startsWith(formCacheKeyPrefix(streamId)))

packages/sdk/src/contracts/searchStreams.ts

@@ -1,10 +1,7 @@
-import { ChangeFieldType, GraphQLQuery, HexString, Logger, TheGraphClient, toStreamID, toUserId, UserID } from '@streamr/utils'
-import { Stream } from '../Stream'
+import { ChangeFieldType, GraphQLQuery, HexString, TheGraphClient, toUserId, UserID } from '@streamr/utils'
 import { ChainPermissions, convertChainPermissionsToStreamPermissions, PUBLIC_PERMISSION_USER_ID, StreamPermission } from '../permission'
-import { filter, map, unique } from '../utils/GeneratorUtils'
+import { filter, unique } from '../utils/GeneratorUtils'
 import { StreamQueryResult } from './StreamRegistry'
-import { parseMetadata } from '../StreamMetadata'
-import { StreamrClient } from '../StreamrClient'
 
 export interface SearchStreamsPermissionFilter {
     userId: HexString
@@ -35,25 +32,7 @@ export const toInternalSearchStreamsPermissionFilter = (filter: SearchStreamsPer
     }
 }
 
-export const searchStreams = (
-    term: string | undefined,
-    permissionFilter: InternalSearchStreamsPermissionFilter | undefined,
-    orderBy: SearchStreamsOrderBy,
-    theGraphClient: TheGraphClient,
-    logger: Logger,
-    client: StreamrClient
-): AsyncGenerator<Stream> => {
-    if ((term === undefined) && (permissionFilter === undefined)) {
-        throw new Error('Requires a search term or a permission filter')
-    }
-    logger.debug('Search for streams', { term, permissionFilter })
-    return map(
-        fetchSearchStreamsResultFromTheGraph(term, permissionFilter, orderBy, theGraphClient),
-        (item: SearchStreamsResultItem) => new Stream(toStreamID(item.stream.id), parseMetadata(item.stream.metadata), client)
-    )
-}
-
-async function* fetchSearchStreamsResultFromTheGraph(
+export async function* searchStreams(
     term: string | undefined,
     permissionFilter: InternalSearchStreamsPermissionFilter | undefined,
     orderBy: SearchStreamsOrderBy,

packages/sdk/src/utils/CachingMap.ts

@@ -2,6 +2,12 @@ import { MapKey } from '@streamr/utils'
 import pMemoize from 'p-memoize'
 import LRU from '../../vendor/quick-lru'
 
+interface Options<P, K> {
+    maxSize: number
+    maxAge: number
+    cacheKey: (args: P) => K
+}
+
 /**
  * Caches into a LRU cache capped at options.maxSize. See documentation for mem/p-memoize.
  * Won't call asyncFn again until options.maxAge or options.maxSize exceeded, or cachedAsyncFn.invalidate() is called.
@@ -18,14 +24,11 @@ export class CachingMap<K extends MapKey, V, P extends any[]> {
 
     private readonly cachedFn: (...args: P) => Promise<V>
     private readonly cache: LRU<K, { data: V, maxAge: number }>
+    private readonly opts: Options<P, K>
 
     constructor(
         asyncFn: (...args: P) => Promise<V>,
-        opts: {
-            maxSize: number
-            maxAge: number
-            cacheKey: (args: P) => K
-        }
+        opts: Options<P, K>
     ) {
         this.cache = new LRU<K, { data: V, maxAge: number }>({
             maxSize: opts.maxSize,
@@ -36,12 +39,17 @@ export class CachingMap<K extends MapKey, V, P extends any[]> {
             cache: this.cache,
             cacheKey: opts.cacheKey
         })
+        this.opts = opts
     }
 
     get(...args: P): Promise<V> {
         return this.cachedFn(...args)
     }
 
+    set(args: P, value: V): void {
+        this.cache.set(this.opts.cacheKey(args), { data: value, maxAge: this.opts.maxAge })
+    }
+
     invalidate(predicate: (key: K) => boolean): void {
         for (const key of this.cache.keys()) {
             if (predicate(key)) {

packages/sdk/test/test-utils/fake/FakeStreamRegistry.ts

@@ -167,6 +167,11 @@ export class FakeStreamRegistry implements Methods<StreamRegistry> {
         })
     }
 
+    // eslint-disable-next-line class-methods-use-this
+    populateMetadataCache(): void {
+        // no-op
+    }
+
     // eslint-disable-next-line class-methods-use-this
     invalidateMetadataCache(): void {
         // no-op
@@ -201,7 +206,7 @@ export class FakeStreamRegistry implements Methods<StreamRegistry> {
     }
 
     // eslint-disable-next-line class-methods-use-this
-    searchStreams(_term: string | undefined, _permissionFilter: InternalSearchStreamsPermissionFilter | undefined): AsyncIterable<Stream> {
+    searchStreams(_term: string | undefined, _permissionFilter: InternalSearchStreamsPermissionFilter | undefined): AsyncGenerator<StreamID> {
         throw new Error('not implemented')
     }
 

packages/sdk/test/unit/Stream.test.ts

@@ -6,7 +6,6 @@ describe('Stream', () => {
 
     it('getPartitionCount', () => {
         const stream = new Stream(
-            undefined as any,
             undefined as any,
             {
                 getStreamMetadata: async () => ({ partitions: 150 })

packages/sdk/test/unit/searchStreams.test.ts

@@ -3,7 +3,6 @@ import 'reflect-metadata'
 import { randomEthereumAddress } from '@streamr/test-utils'
 import { StreamID, TheGraphClient, collect, toStreamID } from '@streamr/utils'
 import { SearchStreamsResultItem, searchStreams } from '../../src/contracts/searchStreams'
-import { mockLoggerFactory } from '../test-utils/utils'
 
 const MOCK_USER = randomEthereumAddress()
 
@@ -44,9 +43,7 @@ describe('searchStreams', () => {
             '/',
             undefined,
             orderBy,
-            theGraphClient as any,
-            mockLoggerFactory().createLogger(module),
-            undefined as any,
+            theGraphClient as any
         ))
 
         const graphQLquery = ((theGraphClient as any).queryEntities as jest.Mock).mock.calls[0][0]()