charlely commented on code in PR #6612: URL: https://github.com/apache/inlong/pull/6612#discussion_r1032117240
########## inlong-tubemq/tubemq-client-twins/tubemq-client-cpp/src/baseproducer.cc: ########## @@ -0,0 +1,796 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "baseproducer.h" + +#include <unistd.h> + +#include <iostream> +#include <sstream> + +#include "const_config.h" +#include "const_rpc.h" +#include "singleton.h" +#include "transport.h" +#include "tubemq/tubemq_errcode.h" +#include "tubemq_transport.h" +#include "utils.h" +#include "version.h" + +namespace tubemq { + +using std::stringstream; + +BaseProducer::BaseProducer() : BaseClient(true) { + status_.Set(0); + masters_map_.clear(); + is_master_actived_.Set(false); + master_reg_status_.Set(0); + master_hb_status_.Set(0); + broker_info_checksum_ = -1; + master_sh_retry_cnt_ = 0; +} + +BaseProducer::~BaseProducer() {} + +bool BaseProducer::Start(string& err_info, const ProducerConfig& config) { + if (!status_.CompareAndSet(0, 1)) { + err_info = "Ok"; + return true; + } + + // check master addr info + if (config.GetMasterAddrInfo().length() == 0) { + err_info = "Parameter error: not set master address info!"; + status_.CompareAndSet(1, 0); + return false; + } + + if (!TubeMQService::Instance()->IsRunning()) { + err_info = "TubeMQ Service not startted!"; + status_.CompareAndSet(1, 0); + return false; + } + + // add to TubeMQService + if (!TubeMQService::Instance()->AddClientObj(err_info, shared_from_this())) { + printf("Add to tubemq service failed!!!\n"); + client_index_ = tb_config::kInvalidValue; + status_.CompareAndSet(1, 0); + return false; + } + + config_ = config; + if (!initMasterAddress(err_info, config.GetMasterAddrInfo())) { + TubeMQService::Instance()->RmvClientObj(shared_from_this()); + status_.CompareAndSet(1, 0); + return false; + } + + // set client_uuid_ + client_uuid_ = buildUUID(); + + // producer register to master + int32_t error_code; + if (!register2Master(error_code, err_info, false)) { + TubeMQService::Instance()->RmvClientObj(shared_from_this()); + status_.CompareAndSet(1, 0); + return false; + } + status_.CompareAndSet(1, 2); // register2Master done, change status_ to `2` + + // set heartbeat timer + heart_beat_timer_ = TubeMQService::Instance()->CreateTimer(); + heart_beat_timer_->expires_after(std::chrono::milliseconds(config_.GetHeartbeatPeriodMs() / 2)); + auto self = shared_from_this(); + heart_beat_timer_->async_wait([self, this](const std::error_code& ec) { + if (ec) { + return; + } + heartBeat2Master(); + }); + + return true; +} + +void BaseProducer::ShutDown() { + if (!status_.CompareAndSet(2, 0)) { + return; + } + + close2Master(); + + TubeMQService::Instance()->RmvClientObj(shared_from_this()); + client_index_ = tb_config::kInvalidValue; +} + +bool BaseProducer::Publish(string& err_info, const string& topic) { + std::lock_guard<std::mutex> lck(pub_topic_list_lock_); + pub_topic_list_[topic] = 1; + err_info = "OK"; + return true; +} + +bool BaseProducer::Publish(string& err_info, const set<string>& topic_list) { + std::lock_guard<std::mutex> lck(pub_topic_list_lock_); + for (const string& topic_name : topic_list) { + pub_topic_list_[topic_name] = 1; + } + err_info = "OK"; + return true; +} + +bool BaseProducer::SendMessage(string& err_info, const Message& message, bool is_sync, + const std::function<void(const ErrorCode&)>& callback) { + Partition partition = selectPartition(message); + std::string broker_id = std::to_string(partition.GetBrokerId()); + + auto request = std::make_shared<RequestContext>(); + TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol(); + buildSendMessageRequestP2B(partition, message, req_protocol); + string target_ip; + int target_port; + getCurrentMasterAddr(target_ip, target_port); + request->codec_ = std::make_shared<TubeMQCodec>(); + request->ip_ = brokers_map_[broker_id].GetHost(); + request->port_ = brokers_map_[broker_id].GetPort(); + request->timeout_ = config_.GetRpcReadTimeoutMs(); + request->request_id_ = Singleton<UniqueSeqId>::Instance().Next(); + req_protocol->request_id_ = request->request_id_; + req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs(); + + if (is_sync) { + ResponseContext response_context; + ErrorCode error = SyncRequest(response_context, request, req_protocol); + if (error.Value() != err_code::kErrSuccess) { + std::cout << "*** err Code = " << error.Value() << ", err Msg = " << error.Message() + << std::endl; + return false; + } + int32_t error_code = 0; + std::string err_msg; + auto rsp = any_cast<TubeMQCodec::RspProtocolPtr>(response_context.rsp_); + bool res = processSendMessageResponseB2P(error_code, err_msg, rsp); + if (!res) { + std::cout << "*** Sync Request return from tubemq server, error_code = " << error_code + << ", err_msg = " << err_msg << std::endl; + return false; + } + } else { + AsyncRequest(request, req_protocol) + .AddCallBack([=](ErrorCode error, const ResponseContext& response_context) { + if (error.Value() == err_code::kErrSuccess) { + auto rsp = any_cast<TubeMQCodec::RspProtocolPtr>(response_context.rsp_); + int32_t error_code = 0; + std::string err_msg; + bool res = processSendMessageResponseB2P(error_code, err_msg, rsp); + if (!res) { + LOG_ERROR("Send Message to Broker successully, but response is not successful!!!"); + std::cout << "*** return from tubemq server, error_code = " << error_code + << ", err_msg = " << err_msg << std::endl; + } + } else { + LOG_ERROR("Async Send Message to Broker failed!!!"); + } + callback(error); + }); + } + + return true; +} + +bool BaseProducer::register2Master(int32_t& error_code, string& err_info, bool need_change) { + // set regist process status to begin + if (!master_reg_status_.CompareAndSet(0, 1)) { + err_info = "register2Master process has began!"; + return false; + } + + LOG_INFO("[PRODUCER] register2Master process begin:"); + + string target_ip; + int target_port; + getCurrentMasterAddr(target_ip, target_port); + bool result = false; + int retry_count = 0; + int max_retry_count = masters_map_.size(); + err_info = "Master register failure, no online master service!"; + while (retry_count < max_retry_count) { + if (!TubeMQService::Instance()->IsRunning()) { + err_info = "TubeMQ Service is stopped!"; + master_reg_status_.CompareAndSet(1, 0); + return false; + } + // Construct the request + auto request = std::make_shared<RequestContext>(); + TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol(); + buildRegisterRequestP2M(req_protocol); + // set parameters + request->codec_ = std::make_shared<TubeMQCodec>(); + request->ip_ = target_ip; + request->port_ = target_port; + request->timeout_ = config_.GetRpcReadTimeoutMs(); + request->request_id_ = Singleton<UniqueSeqId>::Instance().Next(); + req_protocol->request_id_ = request->request_id_; + req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs() - 500; + // send request sync + ResponseContext response_context; + ErrorCode error = SyncRequest(response_context, request, req_protocol); + LOG_INFO("register2Master response come, error.value is %d", error.Value()); + if (error.Value() == err_code::kErrSuccess) { + // process response + auto rsp = any_cast<TubeMQCodec::RspProtocolPtr>(response_context.rsp_); + result = processRegisterResponseM2P(error_code, err_info, rsp); + if (result) { + err_info = "Ok"; + is_master_actived_.Set(true); + break; + } else { + is_master_actived_.Set(false); + } + } + getNextMasterAddr(target_ip, target_port); + retry_count++; + } + if (result) { + printf("Producer register2Master successfully, the registered mater is: %s:%d\n", + target_ip.c_str(), target_port); + } + return result; +} + +void BaseProducer::heartBeat2Master() { + // check status + if (!master_hb_status_.CompareAndSet(0, 1)) { + LOG_INFO("check hb process status, heartBeat2Master process has began!"); + return; + } + + // check the service is running + if (!TubeMQService::Instance()->IsRunning()) { + master_hb_status_.CompareAndSet(1, 0); + LOG_INFO("[PRODUCER] heartBeat2Master failure: TubeMQ Service is stopped! client=%s", + client_uuid_.c_str()); + return; + } + + // check the status of client, wheter register + if (!isClientRunning()) { + master_hb_status_.CompareAndSet(1, 0); + LOG_INFO("[PRODUCER] heartBeat2Master failure: TubeMQ Client stopped! client=%s", + client_uuid_.c_str()); + return; + } + + // todo, check master is actived + + string target_ip; + int target_port; + getCurrentMasterAddr(target_ip, target_port); + auto request = std::make_shared<RequestContext>(); + TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol(); + // build heartbeat 2 master request + buildHeartRequestP2M(req_protocol); + request->codec_ = std::make_shared<TubeMQCodec>(); + request->ip_ = target_ip; + request->port_ = target_port; + request->timeout_ = config_.GetRpcReadTimeoutMs(); + request->request_id_ = Singleton<UniqueSeqId>::Instance().Next(); + req_protocol->request_id_ = request->request_id_; + req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs() - 500; + // async send request + AsyncRequest(request, req_protocol) + .AddCallBack([=](ErrorCode error, const ResponseContext& response_context) { + if (GetClientIndex() == tb_config::kInvalidValue || + !TubeMQService::Instance()->IsRunning() || !isClientRunning()) { + master_hb_status_.CompareAndSet(1, 0); + return; + } + + if (error.Value() != err_code::kErrSuccess) { + master_sh_retry_cnt_++; + LOG_WARN("[PRODUCER] heartBeat2Master failue to (%s:%d) : %s, client=%s", + target_ip.c_str(), target_port, error.Message().c_str(), client_uuid_.c_str()); + if (master_sh_retry_cnt_ >= tb_config::kMaxMasterHBRetryCount) { + LOG_WARN("[PRODUCER] heartBeat2Master found over max-hb-retry(%d), client=%s", + master_sh_retry_cnt_, client_uuid_.c_str()); + master_sh_retry_cnt_ = 0; + is_master_actived_.Set(false); + // asyncRegister2Master(true); + master_hb_status_.CompareAndSet(1, 0); + return; + } + } else { + // process response + auto rsp = any_cast<TubeMQCodec::RspProtocolPtr>(response_context.rsp_); + int32_t error_code = 0; + std::string error_info; + bool ret_result = processHBResponseM2P(error_code, error_info, rsp); + std::cout << "*** Process ret_result = " << ret_result << std::endl; + if (ret_result) { + is_master_actived_.Set(true); + master_sh_retry_cnt_ = 0; + } else { + master_sh_retry_cnt_++; + if (error_code == err_code::kErrHbNoNode || + error_info.find("StandbyException") != string::npos) { + is_master_actived_.Set(false); + bool need_change = !(error_code == err_code::kErrHbNoNode); + LOG_WARN("[PRODUCER] heartBeat2Master found no-node or standby, client=%s, change=%d", + client_uuid_.c_str(), need_change); + // asyncRegister2Master(need_change); + master_hb_status_.CompareAndSet(1, 0); + return; + } + } + } + if (GetClientIndex() == tb_config::kInvalidValue || + !TubeMQService::Instance()->IsRunning() || !isClientRunning()) { + master_hb_status_.CompareAndSet(1, 0); + return; + } + heart_beat_timer_->expires_after(std::chrono::milliseconds(2000)); + auto self = shared_from_this(); + heart_beat_timer_->async_wait([self, this](const std::error_code& ec) { + if (ec) { + return; + } + heartBeat2Master(); + }); + master_hb_status_.CompareAndSet(1, 0); + }); + return; +} + +void BaseProducer::close2Master() { + string target_ip; + int target_port; + getCurrentMasterAddr(target_ip, target_port); + LOG_INFO("[PRODUCER] close2Master begin, clientid=%s", client_uuid_.c_str()); + auto request = std::make_shared<RequestContext>(); + TubeMQCodec::ReqProtocolPtr req_protocol = TubeMQCodec::GetReqProtocol(); + request->codec_ = std::make_shared<TubeMQCodec>(); + request->ip_ = target_ip; + request->port_ = target_port; + request->timeout_ = config_.GetRpcReadTimeoutMs(); + request->request_id_ = Singleton<UniqueSeqId>::Instance().Next(); + req_protocol->request_id_ = request->request_id_; + req_protocol->rpc_read_timeout_ = config_.GetRpcReadTimeoutMs() - 500; + // send message to target + AsyncRequest(request, req_protocol); + LOG_INFO("[PRODUCER] close2Master finished, clientid=%s", client_uuid_.c_str()); + // not need wait response + return; +} + +string BaseProducer::buildUUID() { + stringstream ss; + ss << "PRODUCER_"; // producer client doesn't have `group_name`, use constant string instead + ss << TubeMQService::Instance()->GetLocalHost(); + ss << "-"; + ss << getpid(); + ss << "-"; + ss << Utils::GetCurrentTimeMillis(); + ss << "-"; + ss << GetClientIndex(); + ss << "-"; + ss << kTubeMQClientVersion; + return ss.str(); +} + +bool BaseProducer::isClientRunning() { return (status_.Get() == 2); } Review Comment: Status can use const to define ? I cann't understand 2 status. -- 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]
