NET-286: Storage node constants (#254)

Added constants StorageNode.STREAMR_GERMANY and StorageNode.STREAMR_DOCKER_DEV. The values are instances of StorageNode class.

The parameter of addToStorageNode, removeFromStorageNode and getStreamPartsByStorageNode can now be a StorageNode instance or a string.

Author: teogeb

README.md

@@ -98,6 +98,10 @@ const stream = await client.getOrCreateStream({
     name: 'My awesome stream created via the API',
 })
 console.log(`Stream ${stream.id} has been created!`)
+
+// Optional: to enable historical data resends, add the stream to a storage node
+await stream.addToStorageNode(StorageNode.STREAMR_GERMANY)
+
 // Do something with the stream, for example call stream.publish(message)
 ```
 

src/Config.ts

@@ -13,6 +13,7 @@ import { BytesLike } from '@ethersproject/bytes'
 import { isAddress } from '@ethersproject/address'
 import has from 'lodash/has'
 import get from 'lodash/get'
+import { StorageNode } from './stream/StorageNode'
 
 export type EthereumConfig = ExternalProvider|JsonRpcFetchFunc
 
@@ -144,7 +145,7 @@ export const STREAM_CLIENT_DEFAULTS: StrictStreamrClientOptions = {
         templateSidechainAddress: '0xf1E9d6E254BeA3f0129018AcA1A50AEcb7D528be',
     },
     storageNode: {
-        address: '0x31546eEA76F2B2b3C5cC06B1c93601dc35c9D916',
+        address: StorageNode.STREAMR_GERMANY.getAddress(),
         url: 'https://corea1.streamr.network:8001'
     },
     cache: {

src/rest/StreamEndpoints.ts

@@ -15,6 +15,7 @@ import { EthereumAddress } from '../types'
 import { StreamrClient } from '../StreamrClient'
 // TODO change this import when streamr-client-protocol exports StreamMessage type or the enums types directly
 import { ContentType, EncryptionType, SignatureType, StreamMessageType } from 'streamr-client-protocol/dist/src/protocol/message_layer/StreamMessage'
+import { StorageNode } from '../stream/StorageNode'
 
 const debug = debugFactory('StreamrClient')
 
@@ -255,7 +256,8 @@ export class StreamEndpoints {
         return json
     }
 
-    async getStreamPartsByStorageNode(address: EthereumAddress) {
+    async getStreamPartsByStorageNode(node: StorageNode|EthereumAddress) {
+        const address = (node instanceof StorageNode) ? node.getAddress() : node
         type ItemType = { id: string, partitions: number}
         const json = await authFetch<ItemType[]>(getEndpointUrl(this.client.options.restUrl, 'storageNodes', address, 'streams'), this.client.session)
         let result: StreamPart[] = []

src/stream/StorageNode.ts

@@ -2,6 +2,9 @@ import { EthereumAddress } from '../types'
 
 export class StorageNode {
 
+    static STREAMR_GERMANY = new StorageNode('0x31546eEA76F2B2b3C5cC06B1c93601dc35c9D916')
+    static STREAMR_DOCKER_DEV = new StorageNode('0xde1112f631486CfC759A50196853011528bC5FA0')
+
     private _address: EthereumAddress
 
     constructor(address: EthereumAddress) {

src/stream/index.ts

@@ -5,6 +5,7 @@ import authFetch from '../rest/authFetch'
 
 import { StorageNode } from './StorageNode'
 import { StreamrClient } from '../StreamrClient'
+import { EthereumAddress } from '../types'
 
 // TODO explicit types: e.g. we never provide both streamId and id, or both streamPartition and partition
 export type StreamPartDefinitionOptions = { streamId?: string, streamPartition?: number, id?: string, partition?: number, stream?: Stream|string }
@@ -223,7 +224,8 @@ export class Stream {
         await this.update()
     }
 
-    async addToStorageNode(address: string, { timeout = 30000 }: { timeout: number } = { timeout: 30000 }) {
+    async addToStorageNode(node: StorageNode|EthereumAddress, { timeout = 30000 }: { timeout: number } = { timeout: 30000 }) {
+        const address = (node instanceof StorageNode) ? node.getAddress() : node
         // currently we support only one storage node
         // -> we can validate that the given address is that address
         // -> remove this comparison when we start to support multiple storage nodes
@@ -258,7 +260,8 @@ export class Stream {
         throw new Error(`Unexpected response code ${response.status} when fetching stream storage status`)
     }
 
-    async removeFromStorageNode(address: string) {
+    async removeFromStorageNode(node: StorageNode|EthereumAddress) {
+        const address = (node instanceof StorageNode) ? node.getAddress() : node
         await authFetch(
             getEndpointUrl(this._client.options.restUrl, 'streams', this.id, 'storageNodes', address),
             this._client.session,

test/integration/Encryption.test.js

@@ -6,6 +6,7 @@ import { Defer } from '../../src/utils'
 import { StreamrClient } from '../../src/StreamrClient'
 import { GroupKey } from '../../src/stream/Encryption'
 import Connection from '../../src/Connection'
+import { StorageNode } from '../../src/stream/StorageNode'
 
 import config from './config'
 
@@ -107,7 +108,7 @@ describe('decryption', () => {
             requireEncryptedData: true,
         })
 
-        await stream.addToStorageNode(config.clientOptions.storageNode.address)
+        await stream.addToStorageNode(StorageNode.STREAMR_DOCKER_DEV)
 
         publishTestMessages = getPublishTestMessages(client, {
             stream

test/integration/GapFill.test.ts

@@ -9,6 +9,7 @@ import { Stream } from '../../src/stream'
 import { Subscriber, Subscription } from '../../src/subscribe'
 import { MessageRef } from 'streamr-client-protocol/dist/src/protocol/message_layer'
 import { StreamrClientOptions } from '../../src'
+import { StorageNode } from '../../src/stream/StorageNode'
 
 const MAX_MESSAGES = 10
 
@@ -47,7 +48,7 @@ describeRepeats('GapFill with resends', () => {
             requireSignedData: true,
             name: uid('stream')
         })
-        await stream.addToStorageNode(config.clientOptions.storageNode.address)
+        await stream.addToStorageNode(StorageNode.STREAMR_DOCKER_DEV)
 
         client.debug('connecting before test <<')
         publishTestMessages = getPublishTestMessages(client, stream.id)

test/integration/MultipleClients.test.js

@@ -5,6 +5,7 @@ import { describeRepeats, uid, fakePrivateKey, getWaitForStorage, getPublishTest
 import { StreamrClient } from '../../src/StreamrClient'
 import { counterId, Defer, pLimitFn } from '../../src/utils'
 import Connection from '../../src/Connection'
+import { StorageNode } from '../../src/stream/StorageNode'
 
 import config from './config'
 
@@ -51,7 +52,7 @@ describeRepeats('PubSub with multiple clients', () => {
         stream = await mainClient.createStream({
             name: uid('stream')
         })
-        await stream.addToStorageNode(config.clientOptions.storageNode.address)
+        await stream.addToStorageNode(StorageNode.STREAMR_DOCKER_DEV)
     })
 
     afterEach(async () => {

test/integration/ResendReconnect.test.ts

@@ -8,6 +8,7 @@ import config from './config'
 import { Stream } from '../../src/stream'
 import { Subscription } from '../../src'
 import { PublishRequest } from 'streamr-client-protocol/dist/src/protocol/control_layer'
+import { StorageNode } from '../../src/stream/StorageNode'
 
 const createClient = (opts = {}) => new StreamrClient({
     ...config.clientOptions,
@@ -35,7 +36,7 @@ describe('resend/reconnect', () => {
             name: uid('resends')
         })
 
-        await stream.addToStorageNode(config.clientOptions.storageNode.address)
+        await stream.addToStorageNode(StorageNode.STREAMR_DOCKER_DEV)
     }, 10000)
 
     beforeEach(async () => {

test/integration/Resends.test.ts

@@ -7,6 +7,7 @@ import Connection from '../../src/Connection'
 
 import config from './config'
 import { Stream } from '../../src/stream'
+import { StorageNode } from '../../src/stream/StorageNode'
 
 const MAX_MESSAGES = 10
 const WAIT_FOR_STORAGE_TIMEOUT = 6000
@@ -76,7 +77,7 @@ describe('StreamrClient resends', () => {
                     name: uid('resends')
                 })
 
-                await stream.addToStorageNode(config.clientOptions.storageNode.address)
+                await stream.addToStorageNode(StorageNode.STREAMR_DOCKER_DEV)
             })
 
             beforeEach(async () => {
@@ -337,7 +338,7 @@ describe('StreamrClient resends', () => {
                     name: uid('resends')
                 })
 
-                await stream.addToStorageNode(config.clientOptions.storageNode.address)
+                await stream.addToStorageNode(StorageNode.STREAMR_DOCKER_DEV)
 
                 publishTestMessages = getPublishTestMessages(client, {
                     stream