3. Promisify subscribe/unsubscribe methods

[timoxley]

Changes subscribe/unsubscribe to resolve when the subscribed/unsubscribed events fire respectively.

Simplifies consumer API significantly:

After

const subscription = await client.subscribe('stream1', onMessage)
// subscribed
await client.unsubscribe(subscription)
// unsubscribed

Before

const subscription = client.subscribe('stream1')
subscription.once('subscribed', () => {
  // subscribed
  subscription.once('unsubscribed', () => {
    // unsubscribed
  })
  client.unsubscribe(subscription)
})

I've got something similar as a WIP for resends.

Also should make the error handling story a bit clearer, since subscribe/unsubscribe could throw synchronously or maybe emit an error event.

e.g. to catch all possible errors with the existing streamr-client, you basically have to do something like the below,

client.on('error', onError)
let subscription
try {
    subscription = client.subscribe('stream1', onMessage)
    subscription.on('error', onError)
    subscription.once('subscribed', () => {
        // subscribed
        subscription.once('unsubscribed', () => {
            // unsubscribed
        })
        try {
            client.unsubscribe(subscription)
        } catch (err) {
            onError(err)
        }
    })
} catch (err) {
    onError(err)
}

And it gets even more hairy if you want to robustly detect which call was in progress when the error was thrown/emitted:

let subscription
try {
    subscription = client.subscribe('stream1', onMessage)
} catch (err) {
    onSubscribeError(err) // catch sync errors
}

if (subscription) {
    let onSubscribed

    const onSubscribeErrorAsync = (err) => {
        subscription.off('subscribed', onSubscribed) // remember to remove the success handler if you have an error
        onUnsubscribeError(err)
    }

    onSubscribed = () => {
        subscription.off('error', onSubscribeErrorAsync) // stop catching async errors after success
        // subscribed

        let success = false
        try {
            client.unsubscribe(subscription)
            success = true
        } catch (err) {
            onUnsubscribeError(err)
        }

        // need to detect sync error above so we don't attach spurrious listeners
        if (success) {
            subscription.once('error', onSubscribeErrorAsync) // catch async errors
            let onUnsubscribed
            // same err pattern for unsubscribed
            const onUnsubscribeErrorAsync = (err) => {
                subscription.off('unsubscribed', onUnsubscribed)
                onUnsubscribeError(err)
            }

            onUnsubscribed = () => {
                subscription.off('error', onUnsubscribeErrorAsync)
                // unsubscribed
            }
            subscription.once('unsubscribed', onUnsubscribed)
            subscription.once('error', onUnsubscribeError)
        }
    }

    subscription.once('subscribed', onSubscribed)
}

With these new changes, the above pattern is equivalent to:

const subscription = await client.subscribe('stream1', onMessage).catch(onSubscribeError)

if (subscription) {
  await client.unsubscribe(subscription).catch(onUnsubscribeError)
}

Also related to error handling, this should probably be updated to leverage the new requestId property so it can more accurately detect & reject errors, currently it always rejects if it sees any error event event on the subscription during a subscribe/unsubscribe, meaning it's not impossible for an unrelated error to cause rejection.

[harbu]

Out of curiosity, how does this work if we are in the middle of awaiting method subscribe above?

await Promise.all([
            sub.waitForSubscribed(),
            this._resendAndSubscribe(sub),
        ])

[harbu]

What is this .then(() => this) needed for?