Merge pull request #5 from FlowFuse/first-tests

Auto start and fix docker install tag

Author: hardillb

docker/Dockerfile

@@ -2,13 +2,12 @@ FROM node:20-alpine
 
 ARG REGISTRY
 ARG REGISTRY_TOKEN
-ARG TAG
+ARG TAG=latest
 RUN if [[ ! -z "$REGISTRY_TOKEN" ]]; then echo "//$REGISTRY/:_authToken=$REGISTRY_TOKEN" >> ~/.npmrc ; fi
 RUN if [[ ! -z "$REGISTRY" ]] ; then npm config set @flowfuse:registry "https://$REGISTRY"; fi
 
 WORKDIR /usr/src/mqtt-schema-agent
-COPY package.json .
-RUN npm install --production --no-audit --no-fund
+RUN npm install --production --no-audit --no-fund @flowfuse/mqtt-schema-agent@$TAG
 
 LABEL org.label-schema.name="FlowFuse MQTT Schema Agent" \
     org.label-schema.url="https://flowfuse.com" \

lib/agent.js

@@ -29,6 +29,17 @@ class Agent {
             this.creds = Object.assign(this.creds, { reconnectPeriod: 0 })
             // console.log(this.creds)
 
+            if (this.creds.ssl) {
+                switch (this.creds.protocol) {
+                case 'mqtt:':
+                    this.creds.protocol = 'mqtts:'
+                    break
+                case 'ws:':
+                    this.creds.protocol = 'wss:'
+                    break
+                }
+            }
+
             const options = {
                 protocol: this.creds.protocol,
                 host: this.creds.host,
@@ -39,13 +50,17 @@ class Agent {
                 password: this.creds.credentials.password
             }
 
+            if (!this.creds.verifySSL) {
+                options.rejectUnauthorized = false
+            }
+
             // console.log(options)
 
             // this.client = mqtt.connect(`${this.creds.protocol}//${this.creds.hostname}:${this.creds.port}`, this.creds)
             this.client = mqtt.connect(options)
             this.client.on('connect', function () {
                 agent.connected = true
-                // console.log('connected')
+                console.log(`connected to ${options.host}:${options.port} as ${options.clientId}`)
                 agent.error = null
                 agent.client.subscribe('#')
             })
@@ -73,8 +88,8 @@ class Agent {
                 // console.log(packet)
             })
             this.client.on('error', function (error) {
-                // console.log('error', error.code)
-                // console.log(error)
+                console.log('error', error.code)
+                console.log(error)
                 switch (error.code) {
                 case 'ECONNREFUSED': // connection refused
                 case 'ENOTFOUND': // DNS lookup failed
@@ -102,9 +117,34 @@ class Agent {
                 }
                 // console.log(`agent.error set to: ${agent.error}`)
             })
-            this.client.on('message', function (topic) {
+            this.client.on('message', function (topic, payload) {
                 // console.log(topic)
-                agent.topics[topic] = Date.now()
+                let type = 'unknown'
+                try {
+                    const stringPayload = payload.toString('utf8')
+                    type = 'string'
+                    switch (stringPayload.charAt(0)) {
+                    case '<':
+                        // probably XML
+                        type = 'xml'
+                        break
+                    case '{':
+                    case '[':
+                        // probably JSON
+                        try {
+                            JSON.parse(stringPayload)
+                        } catch (err) {
+
+                        }
+                        type = 'json'
+                    }
+                } catch (err) {
+
+                }
+                agent.topics[topic] = {
+                    timestamp: Date.now(),
+                    type
+                }
             })
         } catch (err) {
             console.log(err)

lib/api.js

@@ -44,6 +44,7 @@ class API {
 
         // default to running
         this.agent = new Agent(this.options)
+        this.agent.start()
     }
 
     async stop () {