post_reprocessing.hpp
1.56 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
/*
* @Author: yangzilong
* @Last Modified by: yangzilong
* @Date: 2021-11-24 19:05:28
* @Email: yangzilong@objecteye.com
* @Description:
*/
#pragma once
#include "../ai_platform/header.h"
#ifdef POST_USE_RABBITMQ
#include "json/json.h"
#include <string.h>
#include <mutex>
#include <memory>
#ifdef _RBMQ_ASYNC_INTERFACE
#include <queue>
#include <atomic>
#include <thread>
#include <condition_variable>
#endif
#include "spdlog/spdlog.h"
#include "./rbmq/RabbitmqClient.hpp"
namespace mq
{
class post_rabbitmq_reprocessing
{
/**
* @brief
* 1. move able only.
* 2. thread safe class.
* 3. support async api.
* */
public:
post_rabbitmq_reprocessing();
~post_rabbitmq_reprocessing();
bool init(const rabbitmq_conn_params_t ¶ms);
bool sync_publish(const std::string &msg);
#ifdef _RBMQ_ASYNC_INTERFACE
bool async_publish(const std::string &msg);
#endif
post_rabbitmq_reprocessing(post_rabbitmq_reprocessing &&other);
post_rabbitmq_reprocessing& operator=(post_rabbitmq_reprocessing &&other);
/* remove copy construct and copy assignment. */
post_rabbitmq_reprocessing(const post_rabbitmq_reprocessing &) = delete;
post_rabbitmq_reprocessing& operator=(const post_rabbitmq_reprocessing &) = delete;
private:
#ifdef _RBMQ_ASYNC_INTERFACE
void run();
std::thread run_thread_;
std::atomic_bool is_run_;
std::condition_variable run_cv_;
std::queue<std::string> message_buffer_;
#endif
std::mutex run_mutex_;
rabbitmq_conn_params_t params_;
std::shared_ptr<CRabbitmqClient> rabbitmq_handle_;
};
}
#endif // #ifdef POST_USE_RABBITMQ