package com.kxs.transfer.api.controller;
import com.alibaba.fastjson.JSON;
import com.aliyun.dts.subscribe.clients.record.DefaultUserRecord;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.kxs.transfer.api.annotation.DtsMsgListener;
import com.kxs.transfer.api.model.KxsDtsLog;
import com.kxs.transfer.api.model.table.DMLData;
import com.kxs.transfer.api.service.KxsDtsLogService;
import com.kxs.transfer.api.service.user.KxsUserAddressService;
import com.kxs.transfer.api.service.user.KxsUserService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
*
* DTS控制器
* 在此控制器里做数据分发
* 比如将数据分发到不同的数据源中
*
*
* @author 没秃顶的码农
* @date 2024-01-25
*/
@RestController
@Slf4j
@RequiredArgsConstructor
public class DtsController {
private final KxsDtsLogService kxsDtsLogService;
private final KxsUserService kxsUserService;
private final KxsUserAddressService kxsUserAddressService;
@DtsMsgListener
public void dtsListener(Long dataId, DMLData dmlData, DefaultUserRecord record){
//过滤无效数据
if("Users".equals(dmlData.getTableName())){
List changeFieldList = dmlData.getChangeFieldList();
//过滤Users老表记录登陆设备的信息日志
if (changeFieldList.size() == 2 && changeFieldList.contains("DeviceId") && changeFieldList.contains("DeviceType")){
record.commit(String.valueOf(record.getSourceTimestamp()));
return;
}
if (changeFieldList.size() == 1 && changeFieldList.contains("DeviceId")){
record.commit(String.valueOf(record.getSourceTimestamp()));
return;
}
//过滤Users本月交易额字段
if (changeFieldList.size() == 1 && changeFieldList.contains("ThisMonthTrade")){
record.commit(String.valueOf(record.getSourceTimestamp()));
return;
}
}
//去重数据
KxsDtsLog dtsLog = kxsDtsLogService.getOne(Wrappers.lambdaQuery().eq(KxsDtsLog::getDataId, dataId));
if(dtsLog != null){
log.info("dts的数据重复:{},dataID:{}", dmlData.getValidFieldDataMap(), dataId);
record.commit(String.valueOf(record.getSourceTimestamp()));
return;
}
//用户表
if("Users".equals(dmlData.getTableName())){
kxsUserService.changeUser(dmlData);
}
//用户地址表
if("UserAddress".equals(dmlData.getTableName())){
kxsUserAddressService.changeData(dmlData);
}
//密码操作
if("UserMoveInfo".equals(dmlData.getTableName())){
kxsUserService.changeUserPwd(dmlData);
}
//存储历史数据
KxsDtsLog kxsDtsLog = new KxsDtsLog();
kxsDtsLog.setDataId(dataId);
kxsDtsLog.setContent(JSON.toJSONString(dmlData));
kxsDtsLog.setTableName(dmlData.getTableName());
kxsDtsLog.setOperation(dmlData.getOperation().toString());
kxsDtsLogService.save(kxsDtsLog);
log.info("消费dts的数据表信息:{}", dmlData.getValidFieldDataMap());
record.commit(String.valueOf(record.getSourceTimestamp()));
}
}