package com.xiaopeng.automaticPilot.controller;

import com.xiaopeng.automaticPilot.service.KafkaProducerService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

/**
 * Kafka 测试控制器
 */
@Slf4j
@RestController
@RequestMapping("/kafka")
@Api(tags = "Kafka 测试接口")
public class KafkaTestController {

    @Autowired
    private KafkaProducerService kafkaProducerService;

    /**
     * 发送消息（异步）
     */
    @PostMapping("/send")
    @ApiOperation("发送消息（异步）")
    public String sendMessage(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage(topic, message);
            return "消息发送成功";
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送消息（带 Key）
     */
    @PostMapping("/sendWithKey")
    @ApiOperation("发送消息（带 Key）")
    public String sendMessageWithKey(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息 Key", required = true) @RequestParam String key,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage(topic, key, message);
            return "消息发送成功";
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送消息（带回调）
     */
    @PostMapping("/sendWithCallback")
    @ApiOperation("发送消息（带回调）")
    public String sendMessageWithCallback(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessageWithCallback(topic, message);
            return "消息发送成功（带回调）";
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送消息（同步）
     */
    @PostMapping("/sendSync")
    @ApiOperation("发送消息（同步）")
    public String sendMessageSync(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessageSync(topic, message);
            return "消息发送成功（同步）";
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送自动驾驶测试消息
     */
    @PostMapping("/sendAutoPilot")
    @ApiOperation("发送自动驾驶测试消息")
    public String sendAutoPilotMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("automatic-pilot-topic", message);
            return "自动驾驶消息发送成功";
        } catch (Exception e) {
            log.error("发送自动驾驶消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送车辆状态测试消息
     */
    @PostMapping("/sendVehicleStatus")
    @ApiOperation("发送车辆状态测试消息")
    public String sendVehicleStatusMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("vehicle-status-topic", message);
            return "车辆状态消息发送成功";
        } catch (Exception e) {
            log.error("发送车辆状态消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送导航测试消息
     */
    @PostMapping("/sendNavigation")
    @ApiOperation("发送导航测试消息")
    public String sendNavigationMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("navigation-topic", message);
            return "导航消息发送成功";
        } catch (Exception e) {
            log.error("发送导航消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送订单测试消息
     */
    @PostMapping("/sendOrder")
    @ApiOperation("发送订单测试消息")
    public String sendOrderMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("order-topic", message);
            return "订单消息发送成功";
        } catch (Exception e) {
            log.error("发送订单消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }
}