Skip to main content

MQTT Message Server and Angular Frontend

Guangzhou, China

Basic Setup​

Msg Server​

git clone https://github.com/mpolinowski/mqtt-msg
cd mqtt-msg
npm install

Database​

docker pull mongo:4.4.14
docker pull mongo-express:1.0.0-alpha.4

docker-compose.yml

version: '3.8'

services:

  mongodb:
    image: mongo:4.4.14
    container_name: mongodb
    restart: unless-stopped
    ports:
      - 27017:27017
    environment:
      MONGO_INITDB_ROOT_USERNAME: root
      MONGO_INITDB_ROOT_PASSWORD: example
    volumes:
      - type: bind
        source: /opt/mongo/db
        target: /data/db
    healthcheck:
      test: echo 'db.runCommand("ping").ok' | mongo mongodb:27017/test --quiet
      interval: 30s
      timeout: 10s
      retries: 3
    networks:
      - mongodb_network

  mongo-express:
    image: mongo-express:1.0.0-alpha.4
    container_name: mongo-express
    restart: unless-stopped
    ports:
      - 8081:8081
    environment:
      ME_CONFIG_MONGODB_ADMINUSERNAME: root
      ME_CONFIG_MONGODB_ADMINPASSWORD: example
      ME_CONFIG_MONGODB_URL: mongodb://root:example@mongodb:27017/
      ME_CONFIG_BASICAUTH_USERNAME: admin
      ME_CONFIG_BASICAUTH_PASSWORD: example
    depends_on:
      - mongodb
    networks:
      - mongodb_network

networks:
  mongodb_network:
    name: mongodb_network

Create directories for data storage:

mkdir -p /opt/mongo/{db,configdb}

Open ports if application runs on a different server and start database:

ufw allow 8081/tcp
ufw allow 27017/tcp
ufw reload
ufw status
docker-compose up -d
docker ps -a

And login to the Mongo-Express web frontend on your server IP and port 8081 with your basicAuth password (s. docker-compose.yml).

Connections​

Connecting to your MQTT Broker​

We will use the mqtt NPM package (npm install mqtt) to connect to the MQTT broker running on an INSTAR WQHD Camera with the following options:

// INSTAR MQTT Broker setup
const {connect} = require('mqtt')
const brokerURL = 'mqtt://192.168.2.115:1883'
const topicPrefix = 'msg/'
const options = {
  // Clean session
  clean: true,
  connectTimeout: 4000,
  // Auth
  clientId: 'mqtt_msg',
  username: 'admin',
  password: 'admin',
}
const client  = connect(brokerURL, options)

We can test the connection by publishing a heartbeat from our Msg client (see app.ts) and reading the message using an MQTT subscription:

initMqttListener() {
        console.log('[INFO] :: Msg server connecting to MQTT broker')

        // MQTT Subscription
        client.on('connect', function() {
            console.log('[INFO] :: Connected to MQTT Broker ' + brokerURL)
            client.subscribe(topicPrefix + '#', function(err: any) {
                if (err) {
                    console.log('[ERROR] :: MQTT error', err)
                } else {
                    client.publish(topicPrefix + 'heartbeat', 'Msg Service is online')
                    console.log('[INFO] :: Topic', topicPrefix + 'heartbeat', 'was updated')
                }
            })
        })

        // Read MQTT Messages
        client.on('message', function(topic: any, message: any) {
            // Read topic/message from buffer
            console.log('[MQTT] ::', topic.toString(), '"' + message.toString() + '"')
            client.end()
        })
    }

Connecting to your MongoDB Instance​

We will use the mongoose NPM package (npm install mongoose) to connect to the database with the following options:

// MongoDB setup
const mongoose = require('mongoose')
const mongoAuth = {
    "auth": {
        "authSource": "admin"
    },
    "user": "root",
    "pass": "example"
}
const mongoURL = 'mongodb://root:example@192.168.2.111:27017/?retryWrites=true&w=majority'

The following code connects to our MongoDB instances and creates a collection mqtt-msg inside the test database:

initDB() {
        connect_db().catch(err => console.log('[ERROR] :: Database error', err));
        async function connect_db() {
            await mongoose.connect(mongoURL, mongoAuth);
          }

        const createCollection = mongoose.model('mqtt-msg', { name: String})

        const test = new createCollection({ name: 'Msg Client'})
        test.save().then(() => console.log('[INFO] :: Msg server connected to database'))
    }

Spinning up the Webserver​

We will use the express NPM package (npm install express) to create a webserver with the following options:

// Express setup
const express = require('express')
const app = express()
const port = process.env.PORT || 8888

To get started we will simply return a timestamp on the webroot / and use the function to trigger our msg client, that so far only establishes connections to our broker and database:

// Run Webfront
app.get('/', (req: any, res: any) => {
    res.json({ server: "running", timestamp: new Date().getTime() })
})


// Webserver
app.listen(port, () => {
    console.log('[INFO] :: Webserver running on Port', port)

    new MsgBot()
})

Test​

Run the app with npm start and check your console output. If everything works you will see the following lines:

[INFO] :: Webserver running on Port 8888
[INFO] :: Msg service is online
[INFO] :: Msg server connecting to MQTT broker
[INFO] :: Connected to MQTT Broker mqtt://192.168.2.115:1883
[INFO] :: Topic msg/heartbeat was updated
[MQTT] :: msg/heartbeat "Msg Service is online"
[INFO] :: Msg server connected to database

You can find the entire app.ts file from this stage below: Establishing Connections.

Writing Incoming Messages to Database​

We can now remove the test from our database initialization:

initDB() {
        connect_db().catch(err => console.log('[ERROR] :: Database error', err));
        async function connect_db() {
            await mongoose.connect(mongoURL, mongoAuth);
          }
    }

And instead move it down into the MQTT function to handle incoming (JSON formatted) MQTT messages and write them into our database:

initMqttListener() {
    console.log('[INFO] :: Msg server connecting to MQTT broker')

    // MQTT Subscription
    client.on('connect', function() {
        console.log('[INFO] :: Connected to MQTT Broker ' + brokerURL)
        client.subscribe(topicPrefix + '#', function(err: any) {
            if (err) {
                console.log('[ERROR] :: MQTT error', err)
            } else {
                client.publish(topicPrefix + 'server', JSON.stringify({ msg: 'Msg Service is online' }))
                console.log('[INFO] :: Topic', topicPrefix + 'server', 'was updated')
            }
        })
    })

    // Read MQTT Messages
    client.on('message', function(topic: any, message: any) {
        // Read topic/message from buffer
        const msg = message.toString()
        console.log('[MQTT] ::', topic.toString(), '"' + msg + '"')
        // Write message to database
        // Define Collection model
        const msgModel = mongoose.model('mqtt-msg', { msg: String, createdOn: Date, topicId: String })
        // Use JSON for messages
        const msgObject = JSON.parse(msg)
        // Add date to each message object
        msgObject.createdOn = new Date()
        // Add ID to each message object
        msgObject.topicId = topic

        new msgModel(msgObject).save().then(() => console.log('[INFO] :: Msg saved to database'))

        client.end()
    })
}

See Writing Messages to Database for full application code.

API & Frontend​

Generating Channel IDs​

To be able to add those messages to an user interface we will have to provide an API. We can start by creating random session IDs for communication channels:

// Return random channel ID
app.get('/channelId', (req: any, res: any) => {
    const {client1, client2} = req.query
    const channelId = topicPrefix + btoa(client1 + '_' + client2 + '_' + new Date().getTime)
    res.json({ channelId })
})

This creates an Express route that returns a unique, randomly generated, ID for each pair of clients:

curl 'http://localhost:8888/channelId?client1=me&client2=you'

{"channelId":"msg/bWVfeW91X2Z1bmN0aW9uIGdldFRpbWUoKSB7IFtuYXRpdmUgY29kZV0gfQ=="}

Angular Frontend​

Scaffold the Angular app:

npm install -g @angular/cli
ng new frontend
cd frontend
ng generate library msg-server

This will create a library for our message server in frontend/projects/msg-server/src/lib that we need to import as MsgServer module in frontend/src/app/app.module.ts:

import { NgModule } from '@angular/core';
import { BrowserModule } from '@angular/platform-browser';
import { MsgServerModule } from 'projects/msg-server/src/public-api';

import { AppRoutingModule } from './app-routing.module';
import { AppComponent } from './app.component';

@NgModule({
  declarations: [
    AppComponent
  ],
  imports: [
    BrowserModule,
    AppRoutingModule,
    MsgServerModule
  ],
  providers: [],
  bootstrap: [AppComponent]
})
export class AppModule { }

Now remove all the placeholder HTML from frontend/src/app/app.component.html and replace it with:

<lib-msg-server></lib-msg-server>

<router-outlet></router-outlet>

Run the app npm start and verify that it is running on http://localhost:4200/ - you should see a simple __msg-server works! __.

We can now replace this HTML with a code snippit from codepen.io. The line we saw above is generated in frontend/projects/msg-server/src/lib/msg-server.component.ts which we can replace with our template code:

import { Component, OnInit } from '@angular/core';

@Component({
  selector: 'lib-msg-server',
  templateUrl: './msg-server.component.html',
  styleUrls: [
    './msg-server.component.html'
  ]
})
export class MsgServerComponent implements OnInit {

  constructor() { }

  ngOnInit(): void {
  }

}

And copy&paste the code from codepen.io:

  • HTML into frontend/projects/msg-server/src/lib/msg-server.component.html
  • CSS into frontend/projects/msg-server/src/lib/msg-server.component.css

And make changes as you see fit to the HTL and CSS.

Connecting the Frontend​

Install the following library:

npm install ngx-mqtt
npm install ngx-mqtt-chat

Import it into frontend/projects/msg-server/src/lib/msg-server.module.ts:

import { CommonModule } from '@angular/common';
import { NgModule } from '@angular/core';
import { FormsModule } from '@angular/forms';

import { NgxMqttChatModule } from 'ngx-mqtt-chat';
import { MsgServerComponent } from './msg-server.component';



@NgModule({
  declarations: [
    MsgServerComponent
  ],
  imports: [
    CommonModule,
    NgxMqttChatModule,
    FormsModule
  ],
  exports: [
    MsgServerComponent
  ]
})

export class MsgServerModule { }

And provide the connection data for your MQTT Broker (the module only seems to support websockets and no authentication ?) in frontend/projects/msg-server/src/lib/msg-server.component.ts:

import { Component, OnInit } from '@angular/core';
import { NgxMqttChatService } from 'ngx-mqtt-chat';

@Component({
  selector: 'lib-msg-server',
  templateUrl: './msg-server.component.html',
  styleUrls: [
    './msg-server.component.css'
  ]
})
export class MsgServerComponent implements OnInit {
  chatView=false;
  constructor(
      private messagingService: NgxMqttChatService
    ) { }

  ngOnInit(): void {
    let mqtt = {
      manageConnectionManually: false,
      hostname: '192.168.2.111',
      port: 1885,
      path: ''
    } as any
    this.messagingService.connect(mqtt),
      this.messagingService.publishToTopic(
        'msg/bWVfeW91X2Z1bmN0aW9uIGdldFRpbWUoKSB7IFtuYXRpdmUgY29kZV0gfQ==',
        { msg:"Client connected..." }
    )
  }

}

Here the messaging service is using the channel ID msg/bWVfeW91X2Z1bmN0aW9uIGdldFRpbWUoKSB7IFtuYXRpdmUgY29kZV0gfQ== created earlier to publish a status message.

Addendum​

Establishing Connections​

// Express setup
const express = require('express')
const app = express()
const port = process.env.PORT || 8888

// MongoDB setup
const mongoose = require('mongoose')
const mongoAuth = {
    "auth": {
        "authSource": "admin"
    },
    "user": "root",
    "pass": "example"
}
const mongoURL = 'mongodb://root:example@192.168.2.111:27017/?retryWrites=true&w=majority'


// INSTAR MQTT Broker setup
const {connect} = require('mqtt')
const brokerURL = 'mqtt://192.168.2.115:1883'
const topicPrefix = 'msg/'
const options = {
  // Clean session
  clean: true,
  connectTimeout: 4000,
  // Auth
  clientId: 'mqtt_msg',
  username: 'admin',
  password: 'admin',
}
const client  = connect(brokerURL, options)


// Msg Bot
export class MsgBot {
    constructor() {
        console.log('[INFO] :: Msg service is online')
        this.initMqttListener()
        this.initDB()
    }

    initDB() {
        connect_db().catch(err => console.log('[ERROR] :: Database error', err));
        async function connect_db() {
            await mongoose.connect(mongoURL, mongoAuth);
          }

        const createCollection = mongoose.model('mqtt-msg', { name: String})

        const test = new createCollection({ name: 'Msg Client'})
        test.save().then(() => console.log('[INFO] :: Msg server connected to database'))
    }

    initMqttListener() {
        console.log('[INFO] :: Msg server connecting to MQTT broker')

        // MQTT Subscription
        client.on('connect', function() {
            console.log('[INFO] :: Connected to MQTT Broker ' + brokerURL)
            client.subscribe(topicPrefix + '#', function(err: any) {
                if (err) {
                    console.log('[ERROR] :: MQTT error', err)
                } else {
                    client.publish(topicPrefix + 'heartbeat', 'Msg Service is online')
                    console.log('[INFO] :: Topic', topicPrefix + 'heartbeat', 'was updated')
                }
            })
        })

        // Read MQTT Messages
        client.on('message', function(topic: any, message: any) {
            // Read topic/message from buffer
            console.log('[MQTT] ::', topic.toString(), '"' + message.toString() + '"')
            client.end()
        })
    }
}


// Run Webfront
app.get('/', (req: any, res: any) => {
    res.json({ server: "running", timestamp: new Date().getTime() })
})


// Webserver
app.listen(port, () => {
    console.log('[INFO] :: Webserver running on Port', port)

    new MsgBot()
})

Writing Messages to Database​

// Express setup
const express = require('express')
const app = express()
const port = process.env.PORT || 8888

// MongoDB setup
const mongoose = require('mongoose')
const mongoAuth = {
    "auth": {
        "authSource": "admin"
    },
    "user": "root",
    "pass": "example"
}
const mongoURL = 'mongodb://root:example@192.168.2.111:27017/?retryWrites=true&w=majority'


// INSTAR MQTT Broker setup
const {connect} = require('mqtt')
const brokerURL = 'mqtt://192.168.2.115:1883'
const topicPrefix = 'msg/'
const options = {
  // Clean session
  clean: true,
  connectTimeout: 4000,
  // Auth
  clientId: 'msg_server',
  username: 'admin',
  password: 'admin',
}
const client  = connect(brokerURL, options)


// Msg Bot
export class MsgBot {

    constructor() {
        console.log('[INFO] :: Msg service is online')
        this.initMqttListener()
        this.initDB()
    }

    initDB() {
        connect_db().catch(err => console.log('[ERROR] :: Database error', err));
        async function connect_db() {
            await mongoose.connect(mongoURL, mongoAuth);
          }
    }

    initMqttListener() {
        console.log('[INFO] :: Msg server connecting to MQTT broker')

        // MQTT Subscription
        client.on('connect', function() {
            console.log('[INFO] :: Connected to MQTT Broker ' + brokerURL)
            client.subscribe(topicPrefix + '#', function(err: any) {
                if (err) {
                    console.log('[ERROR] :: MQTT error', err)
                } else {
                    client.publish(topicPrefix + 'server', JSON.stringify({ msg: 'Msg Service is online' }))
                    console.log('[INFO] :: Topic', topicPrefix + 'server', 'was updated')
                }
            })
        })

        // Read MQTT Messages
        client.on('message', function(topic: any, message: any) {
            // Read topic/message from buffer
            const msg = message.toString()
            console.log('[MQTT] ::', topic.toString(), '"' + msg + '"')
            // Write message to database
            // Define Collection model
            const msgModel = mongoose.model('mqtt-msg', { msg: String, createdOn: Date, topicId: String })
            // Use JSON for messages
            const msgObject = JSON.parse(msg)
            // Add date to each message object
            msgObject.createdOn = new Date()
            // Add ID to each message object
            msgObject.topicId = topic

            new msgModel(msgObject).save().then(() => console.log('[INFO] :: Msg saved to database'))

            client.end()
        })
    }
}


// Run Webfront
app.get('/', (req: any, res: any) => {
    res.json({ server: "running", timestamp: new Date().getTime() })
})


// Webserver
app.listen(port, () => {
    console.log('[INFO] :: Webserver running on Port', port)

    new MsgBot()
})