Skip to content

Commit 3e98bfd

Browse files
author
anyshy
committed
format
1 parent b2615b9 commit 3e98bfd

File tree

2 files changed

+32
-34
lines changed

2 files changed

+32
-34
lines changed

‎include/eosio/kafka_plugin/kafka_plugin.hpp

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,15 +5,11 @@
55
#pragma once
66

77
#include <eosio/chain_plugin/chain_plugin.hpp>
8-
98
#include <appbase/application.hpp>
10-
#include <eosio/chain/controller.hpp>
119
#include <memory>
1210

1311
namespace eosio {
14-
using eosio::chain::controller;
1512
using std::unique_ptr;
16-
using namespace appbase;
1713

1814
/**
1915
* Provides persistence to kafka for:
@@ -34,7 +30,8 @@ namespace eosio {
3430

3531
virtual void set_program_options(options_description &cli, options_description &cfg) override;
3632

37-
void plugin_initialize(const variables_map&);
33+
void plugin_initialize(const variables_map &options);
34+
3835
void plugin_startup();
3936

4037
void plugin_shutdown();

‎kafka_plugin.cpp

Lines changed: 30 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
#include <boost/thread/condition_variable.hpp>
2626

2727
#include <queue>
28+
2829
using eosio::chain::name;
2930
namespace fc { class variant; }
3031

@@ -50,10 +51,10 @@ namespace eosio {
5051

5152
~kafka_plugin_impl();
5253

53-
std::optional<boost::signals2::scoped_connection> accepted_block_connection;
54-
std::optional<boost::signals2::scoped_connection> irreversible_block_connection;
55-
std::optional<boost::signals2::scoped_connection> accepted_transaction_connection;
56-
std::optional<boost::signals2::scoped_connection> applied_transaction_connection;
54+
std::optional <boost::signals2::scoped_connection> accepted_block_connection;
55+
std::optional <boost::signals2::scoped_connection> irreversible_block_connection;
56+
std::optional <boost::signals2::scoped_connection> accepted_transaction_connection;
57+
std::optional <boost::signals2::scoped_connection> applied_transaction_connection;
5758
chain_plugin *chain_plug;
5859
struct action_info {
5960
account_name account;
@@ -65,14 +66,13 @@ namespace eosio {
6566
struct trasaction_info_st {
6667
uint64_t block_number;
6768
fc::time_point block_time;
68-
std::optional<chain::chain_id_type> chain_id;
69+
std::optional <chain::chain_id_type> chain_id;
6970
chain::transaction_trace_ptr trace;
70-
vector<action_info> action_vec;
71+
vector <action_info> action_vec;
7172

7273
};
7374

7475

75-
7676
void consume_blocks();
7777

7878
void accepted_block(const chain::block_state_ptr &);
@@ -103,8 +103,9 @@ namespace eosio {
103103

104104
bool configured{false};
105105

106-
void filter_traction_trace(const chain::transaction_trace_ptr trace,action_name act_name);
107-
void _process_trace(vector<chain::action_trace>::iterator action_trace_ptr,action_name act_name);
106+
void filter_traction_trace(const chain::transaction_trace_ptr trace, action_name act_name);
107+
108+
void _process_trace(vector<chain::action_trace>::iterator action_trace_ptr, action_name act_name);
108109

109110
template<typename Queue, typename Entry>
110111
void queue(Queue &queue, const Entry &e);
@@ -114,20 +115,20 @@ namespace eosio {
114115

115116
size_t max_queue_size = 10000;
116117
int queue_sleep_time = 0;
117-
std::deque<chain::transaction_metadata_ptr> transaction_metadata_queue;
118-
std::deque<chain::transaction_metadata_ptr> transaction_metadata_process_queue;
119-
std::deque<trasaction_info_st> transaction_trace_queue;
120-
std::deque<trasaction_info_st> transaction_trace_process_queue;
121-
std::deque<chain::block_state_ptr> block_state_queue;
122-
std::deque<chain::block_state_ptr> block_state_process_queue;
123-
std::deque<chain::block_state_ptr> irreversible_block_state_queue;
124-
std::deque<chain::block_state_ptr> irreversible_block_state_process_queue;
118+
std::deque <chain::transaction_metadata_ptr> transaction_metadata_queue;
119+
std::deque <chain::transaction_metadata_ptr> transaction_metadata_process_queue;
120+
std::deque <trasaction_info_st> transaction_trace_queue;
121+
std::deque <trasaction_info_st> transaction_trace_process_queue;
122+
std::deque <chain::block_state_ptr> block_state_queue;
123+
std::deque <chain::block_state_ptr> block_state_process_queue;
124+
std::deque <chain::block_state_ptr> irreversible_block_state_queue;
125+
std::deque <chain::block_state_ptr> irreversible_block_state_process_queue;
125126
std::mutex mtx;
126127
std::condition_variable condition;
127128
std::thread consume_thread;
128129
std::atomic_bool done{false};
129130
std::atomic_bool startup{true};
130-
std::optional<chain::chain_id_type> chain_id;
131+
std::optional <chain::chain_id_type> chain_id;
131132
fc::microseconds abi_serializer_max_time;
132133

133134
static const account_name newaccount;
@@ -154,7 +155,7 @@ namespace eosio {
154155

155156
template<typename Queue, typename Entry>
156157
void kafka_plugin_impl::queue(Queue &queue, const Entry &e) {
157-
std::unique_lock<std::mutex> lock(mtx);
158+
std::unique_lock <std::mutex> lock(mtx);
158159
auto queue_size = queue.size();
159160
if (queue_size > max_queue_size) {
160161
lock.unlock();
@@ -193,7 +194,6 @@ namespace eosio {
193194
trasaction_info_st transactioninfo = trasaction_info_st{
194195
.block_number = t->block_num,//chain.pending_block_state()->block_num,
195196
.chain_id = this->chain_id,
196-
// .block_time = chain.pending_block_time(),
197197
.trace =chain::transaction_trace_ptr(t),
198198
};
199199
transactioninfo.block_time = chain.pending_block_time();
@@ -238,7 +238,7 @@ namespace eosio {
238238
try {
239239

240240
while (true) {
241-
std::unique_lock<std::mutex> lock(mtx);
241+
std::unique_lock <std::mutex> lock(mtx);
242242
while (transaction_metadata_queue.empty() &&
243243
transaction_trace_queue.empty() &&
244244
block_state_queue.empty() &&
@@ -409,8 +409,9 @@ namespace eosio {
409409
filter_traction_trace(t.trace, name("transfer"));
410410
if (t.trace->action_traces.size() > 0) {
411411
string transfer_json =
412-
"{\"block_number\":" + std::to_string(t.block_number) + ",\"block_time\":" + std::to_string(time) +
413-
",\"chain_id\":" + "\""+ t.chain_id->str() + "\""+
412+
"{\"block_number\":" + std::to_string(t.block_number) + ",\"block_time\":" +
413+
std::to_string(time) +
414+
",\"chain_id\":" + "\"" + t.chain_id->str() + "\"" +
414415
",\"trace\":" + fc::json::to_string(t.trace, fc::time_point::maximum()).c_str() + "}";
415416
producer->trx_kafka_sendmsg(KAFKA_TRX_TRANSFER, (char *) transfer_json.c_str());
416417
//elog("transfer_json = ${e}",("e",transfer_json));
@@ -523,8 +524,7 @@ namespace eosio {
523524
("kafka-block-start", bpo::value<uint32_t>()->default_value(256),
524525
"If specified then only abi data pushed to kafka until specified block is reached.")
525526
("kafka-compression-codec", bpo::value<std::string>(),
526-
"Compression codec to use for compressing message sets. This is the default value for all topics, may be overriden by the topic configuration property compression.codec.(none, gzip, snappy, lz4)")
527-
;
527+
"Compression codec to use for compressing message sets. This is the default value for all topics, may be overriden by the topic configuration property compression.codec.(none, gzip, snappy, lz4)");
528528

529529
}
530530

@@ -550,13 +550,14 @@ namespace eosio {
550550
transfer_trx_topic = (char *) (options.at("transfer_trx_topic").as<std::string>().c_str());
551551
elog("transfer_trx_topic:${j}", ("j", transfer_trx_topic));
552552
}
553-
if (options.count("kafka-compression-codec") != 0) {
553+
if (options.count("kafka-compression-codec") != 0) {
554554
compression_codec = (char *) (options.at("kafka-compression-codec").as<std::string>().c_str());
555555
elog("kafka-compression-codec:${j}", ("j", compression_codec));
556556
}
557557

558-
if (0 != my->producer->trx_kafka_init(brokers_str, accept_trx_topic, compression_codec, applied_trx_topic,
559-
transfer_trx_topic)) {
558+
if (0 !=
559+
my->producer->trx_kafka_init(brokers_str, accept_trx_topic, compression_codec, applied_trx_topic,
560+
transfer_trx_topic)) {
560561
elog("trx_kafka_init fail");
561562
} else {
562563
elog("trx_kafka_init ok");
@@ -600,7 +601,7 @@ namespace eosio {
600601
}));
601602
my->applied_transaction_connection.emplace(
602603
chain.applied_transaction.connect(
603-
[&](std::tuple<const std::shared_ptr<chain::transaction_trace>&, const std::shared_ptr<const chain::packed_transaction>&> t) {
604+
[&](std::tuple<const std::shared_ptr <chain::transaction_trace> &, const std::shared_ptr<const chain::packed_transaction> &> t) {
604605
my->applied_transaction(std::get<0>(t));
605606
}));
606607
my->init();

0 commit comments

Comments
 (0)