package cn.com.tenlion.smssender.controller;

import cn.com.tenlion.smssender.pojo.bos.HuaChuangSmsBO;
import cn.com.tenlion.smssender.pojo.vos.SmsP2PSendVO;
import cn.com.tenlion.smssender.pojo.vos.SmsSendVO;
import cn.com.tenlion.smssender.service.ISmsSendService;
import com.alibaba.fastjson2.JSON;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.*;

import java.util.Arrays;

/**
 * @ClassName: TestController
 * @Description:
 * @Author: wanggeng
 * @Date: 2022/12/6 22:09
 * @Version: 1.0
 */
@RestController
@RequestMapping("sms")
public class SmsController {

    @Autowired
    private ISmsSendService smsSendService;
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @PostMapping("send")
    public void send(@RequestBody SmsSendVO smsSendVO) {
        HuaChuangSmsBO huaChuangSmsBO = new HuaChuangSmsBO();
        huaChuangSmsBO.setShortMessageNo(smsSendVO.getShortMessageNo());
        huaChuangSmsBO.setMobiles(smsSendVO.getMobiles());
        huaChuangSmsBO.setContent(smsSendVO.getContent());
        smsSendService.send(huaChuangSmsBO);
    }

    @PostMapping("send-p2p")
    public void sendP2P(@RequestBody SmsP2PSendVO smsP2PSendVO) {
        HuaChuangSmsBO huaChuangSmsBO = new HuaChuangSmsBO();
        huaChuangSmsBO.setShortMessageNo(smsP2PSendVO.getShortMessageNo());
        huaChuangSmsBO.setMobileContents(smsP2PSendVO.getMobileContents());
        smsSendService.sendP2P(huaChuangSmsBO);
    }

    @GetMapping("kafka-test/{count}")
    public void sendP2P(@PathVariable("count") Integer count) {
        SmsSendVO smsSendVO = new SmsSendVO();
        smsSendVO.setShortMessageNo("test-short-message-uuid");
        smsSendVO.setMobiles(Arrays.asList("18634604067", "19147182823"));
        smsSendVO.setContent("测试");
        for (int i = 0; i < count; i++) {
            ListenableFuture<SendResult<String, Object>> smsSendListenableFuture = kafkaTemplate.send("sms_send", JSON.toJSONString(smsSendVO));
            smsSendListenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {

                @Override
                public void onFailure(Throwable ex) {
                    System.out.println("发送失败");
                    ex.printStackTrace();
                }

                @Override
                public void onSuccess(SendResult<String, Object> result) {
                    System.out.println("发送成功");
                    System.out.println(result.toString());
                }
            });
        }
    }

}