拉取即时通讯历史消息

mec-biz/src/main/java/com/ym/mec/biz/service/UploadFileService.java

@@ -1,15 +1,13 @@
 package com.ym.mec.biz.service;
 
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
+import java.io.*;
 
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
 import com.ym.mec.common.entity.UploadReturnBean;
@@ -17,6 +15,7 @@ import com.ym.mec.common.exception.BizException;
 import com.ym.mec.thirdparty.storage.StoragePluginContext;
 import com.ym.mec.thirdparty.storage.provider.KS3StoragePlugin;
 import com.ym.mec.util.upload.UploadUtil;
+import org.springframework.web.multipart.MultipartFile;
 
 /** 
  * 上传工具服务层实现类
@@ -40,7 +39,6 @@ public class UploadFileService {
 	private String fileRoot;
 
 	public UploadReturnBean uploadFile(InputStream in, String ext) {
-
 		UploadReturnBean uploadReturn = new UploadReturnBean("", false, "");
 		String fileName = UploadUtil.getFileName(ext);
 
@@ -77,6 +75,32 @@ public class UploadFileService {
 		return uploadReturn;
 	}
 
+
+	@Async
+	public UploadReturnBean uploadImHistoryMsgFile(File msgFile) throws FileNotFoundException {
+		InputStream in = new FileInputStream(msgFile);
+		UploadReturnBean uploadReturn = new UploadReturnBean("", false, "");
+		String fileName = UploadUtil.getFileName(msgFile.getName());
+
+		String root = fileRoot + "im_history_msg/";
+		if (StringUtils.isBlank(root)) {
+			uploadReturn.setMessage("上传临时目录没有配置");
+			return uploadReturn;
+		}
+
+		String staticFloder = "";
+		String folder = UploadUtil.getFileFloder();
+		String filePath = UploadUtil.getFilePath(root, staticFloder, folder);
+		File file = uploadFile(in, filePath, fileName);
+		String url = storagePluginContext.uploadFile(KS3StoragePlugin.PLUGIN_NAME,staticFloder + folder, file);
+
+		FileUtils.deleteQuietly(file);
+
+		uploadReturn.setStatus(true);
+		uploadReturn.setUrl(url);
+		return uploadReturn;
+	}
+
 	public void setMaxSize(int maxSize) {
 		this.maxSize = maxSize;
 	}

mec-client-api/src/main/java/com/ym/mec/im/ImFeignService.java

@@ -109,4 +109,12 @@ public interface ImFeignService {
 
 	@PostMapping(value = "group/send")
 	Object groupSend(@RequestBody ImGroupMessage imGroupMessage);
+
+	/**
+	 * 获取历史消息记录
+	 * @param date
+	 * @return
+	 */
+	@PostMapping(value = "history/get")
+	Object historyGet(String date);
 }

mec-client-api/src/main/java/com/ym/mec/im/fallback/ImFeignServiceFallback.java

@@ -75,9 +75,9 @@ public class ImFeignServiceFallback implements ImFeignService {
     public Object groupSend(ImGroupMessage imGroupMessage) {
         return null;
     }
-/*
+
     @Override
-    public Object privateSendCustom(ImGroupMessage groupMessage) {
+    public Object historyGet(String date) {
         return null;
-    }*/
+    }
 }

mec-common/audit-log/src/main/java/com/yonge/log/dal/dao/HistoryMessageDao.java

@@ -0,0 +1,16 @@
+package com.yonge.log.dal.dao;
+
+import com.yonge.log.dal.model.AuditLog;
+import com.yonge.log.dal.model.HistoryMessage;
+import com.yonge.mongodb.dao.BaseDaoWithMongo;
+import org.springframework.stereotype.Component;
+
+@Component
+public class HistoryMessageDao extends BaseDaoWithMongo<String, HistoryMessage> {
+
+	@Override
+	public Class<HistoryMessage> getClassEntity() {
+		return HistoryMessage.class;
+	}
+
+}

mec-common/audit-log/src/main/java/com/yonge/log/dal/model/HistoryMessage.java

@@ -0,0 +1,150 @@
+package com.yonge.log.dal.model;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.io.Serializable;
+
+@Document(collection = "im_history_message")
+public class HistoryMessage implements Serializable {
+    @Id
+    private String msgUID;
+
+    private String fromUserId;
+
+    private String targetId;
+
+    private Integer targetType;
+
+    private String GroupId;
+
+    private String classname;
+
+    private String dateTime;
+
+    private String source;
+
+    private String isDiscard;
+
+    private String isSensitiveWord;
+
+    private String isForbidden;
+
+    private String isNotForward;
+
+    private String groupUserIds;
+
+    private String content;
+
+    public String getFromUserId() {
+        return fromUserId;
+    }
+
+    public void setFromUserId(String fromUserId) {
+        this.fromUserId = fromUserId;
+    }
+
+    public String getTargetId() {
+        return targetId;
+    }
+
+    public void setTargetId(String targetId) {
+        this.targetId = targetId;
+    }
+
+    public Integer getTargetType() {
+        return targetType;
+    }
+
+    public void setTargetType(Integer targetType) {
+        this.targetType = targetType;
+    }
+
+    public String getGroupId() {
+        return GroupId;
+    }
+
+    public void setGroupId(String groupId) {
+        GroupId = groupId;
+    }
+
+    public String getClassname() {
+        return classname;
+    }
+
+    public void setClassname(String classname) {
+        this.classname = classname;
+    }
+
+    public String getDateTime() {
+        return dateTime;
+    }
+
+    public void setDateTime(String dateTime) {
+        this.dateTime = dateTime;
+    }
+
+    public String getSource() {
+        return source;
+    }
+
+    public void setSource(String source) {
+        this.source = source;
+    }
+
+    public String getIsDiscard() {
+        return isDiscard;
+    }
+
+    public void setIsDiscard(String isDiscard) {
+        this.isDiscard = isDiscard;
+    }
+
+    public String getIsSensitiveWord() {
+        return isSensitiveWord;
+    }
+
+    public void setIsSensitiveWord(String isSensitiveWord) {
+        this.isSensitiveWord = isSensitiveWord;
+    }
+
+    public String getIsForbidden() {
+        return isForbidden;
+    }
+
+    public void setIsForbidden(String isForbidden) {
+        this.isForbidden = isForbidden;
+    }
+
+    public String getIsNotForward() {
+        return isNotForward;
+    }
+
+    public void setIsNotForward(String isNotForward) {
+        this.isNotForward = isNotForward;
+    }
+
+    public String getMsgUID() {
+        return msgUID;
+    }
+
+    public void setMsgUID(String msgUID) {
+        this.msgUID = msgUID;
+    }
+
+    public String getGroupUserIds() {
+        return groupUserIds;
+    }
+
+    public void setGroupUserIds(String groupUserIds) {
+        this.groupUserIds = groupUserIds;
+    }
+
+    public String getContent() {
+        return content;
+    }
+
+    public void setContent(String content) {
+        this.content = content;
+    }
+}

mec-common/audit-log/src/main/java/com/yonge/log/service/HistoryMessageService.java

@@ -0,0 +1,16 @@
+package com.yonge.log.service;
+
+import com.yonge.log.dal.model.HistoryMessage;
+import com.yonge.mongodb.service.BaseServiceWithMongo;
+
+import java.io.File;
+
+public interface HistoryMessageService extends BaseServiceWithMongo<String, HistoryMessage> {
+
+    /**
+     * 保存融云历史聊天记录
+     * @param file
+     * @throws Exception
+     */
+    void saveImHistoryMessage(File file) throws Exception;
+}

mec-common/audit-log/src/main/java/com/yonge/log/service/impl/HistoryMessageServiceImpl.java

@@ -0,0 +1,112 @@
+package com.yonge.log.service.impl;
+
+import com.alibaba.fastjson.JSONObject;
+import com.yonge.log.dal.dao.HistoryMessageDao;
+import com.yonge.log.dal.model.HistoryMessage;
+import com.yonge.log.service.HistoryMessageService;
+import com.yonge.mongodb.dao.BaseDaoWithMongo;
+import com.yonge.mongodb.service.impl.BaseServiceImplWithMongo;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileReader;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.ArrayList;
+import java.util.List;
+
+@Service
+public class HistoryMessageServiceImpl extends BaseServiceImplWithMongo<String, HistoryMessage> implements HistoryMessageService {
+
+	@Autowired
+	private HistoryMessageDao historyMessageDao;
+
+	@Override
+	public BaseDaoWithMongo<String, HistoryMessage> getDAO() {
+		return historyMessageDao;
+	}
+
+	@Override
+	public void saveImHistoryMessage(File file) throws Exception {
+		long length = file.length();
+		if(length > 1048576){
+			doNioReadFile(file);
+		}else {
+			doBufferReadFile(file);
+		}
+	}
+
+	public void doBufferReadFile(File file) throws Exception
+	{
+		BufferedReader reader = null;
+		try {
+			reader = new BufferedReader(new FileReader(file));
+			String line = null;
+			int num = 0;
+			List<HistoryMessage> historyMessages = new ArrayList<>();
+			while ((line = reader.readLine()) != null){
+				try {
+					historyMessages.add(JSONObject.parseObject(line.substring(line.indexOf("{")), HistoryMessage.class));
+					if(historyMessages.size() >= 2000){
+						historyMessageDao.batchInsert(historyMessages,HistoryMessage.class);
+						historyMessages.clear();
+					}
+				}catch (Exception e){
+					e.printStackTrace();
+				}
+				num++;
+			}
+			if(historyMessages.size() > 0){
+				historyMessageDao.batchInsert(historyMessages,HistoryMessage.class);
+			}
+		} finally {
+			reader.close();
+		}
+	}
+
+
+	public void doNioReadFile(File file) throws Exception
+	{
+		String enterStr = "\n";
+
+		FileChannel inChannel = new FileInputStream(file).getChannel();
+		ByteBuffer buffer = ByteBuffer.allocate(1048576);
+
+		StringBuilder newlinesBui = new StringBuilder();
+		while (inChannel.read(buffer) != -1)
+		{
+			buffer.flip();
+			//数据组合.
+			String content = new String(buffer.array());
+			newlinesBui.append(content).toString();
+
+			int fromIndex = 0;
+			int endIndex = -1;
+			//循环找到 \n
+			List<HistoryMessage> historyMessages = new ArrayList<>();
+			while ((endIndex = newlinesBui.indexOf(enterStr, fromIndex)) > -1)
+			{
+				//得到一行
+				String line = newlinesBui.substring(fromIndex, endIndex);
+				try {
+					historyMessages.add(JSONObject.parseObject(line.substring(line.indexOf("{")), HistoryMessage.class));
+					if(historyMessages.size() >= 2000){
+						historyMessageDao.batchInsert(historyMessages,HistoryMessage.class);
+						historyMessages.clear();
+					}
+				}catch (Exception e){
+					e.printStackTrace();
+				}
+				fromIndex = endIndex + 1;
+			}
+			if(historyMessages.size() > 0){
+				historyMessageDao.batchInsert(historyMessages,HistoryMessage.class);
+			}
+			newlinesBui.delete(0, fromIndex);
+			buffer.clear();
+		}
+	}
+}

mec-common/mongo-db/src/main/java/com/yonge/mongodb/dao/BaseDaoWithMongo.java

@@ -3,6 +3,7 @@ package com.yonge.mongodb.dao;
 import java.io.Serializable;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -75,6 +76,15 @@ public abstract class BaseDaoWithMongo<PK, T extends Serializable> {
 	}
 
 	/**
+	 * 写入实体对象
+	 * @param batchBean
+	 * @param clazz
+	 */
+	public void batchInsert(List<T> batchBean, Class clazz) {
+		mongoTemplate.insert(batchBean,clazz);
+	}
+
+	/**
 	 * 通过参数查找所有结果集
 	 * @param params
 	 * @return

mec-im/src/main/java/com/ym/config/ResourceServerConfig.java

@@ -12,6 +12,7 @@ public class ResourceServerConfig extends ResourceServerConfigurerAdapter {
     public void configure(HttpSecurity http) throws Exception {
         http.authorizeRequests().antMatchers("/v2/api-docs","/user/register",
                 "/group/join","/group/create","/group/quit","/room/leave","/room/statusSync",
-                "/room/statusImMsg","/group/batchDismiss","/private/send","/group/send","/group/dismiss","/room/statusImMsg").permitAll().anyRequest().authenticated().and().csrf().disable();
+                "/room/statusImMsg","/group/batchDismiss","/private/send","/group/send",
+                "/group/dismiss","/room/statusImMsg","/history/get").permitAll().anyRequest().authenticated().and().csrf().disable();
     }
 }

mec-im/src/main/java/com/ym/controller/HistoryController.java

@@ -2,14 +2,10 @@ package com.ym.controller;
 
 import com.ym.service.MessageService;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.bind.annotation.RestController;
 
-/**
- * Created by weiqinxiao on 2019/2/25.
- */
 @RestController
 @RequestMapping("/history")
 public class HistoryController{
@@ -18,12 +14,12 @@ public class HistoryController{
     MessageService messageService;
 
     @RequestMapping(value = "/get", method = RequestMethod.POST)
-    public Object get(@RequestBody String date) throws Exception {
+    public Object get(String date) throws Exception {
         return messageService.historyGet(date);
     }
 
     @RequestMapping(value = "/remove", method = RequestMethod.POST)
-    public Object remove(@RequestBody String date) throws Exception {
+    public Object remove(String date) throws Exception {
         return messageService.historyRemove(date);
     }
 }

mec-im/src/main/java/com/ym/service/Impl/MessageServiceImpl.java

@@ -47,7 +47,10 @@ public class MessageServiceImpl implements MessageService {
         return msgSystem;
     }
     private History getHistory(){
-        return new History(appKey,appSecret);
+        RongCloud rongCloud = RongCloud.getInstance(appKey, appSecret);
+        History history = new History(appKey, appSecret);
+        history.setRongCloud(rongCloud);
+        return history;
     }
     private Push getPush(){
         return new Push(appKey,appSecret);

mec-web/src/main/java/com/ym/mec/web/controller/ImHistoryMessageController.java

@@ -0,0 +1,35 @@
+package com.ym.mec.web.controller;
+
+import com.ym.mec.biz.service.UploadFileService;
+import com.ym.mec.common.controller.BaseController;
+import com.ym.mec.common.entity.UploadReturnBean;
+import com.ym.mec.thirdparty.storage.StoragePluginContext;
+import com.yonge.log.service.HistoryMessageService;
+import io.swagger.annotations.Api;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.io.File;
+
+@RequestMapping("imHistoryMessage")
+@Api(tags = "系统操作日志")
+@RestController
+public class ImHistoryMessageController extends BaseController {
+
+	@Autowired
+	private HistoryMessageService historyMessageService;
+	@Autowired
+	private UploadFileService uploadFileService;
+
+
+	@GetMapping("/save")
+	public Object saveImHistoryMessage(String fileDir) throws Exception {
+		File file = new File(fileDir);
+		UploadReturnBean uploadReturnBean = uploadFileService.uploadImHistoryMsgFile(file);
+		historyMessageService.saveImHistoryMessage(file);
+		return succeed();
+	}
+
+}