lizhanhui commented on code in PR #809:
URL: https://github.com/apache/rocketmq-clients/pull/809#discussion_r1701293802
##########
cpp/source/client/TelemetryBidiReactor.cpp:
##########
@@ -383,134 +279,61 @@ void TelemetryBidiReactor::write(TelemetryCommand
command) {
void TelemetryBidiReactor::tryWriteNext() {
SPDLOG_DEBUG("{}#tryWriteNext", peer_address_);
- {
- absl::MutexLock lk(&state_mtx_);
- if (StreamState::Error == write_state_ || StreamState::Closed ==
write_state_) {
- SPDLOG_WARN("Further write to {} is not allowded due to
stream-state={}", peer_address_,
- static_cast<std::uint8_t>(write_state_));
- return;
- }
+ absl::MutexLock lk(&writes_mtx_);
+ if (StreamState::Ready != state_) {
+ SPDLOG_WARN("Further write to {} is not allowed due to stream-state={}",
peer_address_,
+ static_cast<std::uint8_t>(state_));
+ return;
+ }
+ if (writes_.empty()) {
+ SPDLOG_DEBUG("No TelemetryCommand to write. Peer={}", peer_address_);
+ return;
}
- {
- absl::MutexLock lk(&writes_mtx_);
- if (writes_.empty() && StreamState::Closing != write_state_) {
- SPDLOG_DEBUG("No TelemtryCommand to write. Peer={}", peer_address_);
- return;
- }
-
- if (StreamState::Ready == write_state_) {
- write_state_ = StreamState::Inflight;
- }
-
- if (writes_.empty()) {
- // Tell server there is no more write requests.
- StartWritesDone();
- } else {
- SPDLOG_DEBUG("Writing telemetry command to {}: {}", peer_address_,
writes_.front().DebugString());
- StartWrite(&(writes_.front()));
- }
+ if (!writes_.empty()) {
+ SPDLOG_DEBUG("Writing telemetry command to {}: {}", peer_address_,
writes_.front().DebugString());
+ AddHold();
+ StartWrite(&(writes_.front()));
}
}
-void TelemetryBidiReactor::fireClose() {
+void TelemetryBidiReactor::signalClose() {
+ absl::MutexLock lk(&state_mtx_);
+ state_ = StreamState::Closing;
+}
+
+void TelemetryBidiReactor::close() {
SPDLOG_INFO("{}#fireClose", peer_address_);
+ if (peer_address_.rfind(":8080") != std::string::npos) {
Review Comment:
Why we need this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
