Skip to content

向汉 / uc-consumer

Go to a project
Toggle navigation
Toggle navigation pinning
Commit 6856ef82 6856ef82f2f45a4566e254e89b02ebcd98aaa6e0 by xianghan
Options

1.优化日志信息 

2.清理无用的文件
1 parent 80f8168f
Inline Side-by-side
Showing 10 changed files with 140 additions and 367 deletions
......@@ -9,6 +9,7 @@ import com.topdraw.business.module.exp.detail.service.mapper.ExpDetailMapper;
import com.topdraw.utils.RedisUtils;
import com.topdraw.utils.StringUtils;
import com.topdraw.utils.ValidationUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.EmptyResultDataAccessException;
import org.springframework.stereotype.Service;
......@@ -23,6 +24,7 @@ import org.springframework.util.Assert;
*/
@Service
@Transactional(propagation = Propagation.SUPPORTS, readOnly = true, rollbackFor = Exception.class)
@Slf4j
public class ExpDetailServiceImpl implements ExpDetailService {
@Autowired
......@@ -57,8 +59,7 @@ public class ExpDetailServiceImpl implements ExpDetailService {
ExpDetail.copy(resources);
this.expDetailRepository.save(ExpDetail);
} catch (Exception e) {
e.printStackTrace();
throw e;
log.error("修改会员成长值详情异常，ExpDetailServiceImplupdate");
} finally {
this.redisUtils.doUnLock("expDetail::memberId::" + resources.getMemberId());
}
......
......@@ -89,7 +89,7 @@ public class MemberAddressBuilder {
MemberAddress memberAddress = new MemberAddress();
memberAddress.setId(Objects.nonNull(id)?null:id);
memberAddress.setMemberId(Objects.isNull(memberId)?null:memberId);
memberAddress.setMemberId(memberId);
memberAddress.setMemberCode(stringIsNull(memberCode));
memberAddress.setSequence(Objects.nonNull(sequence)?null:DEFAULT_VALUE_1);
......
package com.topdraw.business.module.member.address.service.impl;
import com.topdraw.business.module.member.address.domain.MemberAddress;
import com.topdraw.business.module.member.address.domain.MemberAddressBuilder;
import com.topdraw.business.module.member.address.repository.MemberAddressRepository;
import com.topdraw.business.module.member.address.service.MemberAddressService;
import com.topdraw.business.module.member.address.service.dto.MemberAddressDTO;
import com.topdraw.business.module.member.address.service.mapper.MemberAddressMapper;
import com.topdraw.business.module.member.service.MemberService;
import com.topdraw.business.module.member.service.dto.MemberDTO;
import com.topdraw.utils.ValidationUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
......@@ -45,10 +43,9 @@ public class MemberAddressServiceImpl implements MemberAddressService {
@Override
@Transactional(rollbackFor = Exception.class)
public MemberAddressDTO create(MemberAddress resources) {
log.info("MemberAddressServiceImpl ==>> create ==>> param ==>> [{}]",resources);
log.info("保存会员地址信息，参数 create# memberAddress ==>> resources ==>> {}", resources);
MemberAddress memberAddress = this.memberAddressRepository.save(resources);
log.info("MemberAddressServiceImpl ==>> create ==>> result ==>> [{}]",resources);
MemberAddressDTO memberAddressDTO = new MemberAddressDTO();
BeanUtils.copyProperties(memberAddress, memberAddressDTO);
return memberAddressDTO;
......@@ -58,24 +55,24 @@ public class MemberAddressServiceImpl implements MemberAddressService {
@Override
@Transactional(rollbackFor = Exception.class)
public MemberAddressDTO update(MemberAddress resources) {
log.info("MemberAddressServiceImpl ==>> update ==>> param ==>> [{}]",resources);
Assert.notNull(resources.getId(),"id can't be null");
log.info("修改会员地址信息，参数 update# memberAddress ==>> resources ==>> {}", resources);
try {
MemberAddress _memberAddress = this.memberAddressRepository.findById(resources.getId()).orElseGet(MemberAddress::new);
ValidationUtil.isNull( _memberAddress.getId(),"MemberAddress","id",resources.getId());
_memberAddress.copy(resources);
MemberAddress memberAddress = this.memberAddressRepository.save(_memberAddress);
MemberAddress memberAddress = this.memberAddressRepository.findById(resources.getId()).orElseGet(MemberAddress::new);
ValidationUtil.isNull( memberAddress.getId(),"MemberAddress","id",resources.getId());
memberAddress.copy(resources);
MemberAddress memberAddressSaveResult = this.memberAddressRepository.save(memberAddress);
MemberAddressDTO memberAddressDTO = new MemberAddressDTO();
BeanUtils.copyProperties(memberAddress, memberAddressDTO);
BeanUtils.copyProperties(memberAddressSaveResult, memberAddressDTO);
return memberAddressDTO;
} catch (Exception e) {
e.printStackTrace();
throw e;
log.error("修改会员地址信息异常，update# memberAddress message ==>> {}", e.getMessage());
}
return null;
}
@Override
......@@ -93,25 +90,4 @@ public class MemberAddressServiceImpl implements MemberAddressService {
return this.memberAddressMapper.toDto(memberAddress);
}
/**
* 检查会员
* @param memberAddress
* @return
*/
private MemberDTO checkMember(MemberAddress memberAddress){
Long memberId = memberAddress.getMemberId();
String memberCode = memberAddress.getMemberCode();
return this.checkMember(memberId,memberCode);
}
/**
* 检查会员
* @param memberId 会员id
* @param memberCode 会员code
* @return
*/
private MemberDTO checkMember(Long memberId, String memberCode) {
return this.memberService.checkMember(memberId, memberCode);
}
}
......
......@@ -9,9 +9,6 @@ import com.topdraw.business.module.member.profile.service.dto.MemberProfileDTO;
import com.topdraw.business.module.member.profile.service.mapper.MemberProfileMapper;
import com.topdraw.business.module.member.service.MemberService;
import com.topdraw.business.module.member.service.dto.MemberDTO;
import com.topdraw.util.Base64Util;
import com.topdraw.utils.RedisUtils;
import com.topdraw.utils.StringUtils;
import com.topdraw.utils.ValidationUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
......@@ -42,6 +39,7 @@ public class MemberProfileServiceImpl implements MemberProfileService {
private MemberProfileRepository memberProfileRepository;
@Override
@Transactional(readOnly = true)
public MemberProfileDTO findById(Long id) {
MemberProfile MemberProfile = this.memberProfileRepository.findById(id).orElseGet(MemberProfile::new);
ValidationUtil.isNull(MemberProfile.getId(),"MemberProfile","id",id);
......@@ -49,6 +47,7 @@ public class MemberProfileServiceImpl implements MemberProfileService {
}
@Override
@Transactional(readOnly = true)
public MemberProfileDTO findByMemberId(Long memberId) {
log.info("MemberProfileDTO ==>> findByMemberId ==>> resources ===>> [{}]",memberId);
MemberProfile memberProfile = this.memberProfileRepository.findByMemberId(memberId).orElseGet(MemberProfile::new);
......@@ -56,10 +55,10 @@ public class MemberProfileServiceImpl implements MemberProfileService {
}
@Override
@Transactional(readOnly = true)
public MemberProfileDTO findByMemberCode(String memberCode) {
log.info("MemberProfileDTO ==>> findByMemberCode ==>> resources ===>> [{}]",memberCode);
MemberDTO memberDTO = this.memberService.findByCode(memberCode);
log.info("MemberProfileDTO ==>> findByMemberCode ==>> result ===>> [{}]",memberDTO);
return this.findByMemberId(memberDTO.getId());
}
......@@ -79,21 +78,18 @@ public class MemberProfileServiceImpl implements MemberProfileService {
@Override
@Transactional(rollbackFor = Exception.class)
public MemberProfile createDefault(MemberProfile resources) {
MemberProfile memberProfile = MemberProfileBuilder.build(resources);
return this.create(memberProfile);
return this.create(MemberProfileBuilder.build(resources));
}
@Override
@Transactional(rollbackFor = Exception.class)
public MemberProfile createDefault(Member resources) {
MemberProfile memberProfile = MemberProfileBuilder.build(resources);
return this.create(memberProfile);
return this.create(MemberProfileBuilder.build(resources));
}
@Override
public MemberProfile createDefaultByMemberId(Long resources) {
MemberProfile memberProfile = MemberProfileBuilder.build(resources);
return this.create(memberProfile);
return this.create(MemberProfileBuilder.build(resources));
}
private MemberDTO checkMember(MemberProfile resources){
......@@ -111,11 +107,11 @@ public class MemberProfileServiceImpl implements MemberProfileService {
// 检查会员是否存在
MemberDTO memberDTO = this.checkMember(resources);
MemberProfileDTO _memberProfileDTO = this.findByMemberId(memberDTO.getId());
MemberProfileDTO memberProfileDTO = this.findByMemberId(memberDTO.getId());
if (Objects.nonNull(_memberProfileDTO.getId())) {
resources.setId(_memberProfileDTO.getId());
resources.setCreateTime(_memberProfileDTO.getCreateTime());
if (Objects.nonNull(memberProfileDTO.getId())) {
resources.setId(memberProfileDTO.getId());
resources.setCreateTime(memberProfileDTO.getCreateTime());
}
resources.setMemberId(memberDTO.getId());
......@@ -123,8 +119,8 @@ public class MemberProfileServiceImpl implements MemberProfileService {
log.info("memberProfileServiceImpl ==>> update ==>> memberId ==>> {} || resources =>> {}", memberDTO.getId(), resources);
MemberProfile _memberProfile = this.memberProfileRepository.save(resources);
MemberProfileDTO memberProfileDTO = new MemberProfileDTO();
BeanUtils.copyProperties(_memberProfile, memberProfileDTO);
MemberProfileDTO memberProfileDTO1 = new MemberProfileDTO();
BeanUtils.copyProperties(_memberProfile, memberProfileDTO1);
return memberProfileDTO;
}
......
......@@ -11,7 +11,5 @@ package com.topdraw.exception;
*/
public interface GlobeExceptionCode {
// 会员信息
Integer MEMBER_CODE_IS_NULL = 000100001;
}
......
package com.topdraw.exception;
/**
* @author ：
* @description：
* @function :
* @date ：Created in 2022/5/13 15:50
* @version: ：
* @modified By：
* @since : modified in 2022/5/13 15:50
*/
public class RabbitMqExceptionHandler {
}
package com.topdraw.mq.consumer;
import com.rabbitmq.client.Channel;
import com.topdraw.exception.BadRequestException;
import com.topdraw.mq.domain.TableOperationMsg;
import com.topdraw.resttemplate.RestTemplateClient;
import com.topdraw.util.FileUtil;
import com.topdraw.util.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import java.io.IOException;
import java.time.LocalDate;
import java.util.Map;
@Component
@Slf4j
public class UcEngineManagement2IptvConsumer {
@Autowired
AutoRoute autoUser;
@Autowired
RestTemplateClient restTemplateClient;
@Autowired
private RabbitTemplate rabbitTemplate;
private AutoRoute autoUser;
// @Value("#{rabbitMqErrorLogConfig.getUceError()}")
// private Map<String, String> error;
......@@ -43,37 +31,28 @@ public class UcEngineManagement2IptvConsumer {
@RabbitHandler
@RabbitListener(queues = "#{rabbitMqConfig.getMemberInfoQueue()}",
ackMode = "AUTO")
@Deprecated
public void memberInfoConsumer(Channel channel, Message message, String content) throws IOException {
log.info(" receive MemberInfoAsync msg , content is : {} ", content);
// TODO 已废弃
log.info("同步会员信息，参数 memberInfoConsumer# ==>> {} ", content);
try {
TableOperationMsg tableOperationMsg = this.parseContent(content);
if (StringUtils.isEmpty(content)) {
throw new BadRequestException("无参数");
}
TableOperationMsg tableOperationMsg = JSONUtil.parseMsg2Object(content,TableOperationMsg.class);
log.info("同步会员信息，解析参数后的结果，memberInfoConsumer# ==>> {}", tableOperationMsg);
autoUser.route(tableOperationMsg);
this.autoUser.route(tableOperationMsg);
// channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
log.error("消费uc-engine消息失败, cause ==>> [memberInfoConsumer#{}]", e.getMessage());
log.error("消费uc-engine消息失败, memberInfoConsumer# message ==>> {}", e.getMessage());
// channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
/*if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}*/
}
log.info("ucEventConsumer ====>>>> end");
}
/**
......@@ -91,46 +70,27 @@ public class UcEngineManagement2IptvConsumer {
@RabbitHandler
@RabbitListener(queues = "#{rabbitMqConfig.getUceQueue()}", ackMode = "AUTO")
public void ucEngineConsumer(Channel channel, Message message, String content) throws IOException {
log.info(" receive ucEngineConsumer msg , content is : {} ", content);
log.info("消费uc-engine信息，参数 ucEngineConsumer# content ==>> {} ", content);
try {
TableOperationMsg tableOperationMsg = this.parseContent(content);
autoUser.route(tableOperationMsg);
// channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
if (StringUtils.isEmpty(content)) {
throw new BadRequestException("无参数");
}
} catch (Exception e) {
log.error("消费uc-engine消息失败, cause ==>> [ucEngineConsumer#{}]", e.getMessage());
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
TableOperationMsg tableOperationMsg = JSONUtil.parseMsg2Object(content,TableOperationMsg.class);
/*if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
log.info("同步会员信息，解析参数后的结果，memberInfoConsumer# ==>> {}", tableOperationMsg);
this.autoUser.route(tableOperationMsg);
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}*/
// channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
e.printStackTrace();
}
} catch (Exception e) {
log.error("消费uc-engine消息失败, cause ==>> [ucEngineConsumer#{}]", e.getMessage());
log.info("ucEventConsumer ====>>>> end");
// channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
}
/**
* 数据解析
* @param content
* @return
*/
private TableOperationMsg parseContent(String content) {
TableOperationMsg tableOperationMsg = JSONUtil.parseMsg2Object(content,TableOperationMsg.class);
Assert.notNull(tableOperationMsg,"ERROR -->> operationConsumer -->> parseContent -->> 【dataSyncMsg】 not be null !!");
return tableOperationMsg;
}
}
......
......@@ -7,20 +7,14 @@ import com.topdraw.business.module.task.template.constant.TaskEventType;
import com.topdraw.exception.BadRequestException;
import com.topdraw.mq.domain.DataSyncMsg;
import com.topdraw.resttemplate.RestTemplateClient;
import com.topdraw.util.FileUtil;
import com.topdraw.util.JSONUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.*;
......@@ -50,24 +44,28 @@ public class UcEventBusIptv2ManagementUcEngine {
@RabbitListener(queues = "#{rabbitMqConfig.getEventBusQueue()}",
ackMode = "AUTO")
public void eventBusConsumer(Channel channel, Message message, String content) throws Exception {
log.info(" receive dataSync msg , content is ==>> {} ", content);
log.info("消费eventBus数据，参数 eventBusConsumer# content ==>> {} ", content);
try {
if (StringUtils.isBlank(content)) {
throw new BadRequestException("无参数");
}
DataSyncMsg dataSyncMsg = JSONUtil.parseMsg2Object(content, DataSyncMsg.class);
log.info("解析后的参数 , playContent ==>> {} ", dataSyncMsg);
log.info("消费eventBus数据，解析参数，eventBusConsumer# ==>> {} ", dataSyncMsg);
if (Objects.nonNull(dataSyncMsg)) {
String evt = dataSyncMsg.getEvt();
if (StringUtils.isBlank(evt)) {
log.error("eventBus事件类型（evt）为空");
log.error("消费eventBus数据异常，eventBusConsumer# message ==>> eventBus事件类型（evt）为空");
throw new BadRequestException("参数错误，事件类型 evt不存在");
}
LocalDateTime time = dataSyncMsg.getTime();
if (Objects.isNull(time)) {
log.error("参数错误，事件发送时间（time）不存在");
throw new BadRequestException("参数错误，事件发送时间（time）不存在");
log.error("消费eventBus数据异常，eventBusConsumer# message ==>> 发送时间（time）不存在");
throw new BadRequestException("参数错误，发送时间（time）不得为空");
} /*else {
if (time.isAfter(LocalDateTime.now()) || time.toLocalDate().compareTo(LocalDate.now()) != 0) {
log.error("参数错误，事件发送时间（time）非法 ==>> {}", time);
......@@ -77,8 +75,8 @@ public class UcEventBusIptv2ManagementUcEngine {
String msgData = dataSyncMsg.getMsgData();
if (StringUtils.isBlank(msgData)) {
log.error("eventBus事件消息体（msgData）为空");
throw new BadRequestException("参数错误，事件类型 evt不存在");
log.error("消费eventBus数据异常，eventBusConsumer# message ==>> 消息体（msgData）为空");
throw new BadRequestException("参数错误，消息体（msgData）不得为空");
}
switch (dataSyncMsg.getEvt().toUpperCase()) {
......@@ -97,7 +95,7 @@ public class UcEventBusIptv2ManagementUcEngine {
} catch (Exception e) {
log.error("eventBus 消费异常 ==>> {}",e.getMessage());
log.error("eventBus消费异常，eventBusConsumer# message ==>> {}", e.getMessage());
// TODO使用slf4j记录日志
/*if (MapUtils.isNotEmpty(error)) {
......@@ -113,7 +111,7 @@ public class UcEventBusIptv2ManagementUcEngine {
}*/
}
log.info("eventBusConsumer ====>>>> end");
}
/**
......
......@@ -6,21 +6,15 @@ import com.rabbitmq.client.Channel;
import com.topdraw.exception.BadRequestException;
import com.topdraw.mq.domain.DataSyncMsg;
import com.topdraw.resttemplate.RestTemplateClient;
import com.topdraw.util.FileUtil;
import com.topdraw.util.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import java.io.IOException;
import java.time.LocalDate;
import java.util.Map;
import java.util.Objects;
@Component
@Slf4j
......@@ -48,34 +42,23 @@ public class UcGatewayIptv2IptvConsumer {
@RabbitListener(queues = "#{rabbitMqConfig.getUcgEventQueue()}",
ackMode = "AUTO")
public void eventConsumer(Channel channel, Message message, String content) throws IOException {
log.info(" eventConsumer receive dataSync msg , content is : {} ", content);
log.info("消费uc-gateway任务消息，参数eventConsumer# content ==>> {} ", content);
try {
DataSyncMsg dataSyncMsg = JSONUtil.parseMsg2Object(content, DataSyncMsg.class);
if (Objects.nonNull(dataSyncMsg)) {
boolean jsonObject = this.restTemplateClient.dealTask(dataSyncMsg);
if (!jsonObject) {
throw new BadRequestException("uce处理任务响应超时");
if (StringUtils.isEmpty(content)) {
throw new BadRequestException("无参数");
}
}
} catch (Exception e) {
log.error("普通权益事件处理异常, ==>> {}", e.getMessage());
/*if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
DataSyncMsg dataSyncMsg = JSONUtil.parseMsg2Object(content, DataSyncMsg.class);
log.info("消费uc-gateway任务消息，解析参数结果，eventBusConsumer# dataSyncMsg ==>> {} ", dataSyncMsg);
if (StringUtils.isEmpty(errorStart) || errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
if (!this.restTemplateClient.dealTask(dataSyncMsg)) {
throw new BadRequestException("uce处理任务响应超时");
}
}*/
} catch (Exception e) {
log.error("普通权益事件处理异常，eventConsumer# message ==>> {}", e.getMessage());
}
log.info("ucEventConsumer ====>>>> end");
}
......@@ -93,49 +76,45 @@ public class UcGatewayIptv2IptvConsumer {
@RabbitListener(queues = "#{rabbitMqConfig.getGrowthReportQueue()}",
ackMode = "AUTO")
public void dealGrowthReport(Channel channel, Message message, String content) throws IOException {
log.info("receive dealGrowthReport add message, content {}", content);
log.info("消费uc-gateway成长报告消息，参数 dealGrowthReport# content ==>> {}", content);
try {
JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
if (Objects.nonNull(content)) {
Object msgData = jsonObject.get("msgData");
Boolean response = this.restTemplateClient.saveGrowthReport(JSON.toJSONString(msgData));
if (!response) {
log.error("同步大屏成长报告失败，uce接口响应超时");
}
if (StringUtils.isEmpty(content)) {
throw new BadRequestException("无参数");
}
} catch (Exception e) {
log.error("同步大屏成长报告失败,cause ==>> {}", e.getMessage());
/*if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
log.info("消费uc-gateway任务消息，解析参数结果，dealGrowthReport# jsonObject ==>> {} ", jsonObject);
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
if (!this.restTemplateClient.saveGrowthReport(JSON.toJSONString(jsonObject.get("msgData")))) {
log.error("同步大屏成长报告失败，uce接口响应超时");
}
}*/
} catch (Exception e) {
log.error("消费uc-gateway成长报告消息异常，dealGrowthReport# message ==>> {}", e.getMessage());
}
}
/**
* @description 添加收藏记录
* @description 收藏记录
* @param content 消息内容
*/
@RabbitHandler
@RabbitListener(queues = "#{rabbitMqConfig.getUcgCollectionQueue()}", ackMode = "AUTO")
public void collectionConsumer(Channel channel, Message message, String content) throws IOException {
log.info("receive UserCollection add message, content {}", content);
// TODO 收藏统一的操作入口，uc-gateway需要修改发送的队列
log.info("消费uc-gateway收藏操作消息，参数 collectionConsumer# content ==>> {}", content);
try {
if (StringUtils.isEmpty(content)) {
throw new BadRequestException("无参数");
}
JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
if (Objects.nonNull(content)) {
log.info("消费uc-gateway收藏操作消息，解析参数结果，collectionConsumer# jsonObject ==>> {} ", jsonObject);
String evt = jsonObject.get("evt").toString();
String msgData = jsonObject.get("msgData").toString();
switch (evt.toUpperCase()) {
......@@ -155,24 +134,9 @@ public class UcGatewayIptv2IptvConsumer {
break;
}
}
} catch (Exception e) {
log.error("收藏事件处理异常,cause ==>> {}", e.getMessage());
/*if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}*/
log.error("消费uc-gateway收藏操作消息异常，collectionConsumer# massage ==>> {}", e.getMessage());
}
}
......@@ -188,40 +152,26 @@ public class UcGatewayIptv2IptvConsumer {
@RabbitHandler
@RabbitListener(queues = "#{rabbitMqConfig.getViewRecordQueue()}", ackMode = "AUTO")
public void viewRecordConsumer(Channel channel, Message message, String content) throws IOException {
log.info("viewRecordConsumer receive ViewRecord add message, content {}", content);
log.info("消费uc-gateway观影记录消息，参数 viewRecordConsumer# content ==>> {}", content);
try {
if (StringUtils.isEmpty(content)) {
throw new BadRequestException("无参数");
}
JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
if (Objects.nonNull(content)) {
log.info("消费uc-gateway观影记录消息，解析参数结果，viewRecordConsumer# jsonObject ==>> {} ", jsonObject);
String evt = jsonObject.get("evt").toString();
String msgData = jsonObject.get("msgData").toString();
switch (evt.toUpperCase()) {
// 添加收藏
case "VIEWRECORD":
// 观影
if ("VIEWING".equals(evt.toUpperCase())) {
this.restTemplateClient.dealViewRecord(msgData);
break;
default:
break;
}
}
} catch (Exception e) {
log.error("观影事件处理异常,cause ==>> {}", e.getMessage());
/*if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}*/
log.error("消费uc-gateway观影记录消息，viewRecordConsumer# message ==>> {}", e.getMessage());
}
}
......@@ -239,50 +189,27 @@ public class UcGatewayIptv2IptvConsumer {
@RabbitListener(queues = "#{rabbitMqConfig.getUcgCollectionQueueAdd()}",
ackMode = "AUTO")
public void collectionConsumerAdd(Channel channel, Message message, String content) throws IOException {
log.info("receive collectionConsumerAdd add message, content {}", content);
log.info("消费uc-gateway添加收藏记录事件，参数 collectionConsumerAdd# content ==>> {} ", content);
try {
if (StringUtils.isEmpty(content)) {
throw new BadRequestException("无参数");
}
JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
if (Objects.nonNull(content)) {
log.info("消费uc-gateway添加收藏记录事件，解析参数结果，collectionConsumerAdd# jsonObject ==>> {} ", jsonObject);
String evt = jsonObject.get("evt").toString();
String msgData = jsonObject.get("msgData").toString();
switch (evt.toUpperCase()) {
// 添加收藏
case "ADDCOLLECTION":
if ("ADDCOLLECTION".equals(evt.toUpperCase())) {
this.restTemplateClient.addCollection(msgData);
break;
// 删除收藏
case "DELETECOLLECTION":
this.restTemplateClient.deleteCollection(msgData);
break;
// 删除全部收藏
case "DELETEALLCOLLECTION":
this.restTemplateClient.deleteAllCollection(msgData);
break;
default:
break;
}
}
} catch (Exception e) {
log.error("添加收藏记录事件处理异常,cause ==>> {}", e.getMessage());
/*if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}*/
}
}
......@@ -298,48 +225,27 @@ public class UcGatewayIptv2IptvConsumer {
@RabbitHandler
@RabbitListener(queues = "#{rabbitMqConfig.getUcgCollectionQueueDelete()}", ackMode = "AUTO")
public void collectionConsumerDelete(Channel channel, Message message, String content) throws IOException {
log.info("receive collectionConsumerDelete add message, content {}", content);
log.info("消费uc-gateway删除收藏记录，参数 collectionConsumerDelete# content ==>> {}", content);
try {
if (StringUtils.isEmpty(content)) {
throw new BadRequestException("无参数");
}
JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
if (Objects.nonNull(content)) {
log.info("消费uc-gateway删除收藏记录，解析参数结果，collectionConsumerDelete# jsonObject ==>> {} ", jsonObject);
String evt = jsonObject.get("evt").toString();
String msgData = jsonObject.get("msgData").toString();
switch (evt.toUpperCase()) {
// 添加收藏
case "ADDCOLLECTION":
this.restTemplateClient.addCollection(msgData);
break;
// 删除收藏
case "DELETECOLLECTION":
if ("DELETECOLLECTION".equals(evt.toUpperCase())) {
this.restTemplateClient.deleteCollection(msgData);
break;
// 删除全部收藏
case "DELETEALLCOLLECTION":
this.restTemplateClient.deleteAllCollection(msgData);
break;
default:
break;
}
}
} catch (Exception e) {
log.error("删除收藏记录事件处理异常,cause ==>> {}", e.getMessage());
/* if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}*/
log.error("消费uc-gateway删除收藏记录异常，collectionConsumerDelete# message ==>> {}", e.getMessage());
}
}
......@@ -357,53 +263,27 @@ public class UcGatewayIptv2IptvConsumer {
@RabbitListener(queues = "#{rabbitMqConfig.getUcgCollectionQueueDeleteAll()}",
ackMode = "AUTO")
public void collectionConsumerDeleteAll(Channel channel, Message message, String content) throws IOException {
log.info("receive collectionConsumerDeleteAll add message, content {}", content);
log.info("消费uc-gateway删除全部收藏记录，参数 collectionConsumerDeleteAll# content ==>> {}", content);
try {
if (StringUtils.isEmpty(content)) {
throw new BadRequestException("无参数");
}
JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
if (Objects.nonNull(content)) {
log.info("消费uc-gateway删除全部收藏记录，解析参数结果，collectionConsumerDeleteAll# jsonObject ==>> {} ", jsonObject);
String evt = jsonObject.get("evt").toString();
String msgData = jsonObject.get("msgData").toString();
switch (evt.toUpperCase()) {
// 添加收藏
case "ADDCOLLECTION":
this.restTemplateClient.addCollection(msgData);
break;
// 删除收藏
case "DELETECOLLECTION":
this.restTemplateClient.deleteCollection(msgData);
break;
// 删除全部收藏
case "DELETEALLCOLLECTION":
if ("DELETEALLCOLLECTION".equals(evt.toUpperCase())) {
this.restTemplateClient.deleteAllCollection(msgData);
break;
default:
break;
}
}
} catch (Exception e) {
log.error("删除全部收藏记录事件处理异常,cause ==>> {}", e.getMessage());
/*if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}*/
log.error("消费uc-gateway删除全部收藏记录，collectionConsumerDeleteAll# message ==>> {}", e.getMessage());
}
}
}
......
......@@ -3,21 +3,17 @@ package com.topdraw.mq.consumer;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.topdraw.exception.BadRequestException;
import com.topdraw.mq.domain.SubscribeBean;
import com.topdraw.resttemplate.RestTemplateClient;
import com.topdraw.util.FileUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;
import java.io.IOException;
import java.time.LocalDate;
import java.util.Map;
/**
* 微信事件
......@@ -49,9 +45,15 @@ public class WeiXinEventConsumer {
@RabbitHandler
@RabbitListener(queues = "#{rabbitMqConfig.getWechatQueue()}", ackMode = "AUTO")
public void subOrUnSubEvent(Channel channel, Message message, String content) throws IOException {
log.info("消费wechat-gate公众号关注、取关消息，参数 subOrUnSubEvent# content ==>> {}", content);
try {
log.info("receive wxu subOrUnSub message, content {}", content);
if (StringUtils.isEmpty(content)) {
throw new BadRequestException("无参数");
}
JSONObject jsonObject = JSONObject.parseObject(content);
log.info("消费uc-gate公众号关注、取关消息，解析参数结果，subOrUnSubEvent# jsonObject ==>> {} ", jsonObject);
JSONObject map = jsonObject.getJSONObject("appIdMap");
JSONObject wechatMsg = jsonObject.getJSONObject("allFieldsMap");
......@@ -65,8 +67,6 @@ public class WeiXinEventConsumer {
String event = wechatMsg.getString("Event");
String eventKey = wechatMsg.getString("EventKey");
log.info("event ==>> {}", event);
SubscribeBean subscribeBean = new SubscribeBean();
subscribeBean.setAppid(appid);
subscribeBean.setOpenid(openid);
......@@ -74,38 +74,18 @@ public class WeiXinEventConsumer {
subscribeBean.setEventKey(eventKey);
if (event.equals("subscribe")) {
log.info("send subscribe request start");
this.restTemplateClient.subscribe(subscribeBean);
log.info("send subscribe request end ");
}
if (event.equals("unsubscribe")) {
log.info("send unsubscribe request start");
this.restTemplateClient.unsubscribe(subscribeBean);
log.info("send unsubscribe request end");
}
// channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
} catch (Exception e) {
log.error("WXSubscribeConsumer || subOrUnSub msg error || {} || {}", content, e.getMessage());
/*channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}*/
log.info("ucEventConsumer ====>>>> end");
log.error("消费wechat-gate公众号关注、取关消息异常，subOrUnSubEvent# message ==>> {}", e.getMessage());
}
}
......