NET-178: Storage node methods #197

Methods to query+modify storage nodes. Two new entity classes: StreamPart (streamId+partition) and StorageNode (currently just an address, but can be extended later).

Author: teogeb

src/rest/StreamEndpoints.js

@@ -7,6 +7,7 @@ import debugFactory from 'debug'
 import { getEndpointUrl } from '../utils'
 import { validateOptions } from '../stream/utils'
 import Stream from '../stream'
+import StreamPart from '../stream/StreamPart'
 import { isKeyExchangeStream } from '../stream/KeyExchange'
 
 import authFetch from './authFetch'
@@ -202,6 +203,15 @@ export async function getStreamLast(streamObjectOrId) {
     return json
 }
 
+export async function getStreamPartsByStorageNode(address) {
+    const json = await authFetch(getEndpointUrl(this.options.restUrl, 'storageNodes', address, 'streams'), this.session)
+    let result = []
+    json.forEach((stream) => {
+        result = result.concat(StreamPart.fromStream(stream))
+    })
+    return result
+}
+
 export async function publishHttp(streamObjectOrId, data, requestOptions = {}, keepAlive = true) {
     let streamId
     if (streamObjectOrId instanceof Stream) {

src/stream/StorageNode.js

@@ -0,0 +1,9 @@
+export default class StorageNode {
+    constructor(address) {
+        this._address = address
+    }
+
+    getAddress() {
+        return this._address
+    }
+}

src/stream/StreamPart.js

@@ -0,0 +1,22 @@
+export default class StreamPart {
+    constructor(streamId, streamPartition) {
+        this._streamId = streamId
+        this._streamPartition = streamPartition
+    }
+
+    static fromStream({ id, partitions }) {
+        const result = []
+        for (let i = 0; i < partitions; i++) {
+            result.push(new StreamPart(id, i))
+        }
+        return result
+    }
+
+    getStreamId() {
+        return this._streamId
+    }
+
+    getStreamPartition() {
+        return this._streamPartition
+    }
+}

src/stream/index.js

@@ -1,6 +1,8 @@
 import { getEndpointUrl } from '../utils'
 import authFetch from '../rest/authFetch'
 
+import StorageNode from './StorageNode'
+
 export default class Stream {
     constructor(client, props) {
         this._client = client
@@ -109,6 +111,37 @@ export default class Stream {
         )
     }
 
+    async addToStorageNode(address) {
+        return authFetch(
+            getEndpointUrl(this._client.options.restUrl, 'streams', this.id, 'storageNodes'),
+            this._client.session,
+            {
+                method: 'POST',
+                body: JSON.stringify({
+                    address
+                })
+            },
+        )
+    }
+
+    async removeFromStorageNode(address) {
+        return authFetch(
+            getEndpointUrl(this._client.options.restUrl, 'streams', this.id, 'storageNodes', address),
+            this._client.session,
+            {
+                method: 'DELETE'
+            },
+        )
+    }
+
+    async getStorageNodes() {
+        const json = await authFetch(
+            getEndpointUrl(this._client.options.restUrl, 'streams', this.id, 'storageNodes'),
+            this._client.session,
+        )
+        return json.map((item) => new StorageNode(item.storageNodeAddress))
+    }
+
     async publish(...theArgs) {
         return this._client.publish(this.id, ...theArgs)
     }

test/integration/StreamEndpoints.test.js

@@ -229,6 +229,32 @@ function TestStreamEndpoints(getName) {
             expect(await client.getStream(createdStream.id)).toBe(undefined)
         })
     })
+
+    describe.only('Storage node assignment', () => {
+        it('add', async () => {
+            const storageNodeAddress = ethers.Wallet.createRandom().address
+            const stream = await client.createStream()
+            await stream.addToStorageNode(storageNodeAddress)
+            const storageNodes = await stream.getStorageNodes()
+            expect(storageNodes.length).toBe(1)
+            expect(storageNodes[0].getAddress()).toBe(storageNodeAddress)
+            const storedStreamParts = await client.getStreamPartsByStorageNode(storageNodeAddress)
+            expect(storedStreamParts.length).toBe(1)
+            expect(storedStreamParts[0].getStreamId()).toBe(stream.id)
+            expect(storedStreamParts[0].getStreamPartition()).toBe(0)
+        })
+
+        it('remove', async () => {
+            const storageNodeAddress = ethers.Wallet.createRandom().address
+            const stream = await client.createStream()
+            await stream.addToStorageNode(storageNodeAddress)
+            await stream.removeFromStorageNode(storageNodeAddress)
+            const storageNodes = await stream.getStorageNodes()
+            expect(storageNodes).toHaveLength(0)
+            const storedStreamParts = await client.getStreamPartsByStorageNode(storageNodeAddress)
+            expect(storedStreamParts).toHaveLength(0)
+        })
+    })
 }
 
 describe('StreamEndpoints', () => {