moscajs / aedes-persistence Goto Github PK

'use strict'

var redis = require('mqemitter-redis')

var mq = redis({
  port: 6379,
  host: '127.0.0.1'
})

var persistence = require('aedes-persistence-redis')({
    port: 6379, // Redis port
    host: '127.0.0.1' // Redis host
})
var aedes = require('aedes')({
    //mq: mq,
    //persistence: persistence,
    heartbeatInterval: 1000
})
var server = require('net').createServer(aedes.handle)

var port = 1886

server.listen(port, function() {
    console.log('server listening on port', port)
})

aedes.on('clientError', function(client, err) {
    console.log('client error', client.id, err.message, err.stack)
})

aedes.on('client', function(client) {
    console.log('new client', client.id)
})

var mqtt = require('mqtt')
var client = mqtt.connect({
    host: 'localhost',
    port: 1886,
    clientId: 'client1',
    clean: false
})

client.on('connect', function() {
    client.subscribe('test', { qos: 1 })
})

client.on('message', function(topic, message) {
    // message is Buffer
    console.log(message.toString())
        //client.end()
})

var mqtt = require('mqtt')
var client = mqtt.connect({
    host: 'localhost',
    port: 1886,
    clientId: 'client2',
    clean: false
})

client.on('connect', function() {
    for (var i = 0; i < 10; i++) {
        client.publish('test', 'Hello mqtt ' + i, { qos: 1 })
        console.log('sent')
    }
})

client.on('message', function(topic, message) {
    console.log(message.toString())
})

        var packets = [];
        var stream = broker.persistence.createRetainedStreamCombi ([tag]);

        stream.on('end', function(){
            if (packets.length !== 0) {
                var retained_value = JSON.parse(packets[0].payload).value;
                // Compare retained_value 
            }
        });
        stream.pipe(through.obj(function sendRetained (packet, enc, cb_through) {
            packets.push(packet);
            cb_through();
        }));

 const UserService = require('./service/UserService');
 global.httpContext = require('express-http-context');
 const uuidv4 = require('uuid/v4');
 require('./utils/logger');
 const aedesPersistenceMongoDB = require('aedes-persistence-mongodb');
const aPMDB = new aedesPersistenceMongoDB({
	url: process.env.MQTT_AEDES_DB_URIS, //'mongodb://127.0.0.1/aedes-test', // Optional when you pass db object
	// Optional ttl settings
	ttl: {
		packets: 300, // Number of seconds
		subscriptions: 300
	}
});
let mqmongo = require('mqemitter-mongodb');
let emitter = mqmongo({
	url: process.env.MQTT_AEDES_DB_URIS
});
var aedes = require('aedes')({
	persistence: aPMDB,
	mq: emitter
});
var server = require('net').createServer(aedes.handle);
var port = 1883;
server.listen(port, function() {
	console.log('server listening on port', port);
});
aedes.on('published', function(packet, client) {
	console.log('Published', packet);
	console.log('Client', client);
});
// fired when a client connects
aedes.on('clientConnected', function(client) {
	console.log('Client Connected:', client);
});
aedes.on('client', function(client) {
	console.log('on Client :', client.id, client.broker.id);
});
// fired when a client disconnects
aedes.on('clientDisconnected', function(client) {
	console.log('Client Disconnected:', client.id);
});
aedes.on('clientError', function(client) {
	console.log('Client Disconnected:', client.id);
});
aedes.on('clientError', function(client, err) {
	console.log('client error', client.id, err.message, err.stack);
});
aedes.on('connectionError', function(client, err) {
	console.log('client error', client, err.message, err.stack);
});
aedes.on('subscribe', function(subscriptions, client) {
	if (client) {
		console.log('subscribe from client', subscriptions, client.id);
	}
});
// Accepts the connection if the username and password are valid
var authenticate = async function(client, username, password, callback) {
	var error = new Error('Auth error');
	error.returnCode = 1;
	const response = await UserService.verifyToken(password.toString());
	if (response.succeed == true) {
		// console.log(result);
		if (response.result.roles.includes('Partner') == true) {
			client.user = response.result.partyId;
			callback(null, true);
		} else {
			callback(error, null);
		}
	} else {
		callback(error, null);
	}
};
// In this case the client authorized as alice can publish to /users/alice taking
// the username from the topic and verifing it is the same of the authorized user
var authorizePublish = function(client, packet, callback) {
	var error = new Error('Auth error');
	error.returnCode = 1;
	return callback(error, null);
};
// In this case the client authorized as alice can subscribe to /users/alice taking
// the username from the topic and verifing it is the same of the authorized user
var authorizeSubscribe = function(client, sub, callback) {
	if (sub.topic.split('/')[1] != client.user) {
		return callback(new Error('wrong topic'));
	}
	callback(null, sub);
};
aedes.authenticate = authenticate;
aedes.authorizePublish = authorizePublish;
aedes.authorizeSubscribe = authorizeSubscribe;
aedes.publish({
	cmd: 'publish',
	qos: 2,
	topic: 'private/1000127',
	payload: new Buffer.from('hello daniel 123456!'),
	retain: true
});

  url = require('url');
// Parse
var mqtt_url = url.parse(
  process.env.CLOUDAMQP_MQTT_URL ||
    'mqtt://1000127:eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VySWQiOjUyNywicGFydHlJZCI6MTAwMDEyNywiaWF0IjoxNTcwODc0NjE5LCJleHAiOjE1NzM0NjY2MTl9.IhvRESrN_oH5fRyGusrC-VVWlUvLil1FmQ7If3-pDMU@192.168.3.110:1883'
);
var auth = (mqtt_url.auth || ':').split(':');
var url = 'mqtt://' + mqtt_url.host;

//username: auth[0] + ":" + auth[0] if you are on a shared instance
var options = {
  port: mqtt_url.port,
  clientId: 'mqttjs_2',
  username: auth[0],
  password: auth[1],
  retain: true,
  clean: false
};
console.log('Starting!!!!');

// Create a client connection
var client = mqtt.connect(url, options);

client.on('connect', function() {
  // When connected
  console.log('connected!!');
});
client.subscribe('private/1000127', { qos: 2 }, function(err) {
  if (!err) {
    // when a message arrives, do something with it
    console.log('subscribed!!');
  } else {
    console.log('error on subscribe', err);
    client.end();
  }
});
client.on('message', function(topic, message, packet) {
  console.log("Received '" + message + "' on '" + topic + "'");
});
client.on('reconnect', () => {
  console.log('Client reconnected');
});
 client.on('message', function(topic, message, packet) {
 	console.log("Received '" + message + "' on '" + topic + "'");
 });
client.on('disconnect', result => {
  console.log('disconnect', result);
});

client.on('error', function(error) {
  console.log("Can't connect" + error);
});
client.on('suback', function(error) {
  console.log("Can't suback" + error);
});