0902horn commented on code in PR #809:
URL: https://github.com/apache/rocketmq-clients/pull/809#discussion_r1701482088
##########
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:
This line was for debugging and should removed.
--
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: commits-unsubscr...@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
