-
Notifications
You must be signed in to change notification settings - Fork 297
Feature/tx status bus #1575
Feature/tx status bus #1575
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -36,23 +36,18 @@ namespace torii { | |
CommandService::CommandService( | ||
std::shared_ptr<iroha::torii::TransactionProcessor> tx_processor, | ||
std::shared_ptr<iroha::ametsuchi::Storage> storage, | ||
std::shared_ptr<iroha::torii::StatusBus> status_bus, | ||
std::chrono::milliseconds initial_timeout, | ||
std::chrono::milliseconds nonfinal_timeout) | ||
: tx_processor_(tx_processor), | ||
storage_(storage), | ||
status_bus_(status_bus), | ||
initial_timeout_(initial_timeout), | ||
nonfinal_timeout_(nonfinal_timeout), | ||
cache_(std::make_shared<CacheType>()), | ||
// merge with mutex, since notifications can be made from different | ||
// threads | ||
// TODO 11.07.2018 andrei rework status handling with event bus IR-1517 | ||
responses_(tx_processor_->transactionNotifier().merge( | ||
rxcpp::serialize_one_worker( | ||
rxcpp::schedulers::make_current_thread()), | ||
stateless_notifier_.get_observable())), | ||
log_(logger::log("CommandService")) { | ||
// Notifier for all clients | ||
responses_.subscribe([this](auto iroha_response) { | ||
status_bus_->statuses().subscribe([this](auto iroha_response) { | ||
// find response for this tx in cache; if status of received response | ||
// isn't "greater" than cached one, dismiss received one | ||
auto proto_response = | ||
|
@@ -69,6 +64,17 @@ namespace torii { | |
}); | ||
} | ||
|
||
namespace { | ||
iroha::protocol::ToriiResponse makeResponse( | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Add documentation for the method. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Isn't it self-explanatory? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. But the purpose is hidden. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. anonymous namespace make a clue that it's for internal usage |
||
const shared_model::crypto::Hash &h, | ||
const iroha::protocol::TxStatus &status) { | ||
iroha::protocol::ToriiResponse response; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why are we using raw proto instead of the model? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Because it used for grpc |
||
response.set_tx_hash(shared_model::crypto::toBinaryString(h)); | ||
response.set_tx_status(status); | ||
return response; | ||
} | ||
} // namespace | ||
|
||
void CommandService::Torii(const iroha::protocol::Transaction &request) { | ||
shared_model::proto::TransportBuilder< | ||
shared_model::proto::Transaction, | ||
|
@@ -86,20 +92,17 @@ namespace torii { | |
return; | ||
} | ||
|
||
// setting response | ||
iroha::protocol::ToriiResponse response; | ||
response.set_tx_hash( | ||
shared_model::crypto::toBinaryString(tx_hash)); | ||
response.set_tx_status( | ||
iroha::protocol::TxStatus::STATELESS_VALIDATION_SUCCESS); | ||
|
||
// Send transaction to iroha | ||
tx_processor_->transactionHandle( | ||
std::make_shared<shared_model::proto::Transaction>( | ||
std::move(iroha_tx.value))); | ||
|
||
this->addTxToCacheAndLog( | ||
"Torii", std::move(tx_hash), std::move(response)); | ||
this->pushStatus( | ||
"Torii", | ||
std::move(tx_hash), | ||
makeResponse( | ||
tx_hash, | ||
iroha::protocol::TxStatus::STATELESS_VALIDATION_SUCCESS)); | ||
}, | ||
[this, &request](const auto &error) { | ||
// getting hash from invalid transaction | ||
|
@@ -113,14 +116,12 @@ namespace torii { | |
tx_hash.hex()); | ||
|
||
// setting response | ||
iroha::protocol::ToriiResponse response; | ||
response.set_tx_hash( | ||
shared_model::crypto::toBinaryString(tx_hash)); | ||
response.set_tx_status( | ||
auto response = makeResponse( | ||
tx_hash, | ||
iroha::protocol::TxStatus::STATELESS_VALIDATION_FAILED); | ||
response.set_error_message(std::move(error.error)); | ||
|
||
this->addTxToCacheAndLog( | ||
this->pushStatus( | ||
"Torii", std::move(tx_hash), std::move(response)); | ||
}); | ||
} | ||
|
@@ -145,18 +146,15 @@ namespace torii { | |
return; | ||
} | ||
|
||
// setting response | ||
iroha::protocol::ToriiResponse response; | ||
response.set_tx_hash( | ||
shared_model::crypto::toBinaryString(tx_hash)); | ||
response.set_tx_status( | ||
iroha::protocol::TxStatus::STATELESS_VALIDATION_SUCCESS); | ||
|
||
// Send transaction to iroha | ||
tx_processor_->transactionHandle(tx); | ||
|
||
this->addTxToCacheAndLog( | ||
"ToriiList", std::move(tx_hash), std::move(response)); | ||
this->pushStatus( | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Here and above there is a slight duplication, maybe put it in a function? |
||
"ToriiList", | ||
std::move(tx_hash), | ||
makeResponse(tx_hash, | ||
iroha::protocol::TxStatus:: | ||
STATELESS_VALIDATION_SUCCESS)); | ||
}); | ||
}, | ||
[this, &tx_list](auto &error) { | ||
|
@@ -189,14 +187,12 @@ namespace torii { | |
shared_model::crypto::DefaultHashProvider::makeHash( | ||
shared_model::proto::makeBlob(tx.payload())); | ||
|
||
iroha::protocol::ToriiResponse response; | ||
response.set_tx_hash( | ||
shared_model::crypto::toBinaryString(hash)); | ||
response.set_tx_status( | ||
auto response = makeResponse( | ||
hash, | ||
iroha::protocol::TxStatus::STATELESS_VALIDATION_FAILED); | ||
response.set_error_message(sequence_error); | ||
|
||
this->addTxToCacheAndLog( | ||
this->pushStatus( | ||
"ToriiList", std::move(hash), std::move(response)); | ||
}); | ||
}); | ||
|
@@ -225,16 +221,15 @@ namespace torii { | |
response.CopyFrom(*resp); | ||
} else { | ||
response.set_tx_hash(request.tx_hash()); | ||
if (storage_->getBlockQuery()->hasTxWithHash( | ||
shared_model::crypto::Hash(request.tx_hash()))) { | ||
auto hash = shared_model::crypto::Hash(request.tx_hash()); | ||
if (storage_->getBlockQuery()->hasTxWithHash(hash)) { | ||
response.set_tx_status(iroha::protocol::TxStatus::COMMITTED); | ||
cache_->addItem(std::move(hash), response); | ||
} else { | ||
log_->warn("Asked non-existing tx: {}", | ||
iroha::bytestringToHexstring(request.tx_hash())); | ||
response.set_tx_status(iroha::protocol::TxStatus::NOT_RECEIVED); | ||
} | ||
this->addTxToCacheAndLog( | ||
"Status", std::move(tx_hash), std::move(response)); | ||
} | ||
} | ||
|
||
|
@@ -274,7 +269,8 @@ namespace torii { | |
.build() | ||
.getTransport(); | ||
}()))); | ||
return responses_ | ||
return status_bus_ | ||
->statuses() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is this code is well-formed with clang-format? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. yes |
||
// prepend initial status | ||
.start_with(initial_status) | ||
// select statuses with requested hash | ||
|
@@ -368,18 +364,15 @@ namespace torii { | |
} | ||
} | ||
|
||
void CommandService::addTxToCacheAndLog( | ||
void CommandService::pushStatus( | ||
const std::string &who, | ||
const shared_model::crypto::Hash &hash, | ||
const iroha::protocol::ToriiResponse &response) { | ||
log_->debug("{}: adding item to cache: {}, status {} ", | ||
who, | ||
hash.hex(), | ||
response.tx_status()); | ||
// transactions can be handled from multiple threads, therefore a lock is | ||
// required | ||
std::lock_guard<std::mutex> lock(stateless_tx_status_notifier_mutex_); | ||
stateless_notifier_.get_subscriber().on_next( | ||
status_bus_->publish( | ||
std::make_shared<shared_model::proto::TransactionResponse>( | ||
std::move(response))); | ||
} | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Add comment please