Skip to content
This repository has been archived by the owner on Apr 17, 2019. It is now read-only.

Rewrite StatusStream with rxcpp #1541

Merged
merged 8 commits into from
Jul 13, 2018
Merged

Rewrite StatusStream with rxcpp #1541

merged 8 commits into from
Jul 13, 2018

Conversation

lebdron
Copy link
Contributor

@lebdron lebdron commented Jul 5, 2018

Signed-off-by: Andrei Lebedev lebdron@gmail.com

Description of the Change

Rewrite StatusStream method without manual synchronization with condition variables and mutexes using rxcpp

Benefits

Separate status mapping and transport logic

Possible Drawbacks

Custom rxcpp timeout operator

@lebdron
Rewrite StatusStream with rxcpp
3e7ef03 
Signed-off-by: Andrei Lebedev <lebdron@gmail.com>
@lebdron lebdron requested review from muratovv and vdrobnyi July 5, 2018 14:20
@lebdron
Merge branch 'develop' into refactor/torii-statusstream-rxcpp
0614fdd 
Signed-off-by: Andrei Lebedev <lebdron@gmail.com>

# Conflicts:
#	irohad/torii/impl/command_service.cpp
@l4l l4l added needs-review pr awaits review from maintainers refactoring internal stuff, that are changed/removed that doesn't affect client code labels Jul 6, 2018
@l4l l4l self-requested a review July 6, 2018 15:46
@lebdron
Minor fixes for types and logger calls
9e98991 
Signed-off-by: Andrei Lebedev <lebdron@gmail.com>
l4l
l4l approved these changes Jul 6, 2018
View reviewed changes
irohad/torii/impl/command_service.cpp Outdated
response_writer.Write(resp_sub);
}
rxcpp::observable<
std::shared_ptr<shared_model::interface::TransactionResponse>>
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This type used multiple times, maybe create alias, e.g using Response = std::shared_ptr<shared_model::interface::TransactionResponse>;

irohad/torii/impl/command_service.cpp Outdated
: 2 * proposal_delay_;
},
current_thread))
.take_while([=](const auto &) { return not context->IsCancelled(); })
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Might be useful to add logging here as well

irohad/torii/impl/timeout.hpp Outdated
/**
* Copyright Soramitsu Co., Ltd. All Rights Reserved.
* SPDX-License-Identifier: Apache-2.0
*/
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess there should be one more license

irohad/torii/impl/timeout.hpp Outdated
typedef typename coordination_type::coordinator_type coordinator_type;
typedef rxcpp::util::decay_t<Selector> select_type;
typedef decltype(
(*(select_type *)nullptr)(*(source_value_type *)nullptr)) value_type;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it used?
Also declval might be applicable

irohad/torii/impl/timeout.hpp Outdated
if (selectedWork.empty()) {
return;
}
localState->worker.schedule(selectedWork.get());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe invert if above and move this line up?

muratovv
muratovv previously requested changes Jul 9, 2018
View reviewed changes
irohad/torii/command_service.hpp Outdated
std::mutex notifier_mutex_;
rxcpp::subjects::subject<
std::shared_ptr<shared_model::interface::TransactionResponse>>
notifier_;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The name is not so meaningful, maybe rename or add some documentation.

irohad/torii/impl/command_service.cpp Outdated
@@ -44,9 +46,15 @@ namespace torii {
proposal_delay_(proposal_delay),
start_tx_processing_duration_(1s),
cache_(std::make_shared<CacheType>()),
// merge with mutex, since notifications can be made from different
// threads
responses_(tx_processor_->transactionNotifier().merge(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pls, add TODO about event bus rework.

irohad/torii/impl/command_service.cpp Outdated
return responses_.start_with(initial_status)
.filter(
[&](auto response) { return response->transactionHash() == hash; })
.lift<std::shared_ptr<shared_model::interface::TransactionResponse>>(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think, better to add some comments about what happened here.

irohad/torii/impl/command_service.cpp Outdated
return response.tx_status()
== iroha::protocol::TxStatus::NOT_RECEIVED
? start_tx_processing_duration_
: 2 * proposal_delay_;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

maybe move out of here configuration logic?

irohad/torii/impl/command_service.cpp Outdated
.subscribe(
subscription,
[&](iroha::protocol::ToriiResponse response) {
log_->debug("writing status");
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Move the debug after Write method.

irohad/torii/impl/command_service.cpp Outdated
request_hash->hex());
// run loop while subscription is active or there are pending events in the
// queue
while (subscription.is_subscribed() or not rl.empty()) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

better to wrap this function with an inline method for the improving readability.

irohad/torii/impl/timeout.hpp Outdated
namespace torii {

/**
* Return an observable that terminates with timeout_error if a particular
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe better to write here about "This class is copypaste of ..." for reducing misleading usages of code as an example.

irohad/torii/impl/timeout.hpp Outdated

#include <rxcpp/operators/rx-timeout.hpp>

namespace torii {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe move this file to lib for sharing usage?

@lebdron lebdron dismissed muratovv’s stale review July 11, 2018 17:06

Fixes are attempted

@lebdron lebdron requested a review from muratovv July 11, 2018 17:06
muratovv
muratovv approved these changes Jul 12, 2018
View reviewed changes
irohad/torii/impl/command_service.cpp Outdated
.take_while([=](const auto &) {
auto is_cancelled = context->IsCancelled();
if (is_cancelled) {
log_->debug("client unsubscribed");
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe specify here some identify information about a client?

irohad/torii/impl/command_service.cpp Outdated
log_->debug("writing status");
response_writer->Write(response);
if (response_writer->Write(response)) {
log_->debug("status written");
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same as above

@lebdron lebdron merged commit 9958bc9 into develop Jul 13, 2018
@lebdron lebdron deleted the refactor/torii-statusstream-rxcpp branch July 13, 2018 09:42
@neewy neewy mentioned this pull request Jul 25, 2018
Merged
l4l pushed a commit that referenced this pull request Jul 25, 2018
@lebdron @l4l
Rewrite StatusStream with rxcpp (#1541)
ebcb23d 
* Rewrite StatusStream with rxcpp

* Minor fixes for types and logger calls

* Review fixes; fix irohad_test using wrong example path

* Fix tx hash setter in stateless valid case

* Add client id to debug log

Signed-off-by: Andrei Lebedev <lebdron@gmail.com>
l4l pushed a commit that referenced this pull request Jul 25, 2018
@lebdron @l4l
Rewrite StatusStream with rxcpp (#1541)
2bbf0c5 
* Rewrite StatusStream with rxcpp

* Minor fixes for types and logger calls

* Review fixes; fix irohad_test using wrong example path

* Fix tx hash setter in stateless valid case

* Add client id to debug log

Signed-off-by: Andrei Lebedev <lebdron@gmail.com>
l4l pushed a commit that referenced this pull request Jul 25, 2018
@lebdron @l4l
Rewrite StatusStream with rxcpp (#1541)
cbd433a 
* Rewrite StatusStream with rxcpp

* Minor fixes for types and logger calls

* Review fixes; fix irohad_test using wrong example path

* Fix tx hash setter in stateless valid case

* Add client id to debug log

Signed-off-by: Andrei Lebedev <lebdron@gmail.com>
l4l pushed a commit that referenced this pull request Jul 25, 2018
@lebdron @l4l
Rewrite StatusStream with rxcpp (#1541)
9ae5bb7 
* Rewrite StatusStream with rxcpp

* Minor fixes for types and logger calls

* Review fixes; fix irohad_test using wrong example path

* Fix tx hash setter in stateless valid case

* Add client id to debug log

Signed-off-by: Andrei Lebedev <lebdron@gmail.com>
l4l pushed a commit that referenced this pull request Jul 25, 2018
@lebdron @l4l
Rewrite StatusStream with rxcpp (#1541)
617c496 
* Rewrite StatusStream with rxcpp

* Minor fixes for types and logger calls

* Review fixes; fix irohad_test using wrong example path

* Fix tx hash setter in stateless valid case

* Add client id to debug log

Signed-off-by: Andrei Lebedev <lebdron@gmail.com>
Sign up for free to subscribe to this conversation on GitHub. Already have an account? Sign in.
Reviewers

@l4l l4l l4l approved these changes

@muratovv muratovv muratovv approved these changes

@vdrobnyi vdrobnyi Awaiting requested review from vdrobnyi

Assignees
No one assigned
Labels
needs-review pr awaits review from maintainers refactoring internal stuff, that are changed/removed that doesn't affect client code
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.

None yet
3 participants
@lebdron @l4l @muratovv