Merge branch 'master' into ETH-81-expose-signature-transport

Author: jtakalai

jest.setup.js

@@ -1,16 +1,8 @@
 import GitRevisionPlugin from 'git-revision-webpack-plugin'
-import Debug from 'debug'
 
 const pkg = require('./package.json')
 
 export default async () => {
-    if (process.env.DEBUG_CONSOLE) {
-        // Use debug as console log
-        // This prevents jest messing with console output
-        // Ensuring debug messages are printed alongside console messages, in the correct order
-        console.log = Debug('Streamr::CONSOLE') // eslint-disable-line no-console
-    }
-
     if (!process.env.GIT_VERSION) {
         const gitRevisionPlugin = new GitRevisionPlugin()
         const [GIT_VERSION, GIT_COMMITHASH, GIT_BRANCH] = await Promise.all([

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "streamr-client",
-  "version": "5.2.1",
+  "version": "5.3.0-beta.0",
   "description": "JavaScript client library for Streamr",
   "private": true,
   "repository": {
@@ -117,7 +117,6 @@
     "pretty-bytes": "^5.6.0",
     "process": "^0.11.10",
     "sinon": "^9.2.4",
-    "streamr-test-utils": "^1.3.1",
     "terser-webpack-plugin": "^5.1.1",
     "ts-jest": "^26.5.1",
     "ts-loader": "^8.0.17",
@@ -160,6 +159,7 @@
     "quick-lru": "^6.0.0",
     "readable-stream": "^3.6.0",
     "streamr-client-protocol": "^8.0.0-beta.2",
+    "streamr-test-utils": "^1.3.1",
     "ts-toolbelt": "^9.3.12",
     "uuid": "^8.3.2",
     "webpack-node-externals": "^2.5.2",

package.json

@@ -71,6 +71,10 @@ export type StrictStreamrClientOptions = {
         templateMainnetAddress: EthereumAddress
         templateSidechainAddress: EthereumAddress
     },
+    storageNode: {
+        address: EthereumAddress
+        url: string
+    }
     cache: {
         maxSize: number,
         maxAge: number
@@ -139,6 +143,10 @@ export const STREAM_CLIENT_DEFAULTS: StrictStreamrClientOptions = {
         templateMainnetAddress: '0x5FE790E3751dd775Cb92e9086Acd34a2adeB8C7b',
         templateSidechainAddress: '0xf1E9d6E254BeA3f0129018AcA1A50AEcb7D528be',
     },
+    storageNode: {
+        address: '0x31546eEA76F2B2b3C5cC06B1c93601dc35c9D916',
+        url: 'https://corea1.streamr.network:8001'
+    },
     cache: {
         maxSize: 10000,
         maxAge: 30 * 60 * 1000, // 30 minutes
@@ -160,7 +168,8 @@ export default function ClientConfig(opts: StreamrClientOptions = {}) {
         'dataUnion.factoryMainnetAddress',
         'dataUnion.factorySidechainAddress',
         'dataUnion.templateMainnetAddress',
-        'dataUnion.templateSidechainAddress'
+        'dataUnion.templateSidechainAddress',
+        'storageNode.address'
     ])
 
     const options: StrictStreamrClientOptions = {
@@ -174,7 +183,7 @@ export default function ClientConfig(opts: StreamrClientOptions = {}) {
             ...STREAM_CLIENT_DEFAULTS.cache,
             ...opts.cache,
         }
-        // NOTE: sidechain is not merged with the defaults
+        // NOTE: sidechain and storageNode settings are not merged with the defaults
     }
 
     const parts = options.url!.split('?')

src/Config.ts

@@ -7,7 +7,7 @@ import debugFactory from 'debug'
 import { getEndpointUrl } from '../utils'
 import { validateOptions } from '../stream/utils'
 import { Stream, StreamOperation, StreamProperties } from '../stream'
-import StreamPart from '../stream/StreamPart'
+import { StreamPart } from '../stream/StreamPart'
 import { isKeyExchangeStream } from '../stream/KeyExchange'
 
 import authFetch, { ErrorCode, NotFoundError } from './authFetch'

src/rest/StreamEndpoints.ts

@@ -1,6 +1,6 @@
 import { EthereumAddress } from '../types'
 
-export default class StorageNode {
+export class StorageNode {
 
     private _address: EthereumAddress
 

src/stream/StorageNode.ts

@@ -1,4 +1,4 @@
-export default class StreamPart {
+export class StreamPart {
 
     _streamId: string
     _streamPartition: number

src/stream/StreamPart.ts

@@ -1,7 +1,10 @@
+import fetch from 'node-fetch'
+import { getAddress } from '@ethersproject/address'
+import { waitForCondition } from 'streamr-test-utils'
 import { getEndpointUrl } from '../utils'
 import authFetch from '../rest/authFetch'
 
-import StorageNode from './StorageNode'
+import { StorageNode } from './StorageNode'
 import { StreamrClient } from '../StreamrClient'
 
 // TODO explicit types: e.g. we never provide both streamId and id, or both streamPartition and partition
@@ -221,6 +224,12 @@ export class Stream {
     }
 
     async addToStorageNode(address: string) {
+        // 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
+        if (getAddress(address) !== this._client.options.storageNode.address) {
+            throw new Error('Unknown storage node: ' + address)
+        }
         await authFetch(
             getEndpointUrl(this._client.options.restUrl, 'streams', this.id, 'storageNodes'),
             this._client.session,
@@ -231,6 +240,24 @@ export class Stream {
                 })
             },
         )
+        // wait for propagation: the storage node sees the database change in E&E and
+        // is ready to store the any stream data which we publish
+        const TIMEOUT = 30 * 1000
+        const POLL_INTERVAL = 500
+        await waitForCondition(() => this.isStreamStoredInStorageNode(this.id), TIMEOUT, POLL_INTERVAL,
+            () => `Propagation timeout when adding stream to a storage node: ${this.id}`)
+    }
+
+    private async isStreamStoredInStorageNode(streamId: string) {
+        const url = `${this._client.options.storageNode.url}/api/v1/streams/${encodeURIComponent(streamId)}/storage/partitions/0`
+        const response = await fetch(url)
+        if (response.status === 200) {
+            return true
+        }
+        if (response.status === 404) { // eslint-disable-line padding-line-between-statements
+            return false
+        }
+        throw new Error(`Unexpected response code ${response.status} when fetching stream storage status`)
     }
 
     async removeFromStorageNode(address: string) {

src/stream/index.ts

@@ -106,6 +106,8 @@ describe('decryption', () => {
             requireEncryptedData: true,
         })
 
+        await stream.addToStorageNode(config.clientOptions.storageNode.address)
+
         publishTestMessages = getPublishTestMessages(client, {
             stream
         })

test/integration/Encryption.test.js

@@ -47,6 +47,7 @@ describeRepeats('GapFill with resends', () => {
             requireSignedData: true,
             name: uid('stream')
         })
+        await stream.addToStorageNode(config.clientOptions.storageNode.address)
 
         client.debug('connecting before test <<')
         publishTestMessages = getPublishTestMessages(client, stream.id)