Error handling improvements.

.github/workflows/test-build.yml

@@ -184,7 +184,11 @@ jobs:
           npm install
           npm link streamr-client
           ./gradlew fatjar
-      - name: run-client-testing
-        timeout-minutes: 10
+      - uses: nick-invision/retry@v2
+        name: run-client-testing
         working-directory: streamr-client-testing
-        run: java -jar build/libs/client_testing-1.0-SNAPSHOT.jar -s $TEST_NAME -c config/$CONFIG_NAME.conf -n $NUM_MESSAGES
+        with:
+          max_attempts: 2
+          timeout_minutes: 3
+          retry_on: error
+          command: java -jar build/libs/client_testing-1.0-SNAPSHOT.jar -s $TEST_NAME -c config/$CONFIG_NAME.conf -n $NUM_MESSAGES

src/Config.ts

@@ -20,6 +20,8 @@ export type EthereumConfig = ExternalProvider|JsonRpcFetchFunc
  * @category Important
  */
 export type StrictStreamrClientOptions = {
+  /** Custom human-readable debug id for client. Used in logging. Unique id will be generated regardless. */
+    id?: string,
     /**
     * Authentication: identity used by this StreamrClient instance.
     * Can contain member privateKey or (window.)ethereum
@@ -169,8 +171,8 @@ export default function ClientConfig(opts: StreamrClientOptions = {}) {
             ...opts.dataUnion
         },
         cache: {
-            ...opts.cache,
             ...STREAM_CLIENT_DEFAULTS.cache,
+            ...opts.cache,
         }
         // NOTE: sidechain is not merged with the defaults
     }

src/Ethereum.js

@@ -20,8 +20,8 @@ export default class StreamrEthereum {
             const key = auth.privateKey
             const address = getAddress(computeAddress(key))
             this._getAddress = async () => address
-            this.getSigner = () => new Wallet(key, this.getMainnetProvider())
-            this.getSidechainSigner = async () => new Wallet(key, this.getSidechainProvider())
+            this._getSigner = () => new Wallet(key, this.getMainnetProvider())
+            this._getSidechainSigner = async () => new Wallet(key, this.getSidechainProvider())
         } else if (auth.ethereum) {
             this._getAddress = async () => {
                 try {
@@ -61,6 +61,10 @@ export default class StreamrEthereum {
         }
     }
 
+    canEncrypt() {
+        return !!(this._getAddress && this._getSigner)
+    }
+
     async getAddress() {
         if (!this._getAddress) {
             // _getAddress is assigned in constructor

src/StreamrClient.ts

@@ -188,7 +188,7 @@ export class StreamrClient extends EventEmitter { // eslint-disable-line no-rede
     // TODO annotate connection parameter as internal parameter if possible?
     constructor(options: StreamrClientOptions = {}, connection?: StreamrConnection) {
         super()
-        this.id = counterId(`${this.constructor.name}:${uid}`)
+        this.id = counterId(`${this.constructor.name}:${uid}${options.id || ''}`)
         this.debug = Debug(this.id)
 
         this.options = Config(options)
@@ -217,9 +217,9 @@ export class StreamrClient extends EventEmitter { // eslint-disable-line no-rede
             .on('disconnected', this.onConnectionDisconnected)
             .on('error', this.onConnectionError)
 
+        this.ethereum = new StreamrEthereum(this)
         this.publisher = Publisher(this)
         this.subscriber = new Subscriber(this)
-        this.ethereum = new StreamrEthereum(this)
 
         Plugin(this, new StreamEndpoints(this))
         Plugin(this, new LoginEndpoints(this))
@@ -357,10 +357,14 @@ export class StreamrClient extends EventEmitter { // eslint-disable-line no-rede
         let subTask: Todo
         let sub: Todo
         const hasResend = !!(opts.resend || opts.from || opts.to || opts.last)
-        const onEnd = () => {
+        const onEnd = (err?: Error) => {
             if (sub && typeof onMessage === 'function') {
                 sub.off('message', onMessage)
             }
+
+            if (err) {
+                throw err
+            }
         }
 
         if (hasResend) {
@@ -429,6 +433,13 @@ export class StreamrClient extends EventEmitter { // eslint-disable-line no-rede
         return this.getAddress()
     }
 
+    /**
+     * True if authenticated with private key/ethereum provider
+     */
+    canEncrypt() {
+        return this.ethereum.canEncrypt()
+    }
+
     /**
      * Get token balance in "wei" (10^-18 parts) for given address
      */

src/publish/Encrypt.ts

@@ -10,15 +10,27 @@ const { StreamMessage } = MessageLayer
 type PublisherKeyExhangeAPI = ReturnType<typeof PublisherKeyExhange>
 
 export default function Encrypt(client: StreamrClient) {
-    const publisherKeyExchange = PublisherKeyExhange(client, {
-        groupKeys: {
-            ...client.options.groupKeys,
+    let publisherKeyExchange: ReturnType<typeof PublisherKeyExhange>
+
+    function getPublisherKeyExchange() {
+        if (!publisherKeyExchange) {
+            publisherKeyExchange = PublisherKeyExhange(client, {
+                groupKeys: {
+                    ...client.options.groupKeys,
+                }
+            })
         }
-    })
+        return publisherKeyExchange
+    }
+
     async function encrypt(streamMessage: MessageLayer.StreamMessage, stream: Stream) {
+        if (!client.canEncrypt()) {
+            return
+        }
+
         if (
-            !publisherKeyExchange.hasAnyGroupKey(stream.id)
-            && !stream.requireEncryptedData
+            !stream.requireEncryptedData
+            && !getPublisherKeyExchange().hasAnyGroupKey(stream.id)
         ) {
             // not needed
             return
@@ -27,19 +39,23 @@ export default function Encrypt(client: StreamrClient) {
         if (streamMessage.messageType !== StreamMessage.MESSAGE_TYPES.MESSAGE) {
             return
         }
-        const groupKey = await publisherKeyExchange.useGroupKey(stream.id)
+        const groupKey = await getPublisherKeyExchange().useGroupKey(stream.id)
         await EncryptionUtil.encryptStreamMessage(streamMessage, groupKey)
     }
 
     return Object.assign(encrypt, {
         setNextGroupKey(...args: Parameters<PublisherKeyExhangeAPI['setNextGroupKey']>) {
-            return publisherKeyExchange.setNextGroupKey(...args)
+            return getPublisherKeyExchange().setNextGroupKey(...args)
         },
         rotateGroupKey(...args: Parameters<PublisherKeyExhangeAPI['rotateGroupKey']>) {
-            return publisherKeyExchange.rotateGroupKey(...args)
+            return getPublisherKeyExchange().rotateGroupKey(...args)
+        },
+        start() {
+            return getPublisherKeyExchange().start()
         },
         stop() {
-            return publisherKeyExchange.stop()
+            if (!publisherKeyExchange) { return Promise.resolve() }
+            return getPublisherKeyExchange().stop()
         }
     })
 }

src/publish/index.js

@@ -122,6 +122,7 @@ function getCreateStreamMessage(client) {
             [streamId, streamPartition, publisherId, msgChainId].join('|')
         ),
         ...cacheOptions,
+        maxAge: undefined
     }), {
         clear() {
             mem.clear(getMsgChainer)
@@ -188,6 +189,9 @@ function getCreateStreamMessage(client) {
         rotateGroupKey(maybeStreamId) {
             return encrypt.rotateGroupKey(maybeStreamId)
         },
+        startKeyExchange() {
+            return encrypt.start()
+        },
         clear() {
             computeStreamPartition.clear()
             getMsgChainer.clear()
@@ -300,6 +304,9 @@ export default function Publisher(client) {
                 throw error
             }
         },
+        async startKeyExchange() {
+            return createStreamMessage.startKeyExchange()
+        },
         async stop() {
             sendQueue.clear()
             createStreamMessage.clear()