ClickHouse/src/Storages/NATS/NATSHandler.h

60 lines
1.3 KiB
C++
Raw Normal View History

2022-05-08 12:12:15 +00:00
#pragma once
#include <memory>
#include <mutex>
2022-05-12 19:19:11 +00:00
#include <thread>
2022-05-08 12:12:15 +00:00
#include <nats.h>
#include <base/types.h>
#include <Poco/Logger.h>
namespace DB
{
namespace Loop
{
static const UInt8 RUN = 1;
static const UInt8 STOP = 2;
}
2022-05-09 20:16:11 +00:00
using SubscriptionPtr = std::unique_ptr<natsSubscription, decltype(&natsSubscription_Destroy)>;
2022-05-12 19:11:59 +00:00
using LockPtr = std::unique_ptr<std::lock_guard<std::mutex>>;
2022-05-09 20:16:11 +00:00
2022-05-08 12:12:15 +00:00
class NATSHandler
{
public:
NATSHandler(uv_loop_t * loop_, Poco::Logger * log_);
~NATSHandler();
/// Loop for background thread worker.
void startLoop();
/// Loop to wait for small tasks in a non-blocking mode.
/// Adds synchronization with main background loop.
void iterateLoop();
2022-05-12 19:11:59 +00:00
LockPtr setThreadLocalLoop();
2022-05-08 12:12:15 +00:00
void stopLoop();
2022-05-09 20:16:11 +00:00
void changeConnectionStatus(bool is_running);
2022-05-08 12:12:15 +00:00
bool connectionRunning() const { return connection_running.load(); }
bool loopRunning() const { return loop_running.load(); }
void updateLoopState(UInt8 state) { loop_state.store(state); }
UInt8 getLoopState() { return loop_state.load(); }
natsOptions * getOptions() { return opts; }
private:
uv_loop_t * loop;
natsOptions * opts = nullptr;
Poco::Logger * log;
std::atomic<bool> connection_running, loop_running;
std::atomic<UInt8> loop_state;
std::mutex startup_mutex;
};
}