Add detectFields implementation to Stream

Author: Tuomas Koponen

src/stream/index.ts

@@ -1,4 +1,4 @@
-import { getEndpointUrl } from '../utils'
+import { getEndpointUrl, FieldDetector } from '../utils'
 import authFetch from '../rest/authFetch'
 
 import StorageNode from './StorageNode'
@@ -125,10 +125,24 @@ export default class Stream {
     }
 
     async detectFields() {
-        return authFetch(
-            getEndpointUrl(this._client.options.restUrl, 'streams', this.id, 'detectFields'),
-            this._client.session,
-        )
+        // Get last message of the stream to be used for field detecting
+        const sub = await this._client.resend({
+            stream: this.id,
+            resend: {
+                last: 1,
+            },
+        })
+        const receivedMsgs = await sub.collect()
+
+        if (receivedMsgs.length > 0) {
+            const lastMessage = receivedMsgs[0]
+            const fd = new FieldDetector(lastMessage)
+            const fields = fd.detect()
+
+            // Save field config back to the stream
+            this.config.fields = fields
+            await this.update()
+        }
     }
 
     async addToStorageNode(address: string) {

src/utils/FieldDetector.js

@@ -0,0 +1,33 @@
+export default class FieldDetector {
+    message = null
+
+    constructor(message) {
+        this.message = message
+    }
+
+    detect() {
+        if (this.message == null) {
+            throw new Error('Invalid message provided to FieldDetector constructor')
+        }
+
+        const content = this.message
+        const fields = []
+
+        Object.keys(content).forEach((key) => {
+            let type
+            if (Array.isArray(content[key])) {
+                type = 'list'
+            } else if ((typeof content[key]) === 'object') {
+                type = 'map'
+            } else {
+                type = typeof content[key]
+            }
+            fields.push({
+                name: key,
+                type,
+            })
+        })
+
+        return fields
+    }
+}

src/utils/index.js

@@ -11,8 +11,9 @@ import pkg from '../../package.json'
 
 import AggregatedError from './AggregatedError'
 import Scaffold from './Scaffold'
+import FieldDetector from './FieldDetector'
 
-export { AggregatedError, Scaffold }
+export { AggregatedError, Scaffold, FieldDetector }
 
 const UUID = uuidv4()
 

test/integration/Stream.test.js

@@ -0,0 +1,107 @@
+import StreamrClient from '../../src'
+import { uid, fakePrivateKey, getPublishTestMessages } from '../utils'
+
+import config from './config'
+
+const createClient = (opts = {}) => new StreamrClient({
+    ...config.clientOptions,
+    auth: {
+        privateKey: fakePrivateKey(),
+    },
+    autoConnect: false,
+    autoDisconnect: false,
+    ...opts,
+})
+
+describe('Stream', () => {
+    let client
+    let stream
+
+    beforeEach(async () => {
+        client = createClient()
+        await client.connect()
+
+        stream = await client.createStream({
+            name: uid('stream-integration-test')
+        })
+    })
+
+    afterEach(async () => {
+        await client.disconnect()
+    })
+
+    describe('detectFields()', () => {
+        it('does detect primitive types', async () => {
+            const msg = {
+                number: 123,
+                boolean: true,
+                object: {
+                    k: 1,
+                    v: 2,
+                },
+                array: [1, 2, 3],
+                string: 'test',
+            }
+            const publishTestMessages = getPublishTestMessages(client, {
+                streamId: stream.id,
+                waitForLast: true,
+                createMessage: () => msg,
+            })
+            await publishTestMessages(1)
+
+            expect(stream.config.fields).toEqual([])
+            await stream.detectFields()
+            expect(stream.config.fields).toEqual([
+                {
+                    name: 'number',
+                    type: 'number',
+                },
+                {
+                    name: 'boolean',
+                    type: 'boolean',
+                },
+                {
+                    name: 'object',
+                    type: 'map',
+                },
+                {
+                    name: 'array',
+                    type: 'list',
+                },
+                {
+                    name: 'string',
+                    type: 'string',
+                },
+            ])
+        })
+
+        it('skips unsupported types', async () => {
+            const msg = {
+                null: null,
+                empty: {},
+                func: () => null,
+                nonexistent: undefined,
+                symbol: Symbol('test'),
+            }
+            const publishTestMessages = getPublishTestMessages(client, {
+                streamId: stream.id,
+                waitForLast: true,
+                createMessage: () => msg,
+            })
+            await publishTestMessages(1)
+
+            expect(stream.config.fields).toEqual([])
+            await stream.detectFields()
+            expect(stream.config.fields).toEqual([
+                {
+                    name: 'null',
+                    type: 'map',
+                },
+                {
+                    name: 'empty',
+                    type: 'map',
+                },
+            ])
+        })
+    })
+})