feat(sdk): Raw publish (#3280)

Added the `StreamrClient#publishRaw()` method.

It is the counterpart to the existing `StreamrClient#subscribe({ raw:
true }`) method. In raw publish/subscribe mode, the SDK’s message
pipeline is bypassed: e.g. validation, decryption, and message ordering
are skipped.

Author: teogeb

CHANGELOG.md

@@ -14,6 +14,7 @@ Changes before Tatum release are not documented in this file.
 
 - Proxy connections now support bidirectionality and it is the default behavior. (https://github.com/streamr-dev/network/pull/3260)
 - Added `findProxyNodes` function to `StreamrClient` for discovering proxy nodes via Operator nodes.
+- Add `StreamrClient#publishRaw()` for publishing raw messages (https://github.com/streamr-dev/network/pull/3280)
 
 #### Changed
 

packages/sdk/src/StreamrClient.ts

@@ -2,21 +2,22 @@
  * Importing 'timers' ensures `setImmediate` is available in browsers,
  * as it's polyfilled by `timers-browserify`. In Node.js, it's already global.
  */
-import 'timers'
 import 'reflect-metadata'
+import 'timers'
 import './utils/PatchTsyringe'
 
 import { DhtAddress } from '@streamr/dht'
 import { ProxyDirection, StreamPartDeliveryOptions } from '@streamr/trackerless-network'
-import { DEFAULT_PARTITION_COUNT, EthereumAddress, HexString, Logger, 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'
 import omit from 'lodash/omit'
 import { container as rootContainer } from 'tsyringe'
 import { PublishMetadata, Publisher } from '../src/publish/Publisher'
-import { Identity, IdentityInjectionToken, SignerWithProvider } from './identity/Identity'
 import {
     ConfigInjectionToken,
     NetworkPeerDescriptor,
@@ -29,6 +30,7 @@ import { DestroySignal } from './DestroySignal'
 import { Message, convertStreamMessageToMessage } from './Message'
 import { MetricsPublisher } from './MetricsPublisher'
 import { NetworkNodeFacade } from './NetworkNodeFacade'
+import { ProxyNodeFinder } from './ProxyNodeFinder'
 import { RpcProviderSource } from './RpcProviderSource'
 import { Stream } from './Stream'
 import { StreamIDBuilder } from './StreamIDBuilder'
@@ -37,6 +39,7 @@ import { ChainEventPoller } from './contracts/ChainEventPoller'
 import { ContractFactory } from './contracts/ContractFactory'
 import { Operator } from './contracts/Operator'
 import { OperatorRegistry } from './contracts/OperatorRegistry'
+import { SponsorshipFactory } from './contracts/SponsorshipFactory'
 import { StorageNodeMetadata, StorageNodeRegistry } from './contracts/StorageNodeRegistry'
 import { StreamRegistry } from './contracts/StreamRegistry'
 import { StreamStorageRegistry } from './contracts/StreamStorageRegistry'
@@ -46,7 +49,10 @@ import { LocalGroupKeyStore, UpdateEncryptionKeyOptions } from './encryption/Loc
 import { PublisherKeyExchange } from './encryption/PublisherKeyExchange'
 import { getEthersOverrides as _getEthersOverrides } from './ethereumUtils'
 import { StreamrClientEventEmitter, StreamrClientEvents } from './events'
+import { Identity, IdentityInjectionToken, SignerWithProvider } from './identity/Identity'
+import { createIdentityFromConfig } from './identity/IdentityMapping'
 import { PermissionAssignment, PermissionQuery, toInternalPermissionAssignment, toInternalPermissionQuery } from './permission'
+import { StreamMessage } from './protocol/StreamMessage'
 import { MessageListener, MessageStream } from './subscribe/MessageStream'
 import { ResendOptions, Resends, toInternalResendOptions } from './subscribe/Resends'
 import { Subscriber } from './subscribe/Subscriber'
@@ -57,12 +63,9 @@ import { StreamDefinition } from './types'
 import { map } from './utils/GeneratorUtils'
 import { LoggerFactory } from './utils/LoggerFactory'
 import { addStreamToStorageNode } from './utils/addStreamToStorageNode'
+import { assertCompliantIdentity } from './utils/encryptionCompliance'
 import { pOnce } from './utils/promises'
 import { convertPeerDescriptorToNetworkPeerDescriptor, createTheGraphClient } from './utils/utils'
-import { createIdentityFromConfig } from './identity/IdentityMapping'
-import { assertCompliantIdentity } from './utils/encryptionCompliance'
-import { SponsorshipFactory } from './contracts/SponsorshipFactory'
-import { ProxyNodeFinder } from './ProxyNodeFinder'
 
 // TODO: this type only exists to enable tsdoc to generate proper documentation
 export type SubscribeOptions = StreamDefinition & ExtraSubscribeOptions
@@ -181,6 +184,11 @@ export class StreamrClient {
         return convertStreamMessageToMessage(result)
     }
 
+    async publishRaw(message: StreamMessage, deliveryOptions?: StreamPartDeliveryOptions): Promise<void> {
+        await this.node.broadcast(message, deliveryOptions)
+        this.eventEmitter.emit('messagePublished', message)
+    }
+
     /**
      * Manually updates the encryption key used when publishing messages to a given stream.
      */

packages/sdk/test/end-to-end/publish-subscribe-raw.test.ts

@@ -0,0 +1,52 @@
+import { StreamID, toUserId } from '@streamr/utils'
+import { createTestClient, createTestStream } from '../test-utils/utils'
+import { nextValue } from '../../src/utils/iterators'
+import { EthereumKeyPairIdentity } from '../../src/identity/EthereumKeyPairIdentity'
+import { MessageSigner } from '../../src/signature/MessageSigner'
+import { Wallet } from 'ethers'
+import { MessageID } from '../../src/protocol/MessageID'
+import { ContentType, EncryptionType, SignatureType } from '@streamr/trackerless-network'
+import { StreamMessageType, StreamPermission } from '../../src'
+import { createTestWallet } from '@streamr/test-utils'
+
+describe('publish-subscribe-raw', () => {
+
+    let streamId: StreamID
+    let publisherWallet: Wallet
+
+    beforeEach(async () => {
+        const creatorWallet = await createTestWallet({ gas: true })
+        const creatorClient = createTestClient(creatorWallet.privateKey)
+        const stream = await createTestStream(creatorClient, module)
+        streamId = stream.id
+        publisherWallet = await createTestWallet()
+        stream.grantPermissions({
+            userId: publisherWallet.address.toLowerCase(),
+            permissions: [StreamPermission.PUBLISH]
+        })
+        await creatorClient.destroy()
+    })
+
+    async function createTestMessage() {
+        const messageSigner = new MessageSigner(EthereumKeyPairIdentity.fromPrivateKey(publisherWallet.privateKey))
+        return await messageSigner.createSignedMessage({
+            messageId: new MessageID(streamId, 0, 123456789, 0, toUserId(publisherWallet.address), 'mock-msgChainId'),
+            content: new Uint8Array([1, 2, 3]),
+            contentType: ContentType.BINARY,
+            encryptionType: EncryptionType.NONE,
+            messageType: StreamMessageType.MESSAGE
+        }, SignatureType.ECDSA_SECP256K1_EVM)
+    }
+
+    it('happy path', async () => {
+        const publisher = createTestClient()
+        const subscriber = createTestClient()
+        const subcription = await subscriber.subscribe({ streamId, raw: true })
+        const sentMessage = await createTestMessage()
+        await publisher.publishRaw(sentMessage)
+        const receivedMessage = await nextValue(subcription[Symbol.asyncIterator]())
+        expect(receivedMessage! .streamMessage).toEqual(sentMessage)
+        await publisher.destroy()
+        await subscriber.destroy()
+    })
+})