ClickHouse/src/Storages/RabbitMQ/RabbitMQSource.h

70 lines
1.7 KiB
C++
Raw Normal View History

#pragma once
2022-05-20 19:49:31 +00:00
#include <Processors/ISource.h>
#include <Storages/RabbitMQ/StorageRabbitMQ.h>
#include <Storages/RabbitMQ/RabbitMQConsumer.h>
namespace DB
{
2020-07-13 01:11:35 +00:00
2022-05-20 19:49:31 +00:00
class RabbitMQSource : public ISource
{
public:
RabbitMQSource(
StorageRabbitMQ & storage_,
const StorageSnapshotPtr & storage_snapshot_,
ContextPtr context_,
2020-08-28 08:52:02 +00:00
const Names & columns,
2020-08-31 16:34:16 +00:00
size_t max_block_size_,
2021-11-03 12:43:23 +00:00
bool ack_in_suffix = false);
~RabbitMQSource() override;
String getName() const override { return storage.getName(); }
RabbitMQConsumerPtr getBuffer() { return consumer; }
Chunk generate() override;
bool queueEmpty() const { return !consumer || consumer->hasPendingMessages(); }
2020-08-31 16:34:16 +00:00
bool needChannelUpdate();
2020-08-28 08:52:02 +00:00
void updateChannel();
bool sendAck();
void setTimeLimit(uint64_t max_execution_time_ms_) { max_execution_time_ms = max_execution_time_ms_; }
2022-05-20 19:49:31 +00:00
private:
StorageRabbitMQ & storage;
StorageSnapshotPtr storage_snapshot;
ContextPtr context;
Names column_names;
2020-08-31 16:34:16 +00:00
const size_t max_block_size;
2020-08-28 08:52:02 +00:00
bool ack_in_suffix;
bool is_finished = false;
2020-07-13 01:11:35 +00:00
const Block non_virtual_header;
const Block virtual_header;
Poco::Logger * log;
RabbitMQConsumerPtr consumer;
uint64_t max_execution_time_ms = 0;
2022-05-20 19:49:31 +00:00
Stopwatch total_stopwatch {CLOCK_MONOTONIC_COARSE};
bool isTimeLimitExceeded() const;
2022-05-20 19:49:31 +00:00
RabbitMQSource(
StorageRabbitMQ & storage_,
const StorageSnapshotPtr & storage_snapshot_,
std::pair<Block, Block> headers,
ContextPtr context_,
const Names & columns,
size_t max_block_size_,
bool ack_in_suffix);
Chunk generateImpl();
};
}