客服自动发送消息功能实现

jim-common/src/main/java/org/jim/common/ImConst.java

@@ -57,6 +57,8 @@ public interface ImConst
 
 	public static final String IM_CHAT_LOGIN_SERVICE_PROCESSOR = "im_chat_login_service_processor";
 
+	public static final String  AUTO_MESSAGE_PROCESSOR = "auto_message_processor";
+
 	public static final String SEPARATOR = ":";
 
 	public static final String ADMIN = "ADMIN";

jim-common/src/main/java/org/jim/common/packets/Group.java

@@ -6,10 +6,6 @@ package org.jim.common.packets;
 import java.util.List;
 
 /**
- * 群组是临时性的; 客户/游客每一次 和 客服 聊天 都会创建一个在线群组
- * 但是有例外: 当客户去与认领该客户的客服聊天时,该客服不在线,则创建离线群组
- * 离线群组不删除,在线群组删除(每创建一次群组,都会创建一次会话,群组id 就是 会话id )
- * 群组主键id 和 会话主键id 一致
  * 版本: [1.0]
  * 功能说明: 
  * 作者: WChao 创建时间: 2017年9月21日 下午1:54:04

jim-server/src/main/java/org/jim/server/command/command.properties

@@ -5,7 +5,7 @@
 #\u9274\u6743\u8BF7\u6C42\u5904\u7406\u5668
 3 = org.jim.server.command.handler.AuthReqHandler
 #\u804A\u5929\u8BF7\u6C42\u5904\u7406\u5668
-11 = org.jim.server.command.handler.ChatReqHandler,org.jim.server.command.handler.processor.chat.DefaultAsyncChatMessageProcessor,com.cn.processor.IMChatAsyncChatMessageProcessor
+11 = org.jim.server.command.handler.ChatReqHandler,org.jim.server.command.handler.processor.chat.DefaultAsyncChatMessageProcessor,com.cn.processor.IMChatAsyncChatMessageProcessor,org.jim.server.command.handler.processor.chat.AutoMessageProcessor
 #\u5173\u95ED\u3001\u9000\u51FA\u8BF7\u6C42\u5904\u7406\u5668
 14 = org.jim.server.command.handler.CloseReqHandler
 #\u63E1\u624B\u8BF7\u6C42\u5904\u7406\u5668(TCP\u534F\u8BAE\u63E1\u624B\u5904\u7406\u5668,WS\u534F\u8BAE\u63E1\u624B\u5904\u7406\u5668)

jim-server/src/main/java/org/jim/server/command/handler/LoginReqHandler.java

@@ -16,8 +16,10 @@ import org.jim.common.packets.User;
 import org.jim.common.protocol.IProtocol;
 import org.jim.common.utils.ImKit;
 import org.jim.common.utils.JsonKit;
+import org.jim.server.ImServerGroupContext;
 import org.jim.server.command.AbstractCmdHandler;
 import org.jim.server.command.CommandManager;
+import org.jim.server.command.handler.processor.chat.AutoMsgQueueRunnable;
 import org.jim.server.command.handler.processor.login.LoginCmdProcessor;
 import org.jim.server.enums.VisitTypeEnum;
 import org.slf4j.Logger;
@@ -89,22 +91,48 @@ public class LoginReqHandler extends AbstractCmdHandler {
 			return null;
 		}
 		String userId = user.getId();
+		//判断是什么协议
 		IProtocol protocol = ImKit.protocol(null, channelContext);
 		String terminal = protocol == null ? "" : protocol.name();
 		user.setTerminal(terminal);
 		imSessionContext.getClient().setUser(user);
+		//绑定用户
 		ImAio.bindUser(channelContext,userId,imConfig.getMessageHelper().getBindListener());
-		//初始化绑定或者解绑群组;
+		//初始化绑定,并且解绑与当前用户绑定群组不一致的群组;
 		bindUnbindGroup(channelContext, user, user.getGroups());
 		//判断当前用户 是否具有 查看所有客服权限 如果有 绑定群组
 		hasViewAllServiceAccountAuthority(channelContext,user);
 		loginServiceHandler.onSuccess(channelContext);
-//		loginRespBody.clear();
+
+		if( !VisitTypeEnum.SERVICEACCOUNT.getKey().equals(user.getVisitType()) ){
+			//自动发送客服设置的消息 (在存在群组时 => 相当于客户或游客登录后)
+			sendServiceAccountAutoMsg(user);
+		}
+		//loginRespBody.clear();
 		ImPacket loginRespPacket = new ImPacket(Command.COMMAND_LOGIN_RESP, loginRespBody.toByte());
 		return loginRespPacket;
 	}
 
 	/**
+	 * 自动发送客服设置的消息 (在存在群组时 => 相当于客户或游客登录后)
+	 * @param user
+	 * @return {@link }
+	 * @author Darren
+	 * @date 2020/2/11 9:49
+	 */
+	private void sendServiceAccountAutoMsg(User user) {
+		log.info("执行自动发送客服设置的消息...");
+		ImServerGroupContext groupContext = (ImServerGroupContext)imConfig.getGroupContext();
+		AutoMsgQueueRunnable autoMsgQueueRunnable = new AutoMsgQueueRunnable(groupContext.getTimExecutor());
+		List<Group> groups = user.getGroups();
+		for (Group group : groups) {
+			autoMsgQueueRunnable.addMsg(group);
+			autoMsgQueueRunnable.getExecutor().execute(autoMsgQueueRunnable);
+		}
+
+	}
+
+	/**
 	 * 为 该群组 绑定 具有查看全部客服权限的 所有客服
 	 * @param
 	 * @return {@link }
@@ -167,7 +195,7 @@ public class LoginReqHandler extends AbstractCmdHandler {
 	}
 
 	/**
-	 * 初始化绑定,并且解绑与绑定群组不一致的群组;
+	 * 初始化绑定,并且解绑与当前用户绑定群组不一致的群组;
 	 */
 	public void bindUnbindGroup(ChannelContext channelContext ,  User user, List<Group> groups)throws Exception{
 		String userId = user.getId();

jim-server/src/main/java/org/jim/server/command/handler/processor/chat/AutoMessageProcessor.java

@@ -0,0 +1,148 @@
+package org.jim.server.command.handler.processor.chat;
+
+import com.alibaba.fastjson.JSONObject;
+import org.apache.commons.collections4.CollectionUtils;
+import org.jim.common.ImAio;
+import org.jim.common.ImConst;
+import org.jim.common.ImPacket;
+import org.jim.common.packets.*;
+import org.jim.server.command.CommandManager;
+import org.jim.server.command.handler.ChatReqHandler;
+import org.jim.server.command.handler.processor.CmdProcessor;
+import org.jim.server.model.AutoReplyMessage;
+import org.jim.server.model.ServiceAccountRoleDepartmentMiddle;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.tio.core.ChannelContext;
+import org.tio.utils.lock.SetWithLock;
+
+import java.util.*;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * 自动消息处理器
+ * @author Darren
+ * @date 2020/2/11 10:07
+ */
+public class AutoMessageProcessor implements CmdProcessor, ImConst {
+
+    private static Logger log = LoggerFactory.getLogger(AutoMessageProcessor.class);
+
+    /**
+     * 自动消息处理方法
+     */
+    public void handler(Group group) {
+        log.info("执行自动消息处理方法...");
+        String group_id = group.getGroup_id();
+        String accountRoleDepartmentId = group.getServiceAccountRoleDepartmentId();
+        ServiceAccountRoleDepartmentMiddle accountRoleDepartment
+                = ServiceAccountRoleDepartmentMiddle.dao.findById(accountRoleDepartmentId);
+        if( null != accountRoleDepartment ){
+            String serviceAccountId = accountRoleDepartment.getStr("service_account_id");
+            //获取 sw_auto_reply_message 中的数据
+            List<AutoReplyMessage> list
+                    = AutoReplyMessage.dao.find(
+                    this.createSql(group.getCompanyId(), group.getDepartmentId(), serviceAccountId)
+            );
+            if(CollectionUtils.isNotEmpty(list)){
+                //已经睡眠的时间
+                int alreadySleepTime = 0;
+                for(AutoReplyMessage message : list) {
+                    //需要睡眠几秒 单位 s
+                    Integer needSleepTime = message.getInt("sendtime");
+                    needSleepTime -= alreadySleepTime;
+                    alreadySleepTime += needSleepTime;
+                    try {
+                        Thread.sleep(needSleepTime * 1000);
+                    } catch (InterruptedException e) {
+                        log.error("执行自动消息处理方法时,发生错误...");
+                        e.printStackTrace();
+                    }
+                    //聊天数据包
+                    ImPacket chatPacket = new ImPacket(
+                            Command.COMMAND_CHAT_REQ
+                            , new RespBody(
+                                    Command.COMMAND_CHAT_REQ
+                                    ,createChatBody(message, accountRoleDepartmentId, group_id)
+                            ).toByte()
+                    );
+                    //设置同步序列号;
+                    chatPacket.setSynSeq(0);
+                    SetWithLock<ChannelContext> channelContexts = ImAio.getChannelContextsByUserId(accountRoleDepartmentId);
+                    if( null != channelContexts && channelContexts.size() > 0 ){
+                        //获取读锁
+                        ReentrantReadWriteLock.ReadLock readLock = channelContexts.getLock().readLock();
+                        //加锁
+                        readLock.lock();
+                        try {
+                            Set<ChannelContext> channels= channelContexts.getObj();
+                            for (ChannelContext channelContext : channels) {
+                                ChatReqHandler chatReqHandler = CommandManager.getCommand(Command.COMMAND_CHAT_REQ, ChatReqHandler.class);
+                                chatReqHandler.handler(chatPacket,channelContext);
+                            }
+                        } catch (Exception e) {
+                            log.error(e.toString(),e);
+                        } finally {
+                            //解锁
+                            readLock.unlock();
+                        }
+                    }
+                }
+            }
+        }
+
+    }
+
+    /**
+     * 创建聊天体
+     * @param message
+     * @param accountRoleDepartmentId
+     * @param group_id
+     * @return {@link {@link ChatBody}}
+     * @author Darren
+     * @date 2020/2/11 15:01
+     */
+    private ChatBody createChatBody(AutoReplyMessage message, String accountRoleDepartmentId, String group_id){
+        ChatBody chatBody = ChatBody.newBuilder().build();
+        chatBody.setChatType(ChatType.CHAT_TYPE_PUBLIC.getNumber());
+        chatBody.setContent(message.getStr("content"));
+        chatBody.setFrom(accountRoleDepartmentId);
+        chatBody.setCreateTime( System.currentTimeMillis() );
+        chatBody.setGroup_id(group_id);
+        chatBody.setMsgType(MsgType.MSG_TYPE_TEXT.getNumber());
+        chatBody.setId(UUID.randomUUID().toString().replace("-",""));
+        Map<String, Object> map = new HashMap<>();
+        //发送的消息类型 0:客户/游客发送的消息 1:客服发送的消息 2:客服自动发送的消息
+        map.put("sendType",2);
+        chatBody.setExtras(new JSONObject(map));
+        return chatBody;
+    }
+
+    /**
+     * 创建查询sql
+     * @param companyId
+     * @param departmentId
+     * @param serviceAccountId
+     * @return {@link {@link String}}
+     * @author Darren
+     * @date 2020/2/11 15:01
+     */
+    private String createSql(String companyId, String departmentId, String serviceAccountId){
+        StringBuilder sql = new StringBuilder();
+        sql.append(" select * from sw_auto_reply_message where state = 1 and company_id = " + companyId)
+                .append(" and department_id = " + departmentId)
+                .append(" and service_account_id = " + serviceAccountId)
+                .append(" order by sendtime asc ");
+        return sql.toString();
+    }
+
+    @Override
+    public boolean isProtocol(ChannelContext channelContext) {
+        return true;
+    }
+
+    @Override
+    public String name() {
+        return AUTO_MESSAGE_PROCESSOR;
+    }
+}

jim-server/src/main/java/org/jim/server/command/handler/processor/chat/AutoMsgQueueRunnable.java

@@ -0,0 +1,56 @@
+package org.jim.server.command.handler.processor.chat;
+
+import org.jim.common.ImConst;
+import org.jim.common.packets.Command;
+import org.jim.common.packets.Group;
+import org.jim.server.command.CommandManager;
+import org.jim.server.command.handler.ChatReqHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.tio.core.ChannelContext;
+import org.tio.utils.thread.pool.AbstractQueueRunnable;
+
+import java.util.concurrent.Executor;
+
+/**
+ * @author Darren
+ * @date 2020/2/11 9:31
+ */
+public class AutoMsgQueueRunnable extends AbstractQueueRunnable<Group> {
+
+    private static Logger log = LoggerFactory.getLogger(AutoMsgQueueRunnable.class);
+
+    private AutoMessageProcessor autoMessageProcessor;
+
+    public AutoMsgQueueRunnable(Executor executor) {
+        super(executor);
+        ChatReqHandler chatReqHandler = CommandManager.getCommand(Command.COMMAND_CHAT_REQ,ChatReqHandler.class);
+        autoMessageProcessor = chatReqHandler.getProcessor(ImConst.AUTO_MESSAGE_PROCESSOR,AutoMessageProcessor.class).get(0);
+    }
+
+    @Override
+    public boolean addMsg(Group group) {
+        if(this.isCanceled()){
+            log.error("任务已经取消，添加到消息队列失败");
+            return false;
+        }
+        return msgQueue.add(group);
+    }
+
+    @Override
+    public void runTask() {
+        int size = msgQueue.size();
+        if(size < 0){
+            return ;
+        }
+        Group group = null;
+        //ConcurrentLinkedQueue 中 poll() 方法:
+        //检索并删除此队列的头部，如果此队列为空，则返回 null
+        while( (group = msgQueue.poll()) != null ){
+            if(autoMessageProcessor != null){
+                autoMessageProcessor.handler(group);
+            }
+        }
+    }
+
+}

jim-server/src/main/java/org/jim/server/model/AutoReplyMessage.java

@@ -0,0 +1,11 @@
+package org.jim.server.model;
+
+import com.jfinal.plugin.activerecord.Model;
+
+/**
+ * @author Darren
+ * @date 2020/2/11 11:28
+ */
+public class AutoReplyMessage extends Model<AutoReplyMessage> {
+    public static final AutoReplyMessage dao = new AutoReplyMessage();
+}

server-chat/src/main/java/com/cn/config/PropertyDataBaseConfigBuilder.java

@@ -35,17 +35,19 @@ public class PropertyDataBaseConfigBuilder {
 
         ActiveRecordPlugin arp = new ActiveRecordPlugin(druidPlugin);
         //设置实体 表 关系映射
+        arp.addMapping("sw_group","group_id", ChatGroup.class);
         arp.addMapping("sw_conversation","id", Conversation.class);
         arp.addMapping("sw_conversation_record","id", ConversationRecord.class);
-        arp.addMapping("sw_service_account","id", ServiceAccount.class);
+        arp.addMapping("sw_customer_department_middle","id", CustomerDepartmentMiddle.class);
         arp.addMapping("sw_role_department_middle","id", RoleDepartmentMiddle.class);
+        arp.addMapping("sw_service_account","id", ServiceAccount.class);
         arp.addMapping(
                 "sw_service_account_role_department_middle"
                 ,"id"
                 , ServiceAccountRoleDepartmentMiddle.class
         );
-        arp.addMapping("sw_group","group_id", ChatGroup.class);
-
+        arp.addMapping("sw_visitor_department_middle","id",VisitorDepartmentMiddle.class);
+        arp.addMapping("sw_auto_reply_message","id",AutoReplyMessage.class);
 
         // 与 jfinal web 环境唯一的不同是要手动调用一次相关插件的start()方法
         druidPlugin.start();

server-chat/src/main/java/com/cn/processor/IMChatAsyncChatMessageProcessor.java

@@ -1,5 +1,6 @@
 package com.cn.processor;
 
+import com.alibaba.fastjson.JSONObject;
 import org.jim.server.model.ChatGroup;
 import org.jim.server.model.Conversation;
 import org.jim.server.model.ConversationRecord;
@@ -72,7 +73,16 @@ public class IMChatAsyncChatMessageProcessor extends DefaultAsyncChatMessageProc
                         String serviceAccountId = user.getExtras().getString("serviceAccountId");
                         conversationRecordAttrs.put("from_id",Long.valueOf(serviceAccountId));
                         conversationRecordAttrs.put("to_id",conversation.getLong("user_id"));
-                        conversationRecordAttrs.put("type",1);
+                        JSONObject jsonObject = chatBody.getExtras();
+                        if( null != jsonObject && jsonObject.size() > 0 ){
+                            //客服自动发送的消息
+                            Integer sendType = jsonObject.getInteger("sendType");
+                            if( null != sendType && sendType == 2 ){
+                                conversationRecordAttrs.put("type",sendType);
+                            }
+                        }else{
+                            conversationRecordAttrs.put("type",1);
+                        }
                     }
                     conversationRecordAttrs.put("conversation_id",group.getLong("conversation_id"));
                     ConversationRecord.dao._setAttrs(conversationRecordAttrs).save();