microser/rocketmq/SimpleProducer.h

#ifndef SIMPLEPRODUCER_H
#define SIMPLEPRODUCER_H

#ifdef __cplusplus

#include "../json/mms_json_inter.h"

#include "../rocketmq/DefaultMQProducer.h"
#include "../rocketmq/MQMessage.h"
#include "../rocketmq/SendResult.h"
#include "../rocketmq/SessionCredentials.h"
#include "../rocketmq/MQMessageExt.h"
#include "../rocketmq/ConsumeType.h"
#include "../rocketmq/MQMessageListener.h"

#include <vector>
#include <iostream>
#include <string>

/*
 * lnk20260622 修改：
 * 支持 RocketMQ / Redis Streams 双后端。
 *
 * 编译时：两套源码都编译进去。
 * 运行时：通过配置文件 [MQ] Backend=rocketmq 或 redisstream 选择。
 *
 * 业务层仍然使用：
 *   rocketmq::MQMessageExt
 *   rocketmq::ConsumeStatus
 *   Subscription
 *   InitializeProducer()
 *   rocketmq_producer_send()
 *   InitializeConsumer()
 *
 * 不需要业务层判断当前走 RocketMQ 还是 RedisStream。
 */

/*
 * 关键：
 * mq_compat.h 里必须定义兼容的：
 *   namespace rocketmq {
 *       enum ConsumeStatus;
 *       class MQMessageExt;
 *   }
 *
 * 如果实际编译 RocketMQ，也可以让 mq_compat.h 内部 include RocketMQ 原头。
 */
//#include "../mq/mq_compat.h"
#include "../rocketmq/SimpleProducer.h"

using namespace rocketmq;

void RocketMQ_rocketmq_producer_send(const std::string& body,
                            const std::string& topic,
                            const std::string& tags,
                            const std::string& keys);

extern "C" {
void rocketmq_test_rt();
void rocketmq_test_ud();
void rocketmq_test_rc();
void rocketmq_test_log();
void rocketmq_test_set();
void rocketmq_test_only();
void rocketmq_test_300(int mpnum, int front_index, int type);
}

extern void my_rocketmq_send(Ckafka_data_t& data);

///////////////////////////////////////////////////////
// 生产者统一入口
void RocketMQ_InitializeProducer();
void RocketMQ_ShutdownAndDestroyProducer();

///////////////////////////////////////////////////////
// 消费者统一入口
typedef ConsumeStatus (*MessageCallBack)(
    const MQMessageExt& msg
);

struct Subscription {
    std::string topic;
    std::string tag;
    MessageCallBack callback;

    Subscription(const std::string& t,
                 const std::string& tg,
                 MessageCallBack cb)
        : topic(t), tag(tg), callback(cb) {}
};

void RocketMQ_InitializeConsumer(const std::string& consumerName,
                        const std::string& nameServer,
                        const std::vector<Subscription>& subscriptions);

void RocketMQ_ShutdownAndDestroyConsumer();

void RocketMQ_rocketmq_consumer_receive(
    const std::string& consumerName,
    const std::string& nameServer,
    const std::vector<Subscription>& subscriptions);

void rocketmq_producer_send(const std::string& body,
                            const std::string& topic,
                            const std::string& tags,
                            const std::string& keys);
void InitializeProducer();
void ShutdownAndDestroyProducer();
void InitializeConsumer(const std::string& consumerName,
                        const std::string& nameServer,
                        const std::vector<Subscription>& subscriptions);
void ShutdownAndDestroyConsumer();

void rocketmq_consumer_receive(
    const std::string& consumerName,
    const std::string& nameServer,
    const std::vector<Subscription>& subscriptions);
//////////////////////////////////////////////////////

#endif  // __cplusplus

#endif  // SIMPLEPRODUCER_H