<commit_before>/**
 * Created by stitchcula on 2016/9/4.
 */

import mosca from 'mosca'

export default class Mosca {
    constructor(mongo){
        this.mongo=mongo
        var ascoltatore={
            type:'redis',
            redis:require('redis'),
            host:process.env['REDIS_HOST'],
            port:process.env['REDIS_PORT'],
            password:<PASSWORD>['<PASSWORD>'],
            return_buffers:true
        }
        this.settings={
            port:1883,
            backend:ascoltatore,
            persistence:{
                factory: mosca.persistence.Redis,
                host:process.env['REDIS_HOST'],
                port:process.env['REDIS_PORT'],
                password:<PASSWORD>['<PASSWORD>']
            }
        }
    }
    listen(port){
        if(port) this.settings.port=port
        this.server = new mosca.Server(this.settings)
        this.server.on('ready',this.onReady.bind(this))
        this.server.on('published',this.onPublished.bind(this))
        this.server.on("clientConnected",this.onConnected.bind(this))
    }
    async onReady(){
        this.server.authenticate = function(client, username, password, callback) {
            console.log(`user:${username},pwd:${password}`)
            callback(null, true);
        }
        this.server.authorizePublish=function (client, topic, payload, callback) {
            console.log("authorizePublish:"+client.id)
            callback(null,true)
        }
        this.server.authorizeSubscribe=function (client, topic, callback) {
            console.log("authorizeSubscribe:"+client.id)
            callback(null,true)
        }
        console.log("MQTT broker is running at "+new Date())
    }
    async onPublished(packet){
        //Topic->device id or user uin
        //node_id->modules id or user token
        //node_type->modules or USER
        //msg_type->data/notice
        if(packet.topic.split('/')[0]!="$SMTH")
            return
        var {node_id,node_type,msg_type,msg_id,msg_c}=JSON.parse(packet.payload.toString())
        console.log(packet.topic+":"+node_id+
            " SEND "+JSON.stringify({node_type,msg_type,msg_id,msg_c})+
            " at "+new Date())
        if(node_type=="USER"){//User->Edison
            //todo:
            //user's order, save to db
        }else{//Edison->User
            //save to db
            //device's data or notice, if it's notice,user sure response.
            await this.mongo.collection("publish_msg")
                .insertOne({uin:packet.topic,node_id,node_type,msg_type,msg_id,msg_c})
            if(msg_type=="NOTICE"){
                this.server.published({
                    topic:packet.topic+"/NOTICE",
                    payload:packet.payload,
                    qos:1,retain:false
                },function (x) {
                    console.log(x)
                })
            }
        }
    }
    async onConnected(a){
        console.log(a.id)
    }
}<commit_msg>add keeping online list.<commit_after>/**
 * Created by stitchcula on 2016/9/4.
 */

import mosca from 'mosca'

export default class Mosca {
    constructor(mongo){
        this.mongo=mongo
        var ascoltatore={
            type:'redis',
            redis:require('redis'),
            host:process.env['REDIS_HOST'],
            port:process.env['REDIS_PORT'],
            password:process.env['REDIS_AUTH'],
            return_buffers:true
        }
        this.settings={
            port:1883,
            backend:ascoltatore,
            persistence:{
                factory: mosca.persistence.Redis,
                host:process.env['REDIS_HOST'],
                port:process.env['REDIS_PORT'],
                password:process.env['REDIS_AUTH']
            }
        }
    }
    listen(port){
        if(port) this.settings.port=port
        this.server = new mosca.Server(this.settings)
        this.server.on('ready',this.onReady.bind(this))
        this.server.on('published',this.onPublished.bind(this))
        this.server.on("clientDisconnected",this.onDisconnected.bind(this))
    }
    async onReady(){
        this.server.authenticate =async function(client, uin, token, callback) {//online login
            //todo:校验uin和token
            var res=await this.mongo.collection("online_list")
                .insertOne({session:client.id,uin,time:new Date().getTime()})
            console.log(res)
            console.log(`user:${uin},pwd:${token}`)
            callback(null, true);
        }
        this.server.authorizePublish=function (client, topic, payload, callback) {
            console.log("authorizePublish:"+client.id)
            callback(null,true)
        }
        this.server.authorizeSubscribe=function (client, topic, callback) {
            console.log("authorizeSubscribe:"+client.id)
            callback(null,true)
        }
        console.log("MQTT broker is running at "+new Date())
    }
    async onPublished(packet){
        //Topic->device id or user uin
        //node_id->modules id or user token
        //node_type->modules or USER
        //msg_type->data/notice
        if(packet.topic.split('/')[0]!="$SMTH"||packet.topic.split('/')[2])
            return
        var {node_id,node_type,msg_type,msg_id,msg_c}=JSON.parse(packet.payload.toString())
        console.log(packet.topic+":"+node_id+
            " SEND "+JSON.stringify({node_type,msg_type,msg_id,msg_c})+
            " at "+new Date())
        if(node_type=="USER"){//User->Edison
            //todo:
            //user's order, save to db
        }else{//Edison->User
            //save to db
            //device's data or notice, if it's notice,user sure response.
            await this.mongo.collection("publish_msg")
                .insertOne({uin:packet.topic,node_id,node_type,msg_type,msg_id,msg_c})
            if(msg_type=="NOTICE"){
                this.server.publish({
                    topic:packet.topic+"/NOTICE",
                    payload:packet.payload,
                    qos:1,retain:false
                })
            }
        }
    }
    async onDisconnected(client){//offline logout
        var res=await this.mongo.collection("online_list")
            .deleteOne({session:client.id})
        console.log(res)
    }
}
{'fraction_non_alphanumeric': 0.13511780734092002, 'fraction_numerical': 0.0040070524122455525, 'mean_word_length': 1.9145259224661373, 'pattern_counts': {'":': 2, '<': 7, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '52657', 'n_tokens_mistral': 1887, 'n_tokens_neox': 1833, 'n_words': 279}