# AdjSenderServiceImpl.java

    package cloud.wconnect.frontoffice.service.adjustment.impl;

    import cloud.wconnect.commonlib.component.BeanMapper;
    import cloud.wconnect.commonlib.cons.MoudleEventSubjectEnum;
    import cloud.wconnect.commonlib.pojo.bo.adj.AdjBO;
    import cloud.wconnect.commonlib.pojo.bo.adj.AdjCommentBO;
    import cloud.wconnect.commonlib.pojo.bo.adj.AdjCommentDetailBO;
    import cloud.wconnect.commonlib.pojo.bo.adj.BatchAdjBO;
    import cloud.wconnect.commonlib.util.JsonUtils;
    import cloud.wconnect.frontoffice.cons.adjustment.AdjustmentStatusEnum;
    import cloud.wconnect.frontoffice.entity.adjustment.AdRequest;
    import cloud.wconnect.frontoffice.entity.adjustment.AdRequestComment;
    import cloud.wconnect.frontoffice.service.adjustment.IAdRequestCommentService;
    import cloud.wconnect.frontoffice.service.adjustment.IAdRequestService;
    import cloud.wconnect.frontoffice.service.adjustment.IAdjSenderService;
    import cloud.wconnect.sns.http.processor.Sender;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.google.common.collect.Lists;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections.CollectionUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Lazy;
    import org.springframework.stereotype.Service;
    import software.amazon.awssdk.services.sns.model.PublishResponse;

    import java.time.Instant;
    import java.util.Collections;
    import java.util.List;
    import java.util.stream.Collectors;
    import java.util.stream.IntStream;

    /**
     * @author lichen
     * @description 发送Adjustment数据变更的消息service 实现类
     * @date 2023/3/31 15:43
     */
    @Service
    @Slf4j
    public class AdjSenderServiceImpl implements IAdjSenderService {

        @Qualifier("moduleEventSender")
        @Autowired
        private Sender snsSender;
        @Autowired
        private ObjectMapper objectMapper;
        @Autowired
        private BeanMapper beanMapper;
        @Autowired
        @Lazy
        private IAdRequestService adRequestService;
        @Autowired
        @Lazy
        private IAdRequestCommentService adRequestCommentService;

        /**
         * 调账数据发生变更时，发送事件
         *
         * @param propertyId 资产ID
         * @param requestId  调账数据主键ID
         * @author lichen
         * @date 2024/2/1
         */
        @Override
        public void publishSave(Integer propertyId, Long requestId) {
            AdRequest adRequest = adRequestService.getById(requestId);
            // 排除掉作废的数据
            if (adRequest.getStatus() == AdjustmentStatusEnum.CANCEL.getStatus()) {
                return;
            }

            AdjBO adjBO = beanMapper.map(adRequest, AdjBO.class, (bo) -> {
                bo.setAdjustmentId(requestId);
            });
            //封装时间对象
            BatchAdjBO batchAdjBO = BatchAdjBO.builder()
                    .propertyId(propertyId)
                    .actionType(BatchAdjBO.ACTION_TYPE_SAVE)
                    .data(Lists.newArrayList(adjBO))
                    .updateTime(Instant.now().toEpochMilli())
                    .build();
            //发送消息
            String msg = JsonUtils.convertObjToString(objectMapper, batchAdjBO);
            PublishResponse response = snsSender.publish(MoudleEventSubjectEnum.ADJUSTMENT_AMOUNT.getSubject(), msg);
            log.info("Adjustment send adjustment-amount,msgId:{}, body:{}", response.messageId(), msg);

        }

        /**
         * 调账数据发生删除时，发送事件
         *
         * @param propertyId 资产ID
         * @param adjList    调账数据集合
         * @author lichen
         * @date 2024/2/1
         */
        @Override
        public void publishDelete(Integer propertyId, List<AdRequest> adjList) {
            // 排除掉作废的数据
            List<AdRequest> validAdjList = adjList.stream().filter(adj -> adj.getStatus() != AdjustmentStatusEnum.CANCEL.getStatus()).toList();
            if (CollectionUtils.isEmpty(validAdjList)) return;

            List<AdjBO> adjBOS = beanMapper.mapList(validAdjList, AdjBO.class, (entity, bo) -> {
                bo.setAdjustmentId(entity.getId());
            });
            //封装时间对象
            BatchAdjBO batchAdjBO = BatchAdjBO.builder()
                    .propertyId(propertyId)
                    .actionType(BatchAdjBO.ACTION_TYPE_DELETE)
                    .data(adjBOS)
                    .updateTime(Instant.now().toEpochMilli())
                    .build();
            //发送消息
            String msg = JsonUtils.convertObjToString(objectMapper, batchAdjBO);
            PublishResponse response = snsSender.publish(MoudleEventSubjectEnum.ADJUSTMENT_AMOUNT.getSubject(), msg);
            log.info("Adjustment send adjustment-amount,msgId:{}, body:{}", response.messageId(), msg);
        }

        /**
         * 调账数据发生变更时，发送事件
         *
         * @param propertyId 资产ID
         * @param adjIds     新增的调账id集合
         * @author lichen
         * @date 2024/2/1
         */
        @Override
        public void batchPublishSave(Integer propertyId, List<Long> adjIds) {
            if (CollectionUtils.isEmpty(adjIds)) return;

            List<AdRequest> adjList = adRequestService.listByIds(adjIds);
            // 排除掉作废的数据
            List<AdRequest> validAdjList = adjList.stream().filter(adj -> adj.getStatus() != AdjustmentStatusEnum.CANCEL.getStatus()).toList();
            if (CollectionUtils.isEmpty(validAdjList)) return;

            List<List<AdRequest>> lists = splitList(validAdjList, 200);
            for (List<AdRequest> subList : lists) {
                List<AdjBO> adjBOS = beanMapper.mapList(subList, AdjBO.class, (entity, bo) -> {
                    bo.setAdjustmentId(entity.getId());
                });
                //封装时间对象
                BatchAdjBO batchAdjBO = BatchAdjBO.builder()
                        .propertyId(propertyId)
                        .actionType(BatchAdjBO.ACTION_TYPE_SAVE)
                        .data(adjBOS)
                        .updateTime(Instant.now().toEpochMilli())
                        .build();
                //发送消息
                String msg = JsonUtils.convertObjToString(objectMapper, batchAdjBO);
                PublishResponse response = snsSender.publish(MoudleEventSubjectEnum.ADJUSTMENT_AMOUNT.getSubject(), msg);
                log.info("Adjustment send adjustment-amount,msgId:{}, body:{}", response.messageId(), msg);
            }
        }

        /**
         * 调账备注数据发生变更时，发送事件
         *
         * @param propertyId 资产ID
         * @param adjId      调账id
         * @author lichen
         * @date 2026/02/09
         */
        @Override
        public void publishSaveComment(int propertyId, Long adjId) {
            // 查询调账数据的备注
            List<AdRequestComment> adRequestComments = adRequestCommentService.listByAdjustmentId(Collections.singletonList(adjId));
            List<AdjCommentDetailBO> adjCommentDetailBOS = beanMapper.mapList(adRequestComments, AdjCommentDetailBO.class);

            AdjCommentBO adjCommentBO = new AdjCommentBO();
            adjCommentBO.setPropertyId(propertyId)
                    .setAdjustmentId(adjId)
                    .setComments(adjCommentDetailBOS);

            String msg = JsonUtils.convertObjToString(objectMapper, adjCommentBO);
            PublishResponse response = snsSender.publish(MoudleEventSubjectEnum.ADJUSTMENT_COMMENT.getSubject(), msg);
            log.info("Adjustment send adjustment-comment,msgId:{}, body:{}", response.messageId(), msg);
        }

        /**
         * 分隔集合
         *
         * @param list      集合
         * @param chunkSize 分隔大小
         * @return java.util.List<java.util.List < T>>
         * @author lichen
         * @date 2025/8/19
         */
        private <T> List<List<T>> splitList(List<T> list, int chunkSize) {
            return IntStream.iterate(0, i -> i < list.size(), i -> i + chunkSize)
                    .mapToObj(i -> list.subList(i, Math.min(i + chunkSize, list.size())))
                    .collect(Collectors.toList());
        }

    }