会话相关修改

src/server/include/commons.js

@@ -102,6 +102,8 @@ exports.REDIS_KEYS = {
    Topic: "topics:" + REDIS_KEY_REPLACER
};
exports.STICK_NUM = 90000000000000;
exports.IM_DB = {
    "P2PMSG": "p2p_messages",
    "MUCMSG": "muc_messages",

src/server/models/messages/messages.js

@@ -10,6 +10,7 @@ let RedisModel = require('./../redis.model.js');
let modelUtil = require('../../util/modelUtil');
var imDb = require('../../repository/mysql/db/im.db');
let log = require('../../util/log.js');
let Sessions = require('../sessions/sessions');
const RedisKey = require('../../include/commons').REDIS_KEYS;
const IMTABLE = require('../../include/commons').IM_DB;
@@ -39,23 +40,49 @@ class Messages extends RedisModel {
    /**
     * 根据sessionId获取对应的会话的信息列表
     * 根据sessionId获取对应的会话的信息列表mysql
     * @param sessionId
     */
    getMessagesBySession(sessionId){
    getMessagesBySession(sessionId,handler){
       let session = new  Sessions();
       session.getSessions(sessionId,function(err,res){
            if(err){
                return;
            }else{
                if(res.length==0){
                    log.warn("session is not found!");
                    return;
                }
                let type = res[0].type;
                let db ="";
                if(type==1){
                    db = IMTABLE.MUCMSG;
                }else if(type==2){
                    db = IMTABLE.P2PMSG;
                }else{
                    db = IMTABLE.GROUPMSG;
                }
                let sql = "select * from "+db+" w where w.session_id = ? limit 0,1000";
                imDb.execQuery({
                    "sql": sessionsql,
                    "args": [sessionId],
                    "handler": function (err, res) {
                        if(err) {
                            log.error("sql:"+sql+"data:sessionId："+sessionId);
                        }else{
                            log.info("getMessagesBySession success by sessionId :"+sessionId);
                        }
                        handler(err,res);
                    }
                })
            }
       })
    }
    /**
     * 分页
     * 根据sessionId获取对应的会话的信息列表
     * @param sessionId
     */
    getMessagesBySessionForPage(sessionId,page,pagesize){
    }
    /**
     * 根据消息ID获取单条消息

src/server/models/sessions/participants.js

@@ -32,9 +32,27 @@ class Participants extends RedisModel {
     * 根据sessionId获取对应的议题的成员信息
     * @param sessionId
     */
    getParticipantsBySessionId(sessionId){
    getParticipantsBySessionId(sessionId,handler){
        let  participant_key =  super.makeRedisKey(RedisKey.Participants,sessionId);
        redis.existsAsync(participant_key).then(function(res){
            if(res){
                redis.zrangeAsync(participant_key,0,-1).then(function(res){
                    handler(res);
                })
            }else{
                let sql ="select w.* from participants w where w.session_id =? ";
                imDb.execQuery({
                    "sql": sql,
                    "args": [sessionId],
                    "handler": function (err, res) {
                        if(err) {
                            log.error("getParticipantsBySessionId is fail error: "+err);
                        }
                        handler(res);
                    }
                });
            }
        })
    }
    /**
@@ -44,16 +62,32 @@ class Participants extends RedisModel {
     */
    existsUser(sessionId,userId,handler){
      let  participant_key =  super.makeRedisKey(RedisKey.Participants,sessionId);
      return redis.zrangeAsync(participant_key,0,-1).then(function(res){
          let exists = false
          for(var j in res){
              var value = res[j];
              if(value==userId){
                  exists  = true;
                  break;
              }
      redis.existsAsync(participant_key).then(function(res){
          if(res){//存在redis中直接从redis校验
              redis.zrangeAsync(participant_key,0,-1).then(function(res){
                  let exists = false
                  for(var j in res){
                      var value = res[j];
                      if(value==userId){
                          exists  = true;
                          break;
                      }
                  }
                  handler(exists);
              })
          }else{//不存在从数据库中获取
              let sql ="select count(1) as count from participants w where w.session_id =? and w.participaint_id = ? ";
              imDb.execQuery({
                  "sql": sql,
                  "args": [sessionId,userId],
                  "handler": function (err, res) {
                      if(err) {
                          log.error("existsUser is fail error: "+err);
                      }
                      handler(res[0].count);
                  }
              });
          }
          handler(exists);
      })
    }

src/server/models/sessions/sessions.js

@@ -14,6 +14,7 @@ let imDb = require('../../repository/mysql/db/im.db');
let log = require('../../util/log.js');
const RedisKeys = require('../../include/commons').REDIS_KEYS;
const IMTABLE = require('../../include/commons').IM_DB;
const Commons = require('../../include/commons')
let mongoose = require('mongoose');
class Sessions extends RedisModel {
@@ -22,13 +23,48 @@ class Sessions extends RedisModel {
    }
    /**
     * 根据sessionId获取对应的Session
     * @param sessionId
     * 获取某个用户的全部session列表
     * @param userId
     * @param handler
     */
    getSessionsById(sessionId){
    getMysqlUserSessions(userId,handler){
        let sql ="select select session_id from participants w where w.participaint_id = ? group by w.session_id";
        let sessionsql = "select id,name,type,create_date from session s where s.id in("+sql+")";
        imDb.execQuery({
            "sql": sessionsql,
            "args": [userId],
            "handler": function (err, res) {
                if(err) {
                    log.error("sql:"+sessionsql+"data:userId："+userId);
                }else{
                    log.info("getMysqlUserSessions success by userId :"+userId);
                }
                handler(err,res);
            }
        });
    }
    /**
     * 获取session单个对象
     * @param sessionId
     * @param handler
     */
    getSessions(sessionId,handler){
        let sessionsql = "select id,name,type,create_date from session s where s.id=?";
        imDb.execQuery({
            "sql": sessionsql,
            "args": [sessionId],
            "handler": function (err, res) {
                if(err) {
                    log.error("sql:"+sessionsql+"data:sessionId："+sessionId);
                }else{
                    log.info("getSessions success by sessionId :"+sessionId);
                }
                handler(err,res);
            }
        });
    }
    /**
     * 根据用户ID获取用户的session列表
     * @param userId
@@ -147,6 +183,7 @@ class Sessions extends RedisModel {
            }
         })
    }
    /**
     * 更新最后一条消息
     * @param session_key rediskey
@@ -159,7 +196,7 @@ class Sessions extends RedisModel {
     * @param name 议题名称
     * @returns {*}
     */
     updateLastContent(session_key,session_type,name,message){
    updateLastContent(session_key,session_type,name,message){
        return redis.hmsetAsync(session_key,
            "create_date", message.timestamp,
            "last_content", message.content,
@@ -171,8 +208,11 @@ class Sessions extends RedisModel {
        );
    }
    /**
     * 保存消息，用于消息发送
     * @param message
     * @param sessionId
     */
    saveMessageBySession(message,sessionId) {
        let self = this;
        let messages = new Messages();
@@ -212,19 +252,6 @@ class Sessions extends RedisModel {
        })
    }
    /**
     * 根据sessionId获取对应的消息
     * @param sessionId
     */
    getSessionMessagesByPage(sessionId,page,pageSize){
    }
    /**
     * 非MUC模式创建会话
     * @param sessionId 会话ID
@@ -279,7 +306,53 @@ class Sessions extends RedisModel {
        });
    }
    /**
     *置顶操作
     */
    stickSession(sessionId,user){
        let user_session_key = super.makeRedisKey(RedisKeys.UsersSessions,user);
        let self = this;
        //取出最大的session
        redis.zrevrangeAsync(user_session_key,0,0).then(function(res){
            //获取该session的时间搓
            redis.zscoreAsync(user_session_key,res).then(function(scoreres){
                let nowtime = new Date().getTime();
                //当前时间搓比redis的时间搓更早证明没有置顶过
                if(scoreres<=nowtime){
                    //初始化置顶
                   redis.zaddAsync(user_session_key, Commons.STICK_NUM,sessionId).then(function(res){
                       log.info("stickSession:"+sessionId+",res:"+res);
                       modelUtil.emitData(self.eventEmitter,{"status":200,"msg":"置顶成功！"});
                   })
                }else{
                    //已有置顶的数据，取出来加1保存回去
                    scoreres =  scoreres+1;
                    redis.zaddAsync(user_session_key, scoreres,sessionId).then(function(){
                        log.info("stickSession:"+sessionId+",res:"+res);
                        modelUtil.emitData(self.eventEmitter,{"status":200,"msg":"置顶成功！"});
                    })
                }
            })
        })
    }
    /**
     *取消置顶操作
     */
    cancelStickSession(sessionId,user){
        let user_session_key = super.makeRedisKey(RedisKeys.UsersSessions,user);
        let participants_key = super.makeRedisKey(RedisKeys.Participants,sessionId);
        let self = this;
        redis.zscoreAsync(participants_key,user).then(function(res){
            if(!res){
                res = new Date().getTime();
            }
            redis.zaddAsync(user_session_key, res,sessionId).then(function(res){
                log.info("cancelStickSession:"+sessionId);
                modelUtil.emitData(self.eventEmitter,{"status":200,"msg":"取消置顶成功！"});
            });
        })
    }
}