消息发送整改

src/server/endpoints/v2/application.endpoint.js

@@ -2,10 +2,14 @@
let express = require('express');
let router = express.Router();
let request = require('request');
let http = require('http');
const APIv2 = require('../../include/endpoints').APIv2;
const MODEL_EVENTS = require('../../include/commons').MODEL_EVENTS;
let configFile = require('../../include/commons').CONFIG_FILE;
let config = require('../../resources/config/' + configFile);
let ImDb = require('../../repository/mysql/db/im.db');
let log = require('../../util/log.js');
/**
 * 获取应用角标数。
 *
@@ -16,10 +20,47 @@ const MODEL_EVENTS = require('../../include/commons').MODEL_EVENTS;
 * 参数：
 * uid：用户id
 */
router.get(APIv2.Application.BadgeNo, function (req, res) {
router.get(APIv2.Application.BadgeNo, function (req, respone) {
    let userId = req.query.user_id;
        ImDb.execQuery({
            "sql": "SELECT imei,token from wlyy.wlyy_token WHERE user=?",
            "args": [userId],
            "handler": function (err, result) {
                if (err || result.length == 0) {
                    handler(null, 0);
                    return;
                }
                var options = {
                    hostname: config.wlyyServerConfig.host,
                    port: config.wlyyServerConfig.port,
                    path: '/wlyy/doctor/message/messages',
                    method: 'POST',
                    headers: {
                        'userAgent': '{"token":"' + result[0].token + '","uid":"' + userId + '","imei":"' + result[0].imei + '"}'
                    }
                };
                var req = http.request(options, function (res) {
                    res.setEncoding('utf8');
                    log.info('请求家庭医生平台: http://', options.hostname + ":" + options.port + options.path);
                    res.on('data', function (chunk) {
                        log.info('家庭医生平台返回: ', chunk);
                        var count = 0;
                        if(chunk.status==200){
                            let data =JSON.parse(chunk).data;
                            count = JSON.parse(data.imMsgCount).count+data.system.amount+data.healthIndex.amount+data.sign.amount;
                        }
                        respone.status(200).send({"count":count});
                    });
                });
    res.status(200).send({});
                req.on('error', function (e) {
                    log.error('家庭医生平台接口调用出错: ', e.message);
                    handler(e, null);
                });
                req.end();
            }
        });
});
module.exports = router;

src/server/endpoints/v2/search.endpoint.js

@@ -40,7 +40,7 @@ router.get("/", function (req, res) {
    ControllerUtil.regModelEventHandler(searcher, res);
    if(target == 'all'){
        searcher.searchAll(userId, searcher, keyword);
        searcher.searchAll(userId, keyword);
    } else if(target == 'doctor'){
        searcher.searchDoctors(keyword, page, size)
    } else if(target == 'sessions'){

src/server/models/client/wechat.client.js

@@ -53,16 +53,17 @@ class WechatClient extends RedisModel {
     */
    static sendMessage(message) {
        let patientClient = clientCache.findById(message.to);
        let self = this;
        if (patientClient) {
            this.sendViaWebSocket(patientClient.socket, message);
            WechatClient.sendViaWebSocket(patientClient.socket, message);
        } else {
            log.info("User is not online, user id: ", message.to, ", sending via wechat template message.");
            this.sendViaTemplateMessage(message);
            WechatClient.sendViaTemplateMessage(message);
        }
    };
    sendViaWebSocket(socket, message){
    static sendViaWebSocket(socket, message){
        message.timestamp = ObjectUtil.timestampToLong(message.timestamp);
        socket.emit('message', message);
    }
@@ -72,7 +73,7 @@ class WechatClient extends RedisModel {
     *
     * @param message
     */
    sendViaTemplateMessage(message) {
    static sendViaTemplateMessage(message) {
        function sendWxMessage(openid, name, topic) {
            var replyContent = message.content;
            switch (Number.parseInt(message.contentType)) {

src/server/models/search/mysql.searcher.js

@@ -31,37 +31,38 @@ class MySqlSearcher extends BaseModel {
        async.waterfall([
            function (callback) {
                self.searchDoctors(keyword, 0, 3, function (err, doctors) {
                self.searchDoctors(keyword, 0, 4, function (err, doctors) {
                    if(err) return callback(err, null);
                    let data = [];
                    let datas={};
                    doctors.forEach(function (doctor) {
                        data.push(doctor);
                    });
                    callback(null, data);
                    datas.doctor = data;
                    callback(null, datas);
                });
            },
            function (data, callback) {
                self.searchSessions(userId, keyword, 0, 2, function (err, sessions) {
            function (datas, callback) {
                self.searchSessions(userId, keyword, 0,4, function (err, sessions) {
                    if(err) return callback(err, null);
                    let data = [];
                    sessions.forEach(function (session) {
                        data.push(session);
                    });
                    callback(null, data);
                    datas.groups = data;
                    callback(null, datas);
                });
            },
            function (data, callback) {
                self.searchMessages(userId, keyword, 0, 3, function (err, sessions) {
            function (datas, callback) {
                self.searchMessages(userId, keyword, 0, 4, function (err, sessions) {
                    if(err) return callback(err, null);
                    let data = [];
                    sessions.forEach(function (session) {
                        data.push(session);
                    });
                    callback(null, data);
                    datas.content = data;
                    callback(null, datas);
                });
            }
        ],

src/server/models/sessions/sessions.js

@@ -16,7 +16,6 @@ let MessageRepo = require('../../repository/mysql/message.repo');
let WechatClient = require("../client/wechat.client.js");
let AppClient = require("../client/app.client.js");
let configFile = require('../../include/commons').CONFIG_FILE;
let config = require('../../resources/config/' + configFile);
let redis = RedisClient.redisClient().connection;
@@ -28,7 +27,6 @@ let async = require("async");
const REDIS_KEYS = require('../../include/commons').REDIS_KEYS;
const SESSION_TYPES = require('../../include/commons').SESSION_TYPES;
const STICKY_SESSION_BASE_SCORE = require('../../include/commons').STICKY_SESSION_BASE_SCORE;
const SESSION_BUSINESS_TYPE = require('../../include/commons').SESSION_BUSINESS_TYPE;
class Sessions extends RedisModel {
    constructor() {
@@ -338,69 +336,79 @@ class Sessions extends RedisModel {
                    })
            },
            // 遍历会话
            function (sessionIds, callback) {
            function (sessionIds) {
                let sessionList = [];
                sessionIds.forEach(function (sessionId) {
                    let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
                    let participantsRoleKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipantsRole, sessionId);
                    let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
                    let participantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants,sessionId);
                    redis.multi()
                        .hgetall(sessionKey)                       // 会话实体
                        .hget(participantsRoleKey, userId)         // 用户在此会话中的角色
                        .zscore(sessionParticipantsKey, userId)    // 用户在此会话中最后一次获取未读消息的时间
                        .zrange(participantsKey,0,-1)
                        .execAsync()
                        .then(function (res) {
                            let session = res[0];
                            let role = res[1];
                            let lastFetchTime = res[2];
                            let users = res[3];
                            let sessionName="";
                            let otheruserId ="";
                            if(session.type==SESSION_TYPES.P2P){
                                for(var j in users){
                                    if(users[j]!=userId){
                                        otheruserId = users[j];
                                    }
                                }
                            }
                            if(!role)role =0;
                            if(!lastFetchTime)lastFetchTime=new Date().getTime();
                                // 计算未读消息数
                                let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
                                redis.zcountAsync(messagesByTimestampKey, lastFetchTime, new Date().getTime())
                                    .then(function (count) {
                                        if(!otheruserId)otheruserId=userId;
                                        ParticipantRepo.findNameById(otheruserId, function (err, res) {
                                            if((res&&res.length==0)||session.type!=SESSION_TYPES.P2P){
                                                sessionName = session.name;
                                            }else{
                                                sessionName = res[0].name;
                                            }
                                            sessionList.push({
                                                id: sessionId,
                                                name: sessionName,
                                                create_date: session.create_date,
                                                last_content_type: session.last_content_type,
                                                last_content: session.last_content,
                                                sender_id: session.sender_id,
                                                type: session.type,
                                                sender_name: session.sender_name,
                                                unread_count: count,
                                                business_type: session.business_type,
                                                my_role: role
                                            });
                                            if (sessionId === sessionIds[sessionIds.length - 1]) {
                                                ModelUtil.emitOK(self.eventEmitter, sessionList);
                let functionList =[];
                for(var j = 0 ;j<sessionIds.length;j++){
                    let fun = function(index,callback){
                            if(!callback) {callback =index,index=0};
                            let sessionId = sessionIds[index]
                            let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
                            let participantsRoleKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipantsRole, sessionId);
                            let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
                            let participantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants,sessionId);
                            redis.multi()
                                .hgetall(sessionKey)                       // 会话实体
                                .hget(participantsRoleKey, userId)         // 用户在此会话中的角色
                                .zscore(sessionParticipantsKey, userId)    // 用户在此会话中最后一次获取未读消息的时间
                                .zrange(participantsKey,0,-1)
                                .execAsync()
                                .then(function (res) {
                                    let session = res[0];
                                    let role = res[1];
                                    let lastFetchTime = res[2];
                                    let users = res[3];
                                    let sessionName="";
                                    let otheruserId ="";
                                    if(session.type==SESSION_TYPES.P2P){
                                        for(var j in users){
                                            if(users[j]!=userId){
                                                otheruserId = users[j];
                                            }
                                        }
                                    }
                                    if(!role)role =0;
                                    if(!lastFetchTime)lastFetchTime=new Date().getTime();
                                    // 计算未读消息数
                                    let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
                                    redis.zcountAsync(messagesByTimestampKey, lastFetchTime, new Date().getTime())
                                        .then(function (count) {
                                            if(!otheruserId)otheruserId=userId;
                                            ParticipantRepo.findNameById(otheruserId, function (err, res) {
                                                if((res&&res.length==0)||session.type!=SESSION_TYPES.P2P){
                                                    sessionName = session.name;
                                                }else{
                                                    sessionName = res[0].name;
                                                }
                                                sessionList.push({
                                                    id: sessionId,
                                                    name: sessionName,
                                                    create_date: session.create_date,
                                                    last_content_type: session.last_content_type,
                                                    last_content: session.last_content,
                                                    sender_id: session.sender_id,
                                                    type: session.type,
                                                    sender_name: session.sender_name,
                                                    unread_count: count,
                                                    business_type: session.business_type,
                                                    my_role: role
                                                });
                                                index=(parseInt(index)+1);
                                                if(index==sessionIds.length){
                                                    ModelUtil.emitOK(self.eventEmitter, sessionList);
                                                }else{
                                                    callback(null,index);
                                                }
                                            })
                                        })
                                    })
                                }).catch(function (err) {
                                    logger.error("Get sessions failed: ", ex);
                                    ModelUtil.emitError(self.eventEmitter, "Get sessions failed: " + err);
                                });
                });
                                logger.error("Get sessions failed: ", err);
                            });
                    }
                    functionList.push(fun);
                };
                async.waterfall(functionList);
            }
        ]);
    }

src/server/models/sessions/topics.js

@@ -9,7 +9,6 @@ let SessionRepo = require('../../repository/mysql/session.repo');
let ParticipantRepo = require('../../repository/mysql/participant.repo');
let RedisModel = require('./../redis.model.js');
let ModelUtil = require('../../util/model.util');
let Participants = require("./participants");
let Sessions = require("./sessions");
let log = require('../../util/log.js');
@@ -43,7 +42,6 @@ class Topics extends RedisModel {
            for(var j in res){
                sessionIds.push(res[j].id);
            }
            var test = new Map();
            TopicsRepo.findAllBySessionIdsAndStatus(sessionIds,status,page,pagesize,function(err,res){
                if(err){
                    ModelUtil.emitError(self.eventEmitter, "获取列表失败"+err);
@@ -109,11 +107,22 @@ class Topics extends RedisModel {
                ModelUtil.emitOK(self.eventEmitter, {status:-1,message:"找不到对应的会话记录！"});
            }
            let sessions = new Sessions();
            sessions.getMessages(res[0].session_id,user,startMsgId,endMsgId,page,pagesize,0,function(err,res){
            sessions.getMessages(res[0].session_id,user,startMsgId,endMsgId,page,pagesize,0,function(err,messages){
                if(err){
                    ModelUtil.emitOK(self.eventEmitter, {status:-1,message:err});
                }else{
                    ModelUtil.emitOK(self.eventEmitter, {status:200,data:res});
                    if(messages&&messages.length>0){
                        ParticipantRepo.findAll(res[0].session_id,function (err,users) {
                            if(err){
                                ModelUtil.emitOK(self.eventEmitter, {status:-1,message:err});
                            }else{
                                ModelUtil.emitOK(self.eventEmitter, {status:200,data:{"list":messages,"users":users}});
                            }
                        })
                    }else{
                        ModelUtil.emitOK(self.eventEmitter, {status:200,data:[]});
                    }
                }
            });
        })

src/server/repository/mysql/search.repo.js

@@ -10,11 +10,13 @@ class SearchRepo {
    }
    static searchDoctors(keyword, page, size, handler) {
        if(page<=0)page = 1;
        page = (page-1)*size;
        let sql = "select d.id, d.name, d.sex, d.birthdate, d.avatar, d.level from doctors d where d.name like ? limit ?, ?";
        ImDb.execQuery({
            sql: sql,
            args: ['%' + keyword + '%', page * size, size],
            args: ['%' + keyword + '%',page , size],
            handler: handler
        });
    }
@@ -31,13 +33,14 @@ class SearchRepo {
     * @param handler
     */
    static searchSessions(userId, keyword, page, size, handler) {
        let sql = "select s.id, s.name, s.type, s.create_date, s.business_type from sessions s " +
            "where s.id in (select id from sessions s, participants p where s.id = p.session_id and p.participant_id = ?) " +
        let sql = "select s.id, s.name, s.type, s.create_date, s.business_type," +
            "(SELECT group_concat(d.name) FROM doctors d,participants p1 WHERE p1.participant_id = d.id and p1.session_id = s.id and d.name like ?) as members," +
            "(SELECT group_concat(d.avatar) FROM doctors d,participants p1 WHERE p1.participant_id = d.id and p1.session_id = s.id) as images from sessions s" +
            " where s.id in (select id from sessions s, participants p where s.id = p.session_id and p.participant_id = ?) " +
            "and s.type <> 0 and s.name like ? limit ?, ?";
        ImDb.execQuery({
            sql: sql,
            args: [userId, '%' + keyword + '%', page * size, size],
            args: [ '%' + keyword + '%',userId, '%' + keyword + '%', page * size, size],
            handler: handler
        });
    }
@@ -53,18 +56,21 @@ class SearchRepo {
     */
    static searchMessages(userId, keyword, page, size, handler) {
        let sql = "SELECT * FROM( " +
        "SELECT s.id session_id, s.name session_name, s.type session_type, s.create_date, s.business_type, m.id message_id, m.content message_content " +
        "SELECT s.id session_id, s.name session_name, s.type session_type, s.create_date, s.business_type, count(m.id) count, max(m.content) message_content " +
        "FROM sessions s, participants p, muc_messages m " +
        "WHERE p.participant_id = ? and p.session_id = s.id and s.id = m.session_id and m.content_type = 1 AND m.content LIKE ? " +
        "UNION " +
        "SELECT s.id session_id, s.name session_name, s.type session_type, s.create_date, s.business_type, m.id message_id, m.content message_content " +
        "GROUP BY s.id, s.name, s.type, s.create_date, s.business_type "+
         "UNION " +
        "SELECT s.id session_id, s.name session_name, s.type session_type, s.create_date, s.business_type, count(m.id) count, max(m.content) message_content " +
        "FROM sessions s, participants p, group_messages m " +
        "WHERE p.participant_id = ? and p.session_id = s.id and s.id = m.session_id and m.content_type = 1 AND m.content LIKE ? " +
        "GROUP BY s.id, s.name, s.type, s.create_date, s.business_type "+
        "UNION " +
        "SELECT s.id session_id, s.name session_name, s.type session_type, s.create_date, s.business_type, m.id message_id, m.content message_content " +
        "SELECT s.id session_id, s.name session_name, s.type session_type, s.create_date, s.business_type, count(m.id) count, max(m.content) message_content " +
        "FROM sessions s, participants p, p2p_messages m " +
        "WHERE p.participant_id = ? and p.session_id = s.id and s.id = m.session_id and m.content_type = 1 AND m.content LIKE ? " +
        ") X LIMIT ?, ?";
         "GROUP BY s.id, s.name, s.type, s.create_date, s.business_type "+
         ") X LIMIT ?, ?";
        keyword = '%' + keyword + '%';
        ImDb.execQuery({