You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2022/07/14 08:58:19 UTC
[rocketmq-clients] branch master updated: Remove deprecated features (#48)
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new 419cf0e Remove deprecated features (#48)
419cf0e is described below
commit 419cf0e6850491c4d679bb234a0f2e0dc1fa3adc
Author: Zhanhui Li <li...@gmail.com>
AuthorDate: Thu Jul 14 16:58:15 2022 +0800
Remove deprecated features (#48)
* Remove deprecated features
* Remove manual metrics collection
* Remove consume rate limiter
* Remove unused headers
---
cpp/bazel/curl.bzl | 201 ---------------------
cpp/bazel/rocketmq_deps.bzl | 12 --
cpp/src/main/cpp/base/BUILD.bazel | 1 -
cpp/src/main/cpp/base/CredentialsProvider.cpp | 98 +---------
cpp/src/main/cpp/base/HostInfo.cpp | 77 --------
cpp/src/main/cpp/base/HttpClientImpl.cpp | 90 ---------
cpp/src/main/cpp/base/RateLimiter.cpp | 54 ------
cpp/src/main/cpp/base/TopAddressing.cpp | 76 --------
cpp/src/main/cpp/base/include/Histogram.h | 85 ---------
cpp/src/main/cpp/base/include/HostInfo.h | 47 -----
cpp/src/main/cpp/base/include/HttpClient.h | 53 ------
cpp/src/main/cpp/base/include/HttpClientImpl.h | 46 -----
cpp/src/main/cpp/base/include/RateLimiter.h | 184 -------------------
.../cpp/base/include/StsCredentialsProviderImpl.h | 61 -------
cpp/src/main/cpp/base/include/TopAddressing.h | 52 ------
cpp/src/main/cpp/client/ClientManagerImpl.cpp | 45 +----
cpp/src/main/cpp/client/include/ClientManager.h | 1 -
.../main/cpp/client/include/ClientManagerImpl.h | 64 ++++---
.../cpp/client/include/HeartbeatDataCallback.h | 31 ----
cpp/src/main/cpp/client/include/Topic.h | 64 -------
.../cpp/client/include/TopicRouteChangeCallback.h | 30 ---
cpp/src/main/cpp/rocketmq/ClientImpl.cpp | 1 -
.../cpp/rocketmq/DynamicNameServerResolver.cpp | 129 -------------
cpp/src/main/cpp/rocketmq/Producer.cpp | 1 -
cpp/src/main/cpp/rocketmq/PushConsumer.cpp | 1 -
.../rocketmq/include/ConsumeMessageServiceImpl.h | 1 -
.../rocketmq/include/DynamicNameServerResolver.h | 73 --------
cpp/third_party/cpp_httplib.BUILD | 17 --
cpp/third_party/curl.BUILD | 35 ----
29 files changed, 40 insertions(+), 1590 deletions(-)
diff --git a/cpp/bazel/curl.bzl b/cpp/bazel/curl.bzl
deleted file mode 100644
index b83d054..0000000
--- a/cpp/bazel/curl.bzl
+++ /dev/null
@@ -1,201 +0,0 @@
-# Copyright 2018, OpenCensus Authors
-#
-# Licensed 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.
-
-# Compiler options for building libcurl.
-
-BASE_CURL_COPTS = [
- # Disable everything else except HTTP protocol.
- "-DHTTP_ONLY=1",
- "-DENABLE_IPV6=1",
- "-DGETHOSTNAME_TYPE_ARG2=size_t",
- "-DGETSERVBYPORT_R_ARGS=6",
- "-DGETSERVBYPORT_R_BUFSIZE=4096",
- "-DHAVE_ALARM=1",
- "-DHAVE_ALLOCA_H=1",
- "-DHAVE_ARPA_INET_H=1",
- "-DHAVE_ARPA_TFTP_H=1",
- "-DHAVE_ASSERT_H=1",
- "-DHAVE_BASENAME=1",
- "-DHAVE_BOOL_T=1",
- "-DHAVE_CLOCK_GETTIME_MONOTONIC=1",
- "-DHAVE_CONNECT=1",
- "-DHAVE_DLFCN_H=1",
- "-DHAVE_ENGINE_LOAD_BUILTIN_ENGINES=1",
- "-DHAVE_ERRNO_H=1",
- "-DHAVE_FCNTL=1",
- "-DHAVE_FCNTL_H=1",
- "-DHAVE_FCNTL_O_NONBLOCK=1",
- "-DHAVE_FDOPEN=1",
- "-DHAVE_FREEADDRINFO=1",
- "-DHAVE_FREEIFADDRS=1",
- "-DHAVE_FSETXATTR=1",
- "-DHAVE_FSETXATTR_5=1",
- "-DHAVE_FTRUNCATE=1",
- "-DHAVE_GAI_STRERROR=1",
- "-DHAVE_GETADDRINFO=1",
- "-DHAVE_GETADDRINFO_THREADSAFE=1",
- "-DHAVE_GETEUID=1",
- "-DHAVE_GETHOSTBYADDR=1",
- "-DHAVE_GETHOSTBYADDR_R=1",
- "-DHAVE_GETHOSTBYADDR_R_8=1",
- "-DHAVE_GETHOSTBYNAME=1",
- "-DHAVE_GETHOSTBYNAME_R=1",
- "-DHAVE_GETHOSTBYNAME_R_6=1",
- "-DHAVE_GETHOSTNAME=1",
- "-DHAVE_GETIFADDRS=1",
- "-DHAVE_GETNAMEINFO=1",
- "-DHAVE_GETPPID=1",
- "-DHAVE_GETPWUID=1",
- "-DHAVE_GETPWUID_R=1",
- "-DHAVE_GETRLIMIT=1",
- "-DHAVE_GETSERVBYPORT_R=1",
- "-DHAVE_GETTIMEOFDAY=1",
- "-DHAVE_GMTIME_R=1",
- "-DHAVE_IFADDRS_H=1",
- "-DHAVE_IF_NAMETOINDEX=1",
- "-DHAVE_INET_NTOP=1",
- "-DHAVE_INET_PTON=1",
- "-DHAVE_INTTYPES_H=1",
- "-DHAVE_IOCTL=1",
- "-DHAVE_IOCTL_FIONBIO=1",
- "-DHAVE_IOCTL_SIOCGIFADDR=1",
- "-DHAVE_LIBGEN_H=1",
- "-DHAVE_LL=1",
- "-DHAVE_LOCALE_H=1",
- "-DHAVE_LOCALTIME_R=1",
- "-DHAVE_LONGLONG=1",
- "-DHAVE_MALLOC_H=1",
- "-DHAVE_MEMORY_H=1",
- "-DHAVE_NETDB_H=1",
- "-DHAVE_NETINET_IN_H=1",
- "-DHAVE_NETINET_TCP_H=1",
- "-DHAVE_NET_IF_H=1",
- "-DHAVE_PIPE=1",
- "-DHAVE_POLL=1",
- "-DHAVE_POLL_FINE=1",
- "-DHAVE_POLL_H=1",
- "-DHAVE_POSIX_STRERROR_R=1",
- "-DHAVE_PTHREAD_H=1",
- "-DHAVE_PWD_H=1",
- "-DHAVE_RECV=1",
- "-DHAVE_SELECT=1",
- "-DHAVE_SEND=1",
- "-DHAVE_SETJMP_H=1",
- "-DHAVE_SETLOCALE=1",
- "-DHAVE_SETRLIMIT=1",
- "-DHAVE_SETSOCKOPT=1",
- "-DHAVE_SGTTY_H=1",
- "-DHAVE_SIGACTION=1",
- "-DHAVE_SIGINTERRUPT=1",
- "-DHAVE_SIGNAL=1",
- "-DHAVE_SIGNAL_H=1",
- "-DHAVE_SIGSETJMP=1",
- "-DHAVE_SIG_ATOMIC_T=1",
- "-DHAVE_SOCKADDR_IN6_SIN6_SCOPE_ID=1",
- "-DHAVE_SOCKET=1",
- "-DHAVE_SOCKETPAIR=1",
- "-DHAVE_STDBOOL_H=1",
- "-DHAVE_STDINT_H=1",
- "-DHAVE_STDIO_H=1",
- "-DHAVE_STDLIB_H=1",
- "-DHAVE_STRCASECMP=1",
- "-DHAVE_STRDUP=1",
- "-DHAVE_STRERROR_R=1",
- "-DHAVE_STRINGS_H=1",
- "-DHAVE_STRING_H=1",
- "-DHAVE_STRNCASECMP=1",
- "-DHAVE_STRSTR=1",
- "-DHAVE_STRTOK_R=1",
- "-DHAVE_STRTOLL=1",
- "-DHAVE_STRUCT_SOCKADDR_STORAGE=1",
- "-DHAVE_STRUCT_TIMEVAL=1",
- "-DHAVE_SYS_IOCTL_H=1",
- "-DHAVE_SYS_PARAM_H=1",
- "-DHAVE_SYS_POLL_H=1",
- "-DHAVE_SYS_RESOURCE_H=1",
- "-DHAVE_SYS_SELECT_H=1",
- "-DHAVE_SYS_SOCKET_H=1",
- "-DHAVE_SYS_STAT_H=1",
- "-DHAVE_SYS_TIME_H=1",
- "-DHAVE_SYS_TYPES_H=1",
- "-DHAVE_SYS_UIO_H=1",
- "-DHAVE_SYS_UN_H=1",
- "-DHAVE_SYS_WAIT_H=1",
- "-DHAVE_SYS_XATTR_H=1",
- "-DHAVE_TERMIOS_H=1",
- "-DHAVE_TERMIO_H=1",
- "-DHAVE_TIME_H=1",
- "-DHAVE_UNISTD_H=1",
- "-DHAVE_UTIME=1",
- "-DHAVE_UTIMES=1",
- "-DHAVE_UTIME_H=1",
- "-DHAVE_VARIADIC_MACROS_C99=1",
- "-DHAVE_VARIADIC_MACROS_GCC=1",
- "-DHAVE_WRITABLE_ARGV=1",
- "-DHAVE_WRITEV=1",
- "-DRECV_TYPE_ARG1=int",
- "-DRECV_TYPE_ARG2=void*",
- "-DRECV_TYPE_ARG3=size_t",
- "-DRECV_TYPE_ARG4=int",
- "-DRECV_TYPE_RETV=ssize_t",
- "-DRETSIGTYPE=void",
- "-DSELECT_QUAL_ARG5=",
- "-DSELECT_TYPE_ARG1=int",
- "-DSELECT_TYPE_ARG234=fd_set*",
- "-DSELECT_TYPE_RETV=int",
- "-DSEND_QUAL_ARG2=const",
- "-DSEND_TYPE_ARG1=int",
- "-DSEND_TYPE_ARG2=void*",
- "-DSEND_TYPE_ARG3=size_t",
- "-DSEND_TYPE_ARG4=int",
- "-DSEND_TYPE_RETV=ssize_t",
- "-DSIZEOF_CURL_OFF_T=8",
- "-DSIZEOF_INT=4",
- "-DSIZEOF_LONG=8",
- "-DSIZEOF_OFF_T=8",
- "-DSIZEOF_SHORT=2",
- "-DSIZEOF_SIZE_T=8",
- "-DSIZEOF_TIME_T=8",
- "-DSTDC_HEADERS=1",
- "-DSTRERROR_R_TYPE_ARG3=size_t",
- "-DTIME_WITH_SYS_TIME=1",
- "-DUSE_THREADS_POSIX=1",
- "-DUSE_UNIX_SOCKETS=1",
-
- # Extra defines needed by curl
- "-DBUILDING_LIBCURL",
- "-DCURL_HIDDEN_SYMBOLS",
-]
-
-LINUX_CURL_COPTS = [
- "-DHAVE_LINUX_TCP_H=1",
- "-DHAVE_MSG_NOSIGNAL=1",
-]
-
-CURL_COPTS = select({
- "//:windows": [
- # Disable everything else except HTTP protocol.
- "/DHTTP_ONLY=1",
- "/DCURL_STATICLIB",
- "/DWIN32",
- "/DBUILDING_LIBCURL",
- "/DUSE_WIN32_IDN",
- "/DWANT_IDN_PROTOTYPES",
- "/DUSE_IPV6",
- "/DUSE_WINDOWS_SSPI",
- "/DUSE_SCHANNEL",
- ],
- "//:osx": BASE_CURL_COPTS,
- "//conditions:default": BASE_CURL_COPTS + LINUX_CURL_COPTS,
-})
diff --git a/cpp/bazel/rocketmq_deps.bzl b/cpp/bazel/rocketmq_deps.bzl
index 983a476..fd201a2 100644
--- a/cpp/bazel/rocketmq_deps.bzl
+++ b/cpp/bazel/rocketmq_deps.bzl
@@ -134,18 +134,6 @@ def rocketmq_deps():
],
)
- maybe(
- http_archive,
- name = "com_github_yhirose_cpp_httplib",
- sha256 = "0ff62e28eb0f6e563178d44b77c94dddb8702141d83dd34b83cb046399c2b1d5",
- build_file = "@org_apache_rocketmq//third_party:cpp_httplib.BUILD",
- strip_prefix = "cpp-httplib-0.9.4",
- urls = [
- "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/cpp-httplib/cpp-httplib-0.9.4.tar.gz",
- "https://github.com/yhirose/cpp-httplib/archive/refs/tags/v0.9.4.tar.gz",
- ],
- )
-
maybe(
http_archive,
name = "com_google_googleapis",
diff --git a/cpp/src/main/cpp/base/BUILD.bazel b/cpp/src/main/cpp/base/BUILD.bazel
index 959bba5..3768b14 100644
--- a/cpp/src/main/cpp/base/BUILD.bazel
+++ b/cpp/src/main/cpp/base/BUILD.bazel
@@ -40,7 +40,6 @@ cc_library(
"@boringssl//:crypto",
"@boringssl//:ssl",
"//external:madler_zlib",
- "@com_github_yhirose_cpp_httplib//:cpp_httplib",
"@asio//:asio",
],
)
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/CredentialsProvider.cpp b/cpp/src/main/cpp/base/CredentialsProvider.cpp
index 548e9d8..5a35bf1 100644
--- a/cpp/src/main/cpp/base/CredentialsProvider.cpp
+++ b/cpp/src/main/cpp/base/CredentialsProvider.cpp
@@ -14,22 +14,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+#include "rocketmq/CredentialsProvider.h"
+
#include <cstdlib>
#include <fstream>
#include <iostream>
#include <string>
+#include "MixAll.h"
#include "absl/memory/memory.h"
#include "absl/strings/match.h"
#include "fmt/format.h"
#include "ghc/filesystem.hpp"
#include "google/protobuf/struct.pb.h"
#include "google/protobuf/util/json_util.h"
-#include "spdlog/spdlog.h"
-
-#include "MixAll.h"
-#include "StsCredentialsProviderImpl.h"
#include "rocketmq/Logger.h"
+#include "spdlog/spdlog.h"
ROCKETMQ_NAMESPACE_BEGIN
@@ -117,94 +117,4 @@ Credentials ConfigFileCredentialsProvider::getCredentials() {
return Credentials(access_key_, access_secret_);
}
-StsCredentialsProvider::StsCredentialsProvider(std::string ram_role_name)
- : impl_(absl::make_unique<StsCredentialsProviderImpl>(std::move(ram_role_name))) {
-}
-
-Credentials StsCredentialsProvider::getCredentials() {
- return impl_->getCredentials();
-}
-
-StsCredentialsProviderImpl::StsCredentialsProviderImpl(std::string ram_role_name)
- : ram_role_name_(std::move(ram_role_name)) {
-}
-
-StsCredentialsProviderImpl::~StsCredentialsProviderImpl() {
- http_client_->shutdown();
-}
-
-Credentials StsCredentialsProviderImpl::getCredentials() {
- if (std::chrono::system_clock::now() >= expiration_) {
- refresh();
- }
-
- {
- absl::MutexLock lk(&mtx_);
- return Credentials(access_key_, access_secret_, session_token_, expiration_);
- }
-}
-
-void StsCredentialsProviderImpl::refresh() {
- std::string path = fmt::format("{}{}", RAM_ROLE_URL_PREFIX, ram_role_name_);
- absl::Mutex sync_mtx;
- absl::CondVar sync_cv;
- bool completed = false;
- auto callback = [&, this](int code, const std::multimap<std::string, std::string>& headers, const std::string& body) {
- SPDLOG_DEBUG("Received STS response. Code: {}", code);
- if (static_cast<int>(HttpStatus::OK) == code) {
- google::protobuf::Struct doc;
- google::protobuf::util::Status status = google::protobuf::util::JsonStringToMessage(body, &doc);
- if (status.ok()) {
- const auto& fields = doc.fields();
- assert(fields.contains(FIELD_ACCESS_KEY));
- std::string access_key = fields.at(FIELD_ACCESS_KEY).string_value();
- assert(fields.contains(FIELD_ACCESS_SECRET));
- std::string access_secret = fields.at(FIELD_ACCESS_SECRET).string_value();
- assert(fields.contains(FIELD_SESSION_TOKEN));
- std::string session_token = fields.at(FIELD_SESSION_TOKEN).string_value();
- assert(fields.contains(FIELD_EXPIRATION));
- std::string expiration_string = fields.at(FIELD_EXPIRATION).string_value();
- absl::Time expiration_instant;
- std::string parse_error;
- if (absl::ParseTime(EXPIRATION_DATE_TIME_FORMAT, expiration_string, absl::UTCTimeZone(), &expiration_instant,
- &parse_error)) {
- absl::MutexLock lk(&mtx_);
- access_key_ = std::move(access_key);
- access_secret_ = std::move(access_secret);
- session_token_ = std::move(session_token);
- expiration_ = absl::ToChronoTime(expiration_instant);
- } else {
- SPDLOG_WARN("Failed to parse expiration time. Message: {}", parse_error);
- }
-
- } else {
- SPDLOG_WARN("Failed to parse STS response. Message: {}", status.message().as_string());
- }
- } else {
- SPDLOG_WARN("STS response code is not OK. Code: {}", code);
- }
-
- {
- absl::MutexLock lk(&sync_mtx);
- completed = true;
- sync_cv.Signal();
- }
- };
-
- http_client_->get(HttpProtocol::HTTP, RAM_ROLE_HOST, 80, path, callback);
-
- while (!completed) {
- absl::MutexLock lk(&sync_mtx);
- sync_cv.Wait(&sync_mtx);
- }
-}
-
-const char* StsCredentialsProviderImpl::RAM_ROLE_HOST = "100.100.100.200";
-const char* StsCredentialsProviderImpl::RAM_ROLE_URL_PREFIX = "/latest/meta-data/Ram/security-credentials/";
-const char* StsCredentialsProviderImpl::FIELD_ACCESS_KEY = "AccessKeyId";
-const char* StsCredentialsProviderImpl::FIELD_ACCESS_SECRET = "AccessKeySecret";
-const char* StsCredentialsProviderImpl::FIELD_SESSION_TOKEN = "SecurityToken";
-const char* StsCredentialsProviderImpl::FIELD_EXPIRATION = "Expiration";
-const char* StsCredentialsProviderImpl::EXPIRATION_DATE_TIME_FORMAT = "%Y-%m-%d%ET%H:%H:%S%Ez";
-
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/HostInfo.cpp b/cpp/src/main/cpp/base/HostInfo.cpp
deleted file mode 100644
index 8b57247..0000000
--- a/cpp/src/main/cpp/base/HostInfo.cpp
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * 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 "HostInfo.h"
-#include "absl/strings/match.h"
-#include "rocketmq/RocketMQ.h"
-#include <cstdlib>
-#include <cstring>
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-const char* HostInfo::ENV_LABEL_SITE = "SIGMA_APP_SITE";
-const char* HostInfo::ENV_LABEL_UNIT = "SIGMA_APP_UNIT";
-const char* HostInfo::ENV_LABEL_APP = "SIGMA_APP_NAME";
-const char* HostInfo::ENV_LABEL_STAGE = "SIGMA_APP_STAGE";
-
-HostInfo::HostInfo() {
- getEnv(ENV_LABEL_SITE, site_);
- getEnv(ENV_LABEL_UNIT, unit_);
- getEnv(ENV_LABEL_APP, app_);
- getEnv(ENV_LABEL_STAGE, stage_);
-}
-
-void HostInfo::getEnv(const char* env, std::string& holder) {
- if (!strlen(env)) {
- return;
- }
-
- char* value = getenv(env);
- if (nullptr != value) {
- holder.clear();
- holder.append(value);
- }
-}
-
-bool HostInfo::hasHostInfo() const {
- return !unit_.empty() && !stage_.empty();
-}
-
-std::string HostInfo::queryString() const {
- if (!hasHostInfo()) {
- return std::string();
- }
-
- std::string query_string("labels=");
- appendLabel(query_string, "site", site_);
- appendLabel(query_string, "unit", unit_);
- appendLabel(query_string, "app", app_);
- appendLabel(query_string, "stage", stage_);
- return query_string;
-}
-
-void HostInfo::appendLabel(std::string& query_string, const char* key, const std::string& value) {
- if (value.empty()) {
- return;
- }
-
- if (absl::EndsWith(query_string, "=")) {
- query_string.append(key).append(":").append(value);
- } else {
- query_string.append(",").append(key).append(":").append(value);
- }
-}
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/HttpClientImpl.cpp b/cpp/src/main/cpp/base/HttpClientImpl.cpp
deleted file mode 100644
index cb25ae9..0000000
--- a/cpp/src/main/cpp/base/HttpClientImpl.cpp
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * 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 "HttpClientImpl.h"
-
-#include <memory>
-#include <string>
-
-#include "fmt/format.h"
-#include "spdlog/spdlog.h"
-
-#include "rocketmq/RocketMQ.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-void HttpClientImpl::start() {
-}
-
-void HttpClientImpl::shutdown() {
-}
-
-/**
- * @brief We current implement this function in sync mode since async http request in CURL is sort of unnecessarily
- * complex.
- *
- * @param protocol
- * @param host
- * @param port
- * @param path
- * @param cb
- */
-void HttpClientImpl::get(
- HttpProtocol protocol, const std::string& host, std::uint16_t port, const std::string& path,
- const std::function<void(int, const std::multimap<std::string, std::string>&, const std::string&)>& cb) {
-
- std::string key;
- switch (protocol) {
- case HttpProtocol::HTTP:
- key = fmt::format("http://{}:{}", host, port);
- break;
- case HttpProtocol::HTTPS:
- key = fmt::format("https://{}:{}", host, port);
- break;
- }
-
- std::shared_ptr<httplib::Client> client;
- {
- absl::MutexLock lk(&clients_mtx_);
- if (clients_.contains(key)) {
- client = clients_[key];
- }
-
- if (!client || !client->is_valid()) {
- client = std::make_shared<httplib::Client>(key);
- clients_.insert_or_assign(key, client);
- }
- }
-
- if (!client || !client->is_valid()) {
- int code = 400;
- std::multimap<std::string, std::string> headers;
- std::string response;
- cb(code, headers, response);
- return;
- }
-
- auto res = client->Get(path.c_str());
-
- std::multimap<std::string, std::string> headers;
- for (auto& header : headers) {
- headers.insert({header.first, header.second});
- }
-
- cb(res->status, headers, res->body);
-}
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/RateLimiter.cpp b/cpp/src/main/cpp/base/RateLimiter.cpp
deleted file mode 100644
index 1a8247e..0000000
--- a/cpp/src/main/cpp/base/RateLimiter.cpp
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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 "RateLimiter.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-RateLimiterObserver::RateLimiterObserver() : stopped_(false) {
- tick_thread_ = std::thread([this] {
- while (!stopped_.load(std::memory_order_relaxed)) {
- {
- std::lock_guard<std::mutex> lk(members_mtx_);
- for (auto it = members_.begin(); it != members_.end();) {
- std::shared_ptr<Tick> tick = it->lock();
- if (!tick) {
- it = members_.erase(it);
- continue;
- } else {
- ++it;
- }
- tick->tick();
- }
- }
- std::this_thread::sleep_for(std::chrono::milliseconds(100));
- }
- });
-}
-
-void RateLimiterObserver::subscribe(const std::shared_ptr<Tick>& tick) {
- std::lock_guard<std::mutex> lk(members_mtx_);
- members_.emplace_back(tick);
-}
-
-void RateLimiterObserver::stop() {
- stopped_.store(true, std::memory_order_relaxed);
- if (tick_thread_.joinable()) {
- tick_thread_.join();
- }
-}
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/TopAddressing.cpp b/cpp/src/main/cpp/base/TopAddressing.cpp
deleted file mode 100644
index dd45e67..0000000
--- a/cpp/src/main/cpp/base/TopAddressing.cpp
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * 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 "TopAddressing.h"
-
-#include "HttpClientImpl.h"
-#include "absl/memory/memory.h"
-#include "absl/strings/match.h"
-#include "absl/strings/str_split.h"
-#include "spdlog/spdlog.h"
-#include <utility>
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-TopAddressing::TopAddressing() : TopAddressing("jmenv.tbsite.net", 8080, "/rocketmq/nsaddr") {
-}
-
-TopAddressing::TopAddressing(std::string host, int port, std::string path)
- : host_(std::move(host)), port_(port), path_(std::move(path)), http_client_(absl::make_unique<HttpClientImpl>()) {
- http_client_->start();
-}
-
-TopAddressing::~TopAddressing() {
- http_client_->shutdown();
-}
-
-void TopAddressing::fetchNameServerAddresses(const std::function<void(bool, const std::vector<std::string>&)>& cb) {
- SPDLOG_DEBUG("Prepare to send HTTP request, timeout=3s");
- std::string base(fmt::format("http://{}:{}", host_, port_));
- // Append host info if necessary.
- std::string query_string(path_);
-
- if (absl::StrContains(query_string, "?")) {
- query_string.append("&");
- } else {
- query_string.append("?");
- }
-
- if (host_info_.hasHostInfo()) {
- query_string.append(host_info_.queryString());
- } else {
- query_string.append("nofix=1");
- }
-
- auto callback = [cb](int code, const std::multimap<std::string, std::string>& metadata, const std::string& body) {
- SPDLOG_DEBUG("Receive HTTP response. Code: {}, body: {}", code, body);
- if (static_cast<int>(HttpStatus::OK) == code) {
- cb(true, absl::StrSplit(body, ';'));
- } else {
- std::vector<std::string> name_server_list;
- cb(false, name_server_list);
- }
- };
-
- http_client_->get(HttpProtocol::HTTP, host_, port_, query_string, callback);
-}
-
-void TopAddressing::injectHttpClient(std::unique_ptr<HttpClient> http_client) {
- http_client_->shutdown();
- http_client_.swap(http_client);
-}
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/include/Histogram.h b/cpp/src/main/cpp/base/include/Histogram.h
deleted file mode 100644
index 89b7f3f..0000000
--- a/cpp/src/main/cpp/base/include/Histogram.h
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * 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.
- */
-#pragma once
-
-#include <atomic>
-#include <cassert>
-#include <memory>
-#include <string>
-#include <vector>
-
-#include "rocketmq/RocketMQ.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-class Histogram {
-public:
- Histogram(std::string title, int32_t capacity) : title_(std::move(title)), capacity_(capacity) {
- data_.reserve(capacity);
- for (int i = 0; i < capacity_; ++i) {
- data_.push_back(std::unique_ptr<std::atomic<int32_t>>(new std::atomic<int32_t>(0)));
- }
- }
-
- void countIn(int grade) {
- if (grade < 0) {
- return;
- }
-
- if (grade >= capacity_) {
- data_[capacity_ - 1]->fetch_add(1, std::memory_order_relaxed);
- return;
- }
-
- data_[grade]->fetch_add(1, std::memory_order_relaxed);
- }
-
- /**
- * Change labels of histogram duration the initialization phase only.
- * @return
- */
- std::vector<std::string>& labels() {
- return labels_;
- }
-
- void reportAndReset(std::string& result) {
- assert(labels_.size() == static_cast<std::vector<std::string>::size_type>(capacity_));
- std::vector<int32_t> values;
- values.reserve(capacity_);
- for (auto& item : data_) {
- int value = item->load(std::memory_order_relaxed);
- values.push_back(value);
- item->fetch_sub(value, std::memory_order_relaxed);
- }
- result.clear();
- result.append(title_).append(":");
- for (std::vector<std::string>::size_type i = 0; i < labels_.size(); ++i) {
- if (i) {
- result.append(", ");
- }
- result.append(labels_[i]).append(std::to_string(values[i]));
- }
- }
-
-private:
- std::string title_;
- std::vector<std::unique_ptr<std::atomic<int32_t>>> data_;
- int32_t capacity_;
- std::vector<std::string> labels_;
-};
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/include/HostInfo.h b/cpp/src/main/cpp/base/include/HostInfo.h
deleted file mode 100644
index 1b2a543..0000000
--- a/cpp/src/main/cpp/base/include/HostInfo.h
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.
- */
-#pragma once
-
-#include <string>
-
-#include "rocketmq/RocketMQ.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-struct HostInfo {
- std::string site_;
- std::string unit_;
- std::string app_;
- std::string stage_;
-
- explicit HostInfo();
-
- bool hasHostInfo() const;
-
- std::string queryString() const;
-
- static const char* ENV_LABEL_SITE;
- static const char* ENV_LABEL_UNIT;
- static const char* ENV_LABEL_APP;
- static const char* ENV_LABEL_STAGE;
-
-private:
- static void getEnv(const char* env, std::string& holder);
-
- static void appendLabel(std::string& query_string, const char* key, const std::string& value);
-};
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/include/HttpClient.h b/cpp/src/main/cpp/base/include/HttpClient.h
deleted file mode 100644
index 77f6955..0000000
--- a/cpp/src/main/cpp/base/include/HttpClient.h
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.
- */
-#pragma once
-
-#include <cstdint>
-#include <functional>
-#include <map>
-#include <string>
-
-#include "rocketmq/RocketMQ.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-enum class HttpProtocol : int8_t
-{
- HTTP = 1,
- HTTPS = 2,
-};
-
-enum class HttpStatus : int
-{
- OK = 200,
- INTERNAL = 500,
-};
-
-class HttpClient {
-public:
- virtual ~HttpClient() = default;
-
- virtual void start() = 0;
-
- virtual void shutdown() = 0;
-
- virtual void
- get(HttpProtocol protocol, const std::string& host, std::uint16_t port, const std::string& path,
- const std::function<void(int, const std::multimap<std::string, std::string>&, const std::string&)>& cb) = 0;
-};
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/include/HttpClientImpl.h b/cpp/src/main/cpp/base/include/HttpClientImpl.h
deleted file mode 100644
index e54bb43..0000000
--- a/cpp/src/main/cpp/base/include/HttpClientImpl.h
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.
- */
-#pragma once
-
-#include "absl/base/thread_annotations.h"
-#include "absl/container/flat_hash_map.h"
-#include "absl/synchronization/mutex.h"
-#include "httplib.h"
-
-#include "HttpClient.h"
-#include "rocketmq/RocketMQ.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-class HttpClientImpl : public HttpClient {
-public:
- ~HttpClientImpl() override = default;
-
- void start() override;
-
- void shutdown() override;
-
- void
- get(HttpProtocol protocol, const std::string& host, std::uint16_t port, const std::string& path,
- const std::function<void(int, const std::multimap<std::string, std::string>&, const std::string&)>& cb) override;
-
-private:
- absl::flat_hash_map<std::string, std::shared_ptr<httplib::Client>> clients_ GUARDED_BY(clients_mtx_);
- absl::Mutex clients_mtx_;
-};
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/include/RateLimiter.h b/cpp/src/main/cpp/base/include/RateLimiter.h
deleted file mode 100644
index 76982a3..0000000
--- a/cpp/src/main/cpp/base/include/RateLimiter.h
+++ /dev/null
@@ -1,184 +0,0 @@
-/*
- * 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.
- */
-#pragma once
-
-#include <array>
-#include <atomic>
-#include <condition_variable>
-#include <memory>
-#include <mutex>
-#include <string>
-#include <thread>
-#include <vector>
-
-#include "rocketmq/RocketMQ.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-class Tick {
-public:
- virtual ~Tick() = default;
-
- virtual void tick() = 0;
-};
-
-class RateLimiterObserver {
-public:
- RateLimiterObserver();
-
- void subscribe(const std::shared_ptr<Tick>& tick);
-
- void stop();
-
-private:
- std::vector<std::weak_ptr<Tick>> members_;
- std::mutex members_mtx_;
-
- std::atomic_bool stopped_;
-
- std::thread tick_thread_;
-};
-
-template <int PARTITION>
-class RateLimiter : public Tick {
-public:
- explicit RateLimiter(uint32_t permit) : permits_{0}, interval_(1000 / PARTITION) {
- uint32_t avg = permit / PARTITION;
- for (auto& i : partition_) {
- i = avg;
- }
-
- uint32_t r = permit % PARTITION;
-
- if (r) {
- uint32_t step = PARTITION / r;
- for (uint32_t i = 0; i < r; ++i) {
- partition_[i * step]++;
- }
- }
- }
-
- ~RateLimiter() override = default;
-
- std::array<uint32_t, PARTITION>& partition() {
- return permits_;
- }
-
- int slot() {
- auto current = std::chrono::steady_clock::now();
- long ms = std::chrono::duration_cast<std::chrono::milliseconds>(current.time_since_epoch()).count();
- return ms / interval_ % PARTITION;
- }
-
- uint32_t available() {
- uint32_t available_permit = 0;
- int idx = slot();
- // Reuse quota of the past half of the cycle
- for (int j = 0; j <= PARTITION / 2; ++j) {
- int index = idx - j;
- if (index < 0) {
- index += PARTITION;
- }
- if (permits_[index] > 0) {
- available_permit += permits_[index];
- }
- }
-
- return available_permit;
- }
-
- uint32_t acquire(uint32_t permit) {
- int idx = slot();
- {
- std::unique_lock<std::mutex> lk(mtx_);
- if (permits_[idx] >= permit) {
- permits_[idx] -= permit;
- return permit;
- }
-
- uint32_t acquired = 0;
- if (permits_[idx] > 0) {
- acquired += permits_[idx];
- permits_[idx] = 0;
- }
-
- // Reuse quota of the past half of the cycle
- for (int j = 1; j <= PARTITION / 2; ++j) {
- int index = idx - j;
- if (index < 0) {
- index += PARTITION;
- }
- while (permits_[index] > 0) {
- --permits_[index];
- if (++acquired >= permit) {
- break;
- }
- }
- }
- return acquired;
- }
- }
-
- void acquire() {
- int idx = slot();
- {
- std::unique_lock<std::mutex> lk(mtx_);
- if (permits_[idx] > 0) {
- --permits_[idx];
- return;
- }
-
- // Reuse quota of the past half of the cycle
- for (int j = 1; j <= PARTITION / 2; ++j) {
- int index = idx - j;
- if (index < 0) {
- index += PARTITION;
- }
- if (permits_[index] > 0) {
- --permits_[index];
- return;
- }
- }
-
- cv_.wait(lk, [this]() {
- int idx = slot();
- return permits_[idx] > 0;
- });
- idx = slot();
- --permits_[idx];
- }
- }
-
- void tick() override {
- std::this_thread::sleep_for(std::chrono::milliseconds(1000 / PARTITION));
- int idx = slot();
- {
- std::unique_lock<std::mutex> lk(mtx_);
- permits_[idx] = partition_[idx];
- cv_.notify_all();
- }
- }
-
-private:
- std::array<uint32_t, PARTITION> partition_;
- std::array<uint32_t, PARTITION> permits_;
- int interval_;
- std::mutex mtx_;
- std::condition_variable cv_;
-};
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/include/StsCredentialsProviderImpl.h b/cpp/src/main/cpp/base/include/StsCredentialsProviderImpl.h
deleted file mode 100644
index 0789e4a..0000000
--- a/cpp/src/main/cpp/base/include/StsCredentialsProviderImpl.h
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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.
- */
-#pragma once
-
-#include "absl/base/thread_annotations.h"
-#include "absl/synchronization/mutex.h"
-
-#include "HttpClient.h"
-#include "rocketmq/CredentialsProvider.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-class StsCredentialsProviderImpl : public CredentialsProvider {
-public:
- explicit StsCredentialsProviderImpl(std::string ram_role_name);
-
- ~StsCredentialsProviderImpl() override;
-
- Credentials getCredentials() override;
-
- void withHttpClient(std::unique_ptr<HttpClient> http_client) {
- http_client_ = std::move(http_client);
- }
-
-private:
- static const char* RAM_ROLE_HOST;
- static const char* RAM_ROLE_URL_PREFIX;
- static const char* FIELD_ACCESS_KEY;
- static const char* FIELD_ACCESS_SECRET;
- static const char* FIELD_SESSION_TOKEN;
- static const char* FIELD_EXPIRATION;
- static const char* EXPIRATION_DATE_TIME_FORMAT;
-
- std::string ram_role_name_;
-
- std::string access_key_ GUARDED_BY(mtx_);
- std::string access_secret_ GUARDED_BY(mtx_);
- std::string session_token_ GUARDED_BY(mtx_);
- std::chrono::system_clock::time_point expiration_;
-
- absl::Mutex mtx_;
- void refresh() LOCKS_EXCLUDED(mtx_);
-
- std::unique_ptr<HttpClient> http_client_;
-};
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/base/include/TopAddressing.h b/cpp/src/main/cpp/base/include/TopAddressing.h
deleted file mode 100644
index 0260259..0000000
--- a/cpp/src/main/cpp/base/include/TopAddressing.h
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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.
- */
-#pragma once
-
-#include <cassert>
-#include <cstdlib>
-#include <functional>
-#include <memory>
-#include <string>
-#include <vector>
-
-#include "HostInfo.h"
-#include "HttpClient.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-class TopAddressing {
-public:
- TopAddressing();
-
- TopAddressing(std::string host, int port, std::string path);
-
- virtual ~TopAddressing();
-
- void fetchNameServerAddresses(const std::function<void(bool, const std::vector<std::string>&)>& cb);
-
- void injectHttpClient(std::unique_ptr<HttpClient> http_client);
-
-private:
- std::string host_;
- int port_{8080};
- std::string path_;
- HostInfo host_info_;
-
- std::unique_ptr<HttpClient> http_client_;
-};
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/client/ClientManagerImpl.cpp b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
index 78b13f7..44accf1 100644
--- a/cpp/src/main/cpp/client/ClientManagerImpl.cpp
+++ b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
@@ -50,11 +50,7 @@ ROCKETMQ_NAMESPACE_BEGIN
ClientManagerImpl::ClientManagerImpl(std::string resource_namespace)
: scheduler_(std::make_shared<SchedulerImpl>()), resource_namespace_(std::move(resource_namespace)),
state_(State::CREATED),
- callback_thread_pool_(absl::make_unique<ThreadPoolImpl>(std::thread::hardware_concurrency())),
- latency_histogram_("Message-Latency", 11) {
- spdlog::set_level(spdlog::level::trace);
- assignLabels(latency_histogram_);
-
+ callback_thread_pool_(absl::make_unique<ThreadPoolImpl>(std::thread::hardware_concurrency())) {
certificate_verifier_ = grpc::experimental::ExternalCertificateVerifier::Create<InsecureCertificateVerifier>();
tls_channel_credential_options_.set_verify_server_certs(false);
tls_channel_credential_options_.set_check_call_host(false);
@@ -118,14 +114,6 @@ void ClientManagerImpl::start() {
scheduler_->schedule(heartbeat_functor, HEARTBEAT_TASK_NAME, std::chrono::seconds(1), std::chrono::seconds(10));
SPDLOG_DEBUG("Heartbeat task-id={}", heartbeat_task_id_);
- auto stats_functor_ = [client_instance_weak_ptr]() {
- auto client_instance = client_instance_weak_ptr.lock();
- if (client_instance) {
- client_instance->logStats();
- }
- };
- stats_task_id_ =
- scheduler_->schedule(stats_functor_, STATS_TASK_NAME, std::chrono::seconds(0), std::chrono::seconds(10));
state_.store(State::STARTED, std::memory_order_relaxed);
}
@@ -143,10 +131,6 @@ void ClientManagerImpl::shutdown() {
scheduler_->cancel(heartbeat_task_id_);
}
- if (stats_task_id_) {
- scheduler_->cancel(stats_task_id_);
- }
-
scheduler_->shutdown();
{
@@ -159,20 +143,6 @@ void ClientManagerImpl::shutdown() {
SPDLOG_DEBUG("ClientManager stopped");
}
-void ClientManagerImpl::assignLabels(Histogram& histogram) {
- histogram.labels().emplace_back("[000ms~020ms): ");
- histogram.labels().emplace_back("[020ms~040ms): ");
- histogram.labels().emplace_back("[040ms~060ms): ");
- histogram.labels().emplace_back("[060ms~080ms): ");
- histogram.labels().emplace_back("[080ms~100ms): ");
- histogram.labels().emplace_back("[100ms~120ms): ");
- histogram.labels().emplace_back("[120ms~140ms): ");
- histogram.labels().emplace_back("[140ms~160ms): ");
- histogram.labels().emplace_back("[160ms~180ms): ");
- histogram.labels().emplace_back("[180ms~200ms): ");
- histogram.labels().emplace_back("[200ms~inf): ");
-}
-
std::vector<std::string> ClientManagerImpl::cleanOfflineRpcClients() {
absl::flat_hash_set<std::string> hosts;
{
@@ -964,13 +934,6 @@ MessageConstSharedPtr ClientManagerImpl::wrapMessage(const rmq::Message& item) {
// Decoded Time-Point
extension.decode_time = std::chrono::system_clock::now();
- // Extension
- {
- auto elapsed = MixAll::millisecondsOf(std::chrono::system_clock::now() - extension.store_time);
- if (elapsed >= 0) {
- latency_histogram_.countIn(elapsed / 20);
- }
- }
return MessageConstSharedPtr(raw);
}
@@ -1458,12 +1421,6 @@ std::error_code ClientManagerImpl::notifyClientTermination(const std::string& ta
return ec;
}
-void ClientManagerImpl::logStats() {
- std::string stats;
- latency_histogram_.reportAndReset(stats);
- SPDLOG_INFO("{}", stats);
-}
-
void ClientManagerImpl::submit(std::function<void()> task) {
State current_state = state();
if (current_state == State::STOPPING || current_state == State::STOPPED) {
diff --git a/cpp/src/main/cpp/client/include/ClientManager.h b/cpp/src/main/cpp/client/include/ClientManager.h
index 2ec5b2e..56325fa 100644
--- a/cpp/src/main/cpp/client/include/ClientManager.h
+++ b/cpp/src/main/cpp/client/include/ClientManager.h
@@ -28,7 +28,6 @@
#include "RpcClient.h"
#include "Scheduler.h"
#include "TelemetryBidiReactor.h"
-#include "TopAddressing.h"
#include "TopicRouteData.h"
#include "rocketmq/SendCallback.h"
#include "rocketmq/State.h"
diff --git a/cpp/src/main/cpp/client/include/ClientManagerImpl.h b/cpp/src/main/cpp/client/include/ClientManagerImpl.h
index 63b0713..0769769 100644
--- a/cpp/src/main/cpp/client/include/ClientManagerImpl.h
+++ b/cpp/src/main/cpp/client/include/ClientManagerImpl.h
@@ -28,20 +28,14 @@
#include "Client.h"
#include "ClientManager.h"
-#include "HeartbeatDataCallback.h"
-#include "Histogram.h"
#include "InsecureCertificateVerifier.h"
#include "InvocationContext.h"
#include "ReceiveMessageCallback.h"
-#include "RpcClient.h"
#include "RpcClientImpl.h"
-#include "Scheduler.h"
#include "SchedulerImpl.h"
#include "SendMessageContext.h"
#include "TelemetryBidiReactor.h"
#include "ThreadPoolImpl.h"
-#include "TopAddressing.h"
-#include "TopicRouteChangeCallback.h"
#include "TopicRouteData.h"
#include "absl/base/thread_annotations.h"
#include "absl/container/flat_hash_map.h"
@@ -68,8 +62,6 @@ public:
void shutdown() override LOCKS_EXCLUDED(rpc_clients_mtx_);
- static void assignLabels(Histogram& histogram);
-
std::shared_ptr<grpc::Channel> createChannel(const std::string& target_host) override;
/**
@@ -81,7 +73,9 @@ public:
* @param timeout RPC timeout.
* @param cb Callback to execute once the request/response completes.
*/
- void resolveRoute(const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ void resolveRoute(const std::string& target_host,
+ const Metadata& metadata,
+ const QueryRouteRequest& request,
std::chrono::milliseconds timeout,
const std::function<void(const std::error_code&, const TopicRouteDataPtr&)>& cb) override
LOCKS_EXCLUDED(rpc_clients_mtx_);
@@ -92,7 +86,9 @@ public:
*/
std::vector<std::string> cleanOfflineRpcClients() LOCKS_EXCLUDED(clients_mtx_, rpc_clients_mtx_);
- bool send(const std::string& target_host, const Metadata& metadata, SendMessageRequest& request,
+ bool send(const std::string& target_host,
+ const Metadata& metadata,
+ SendMessageRequest& request,
SendCallback cb) override LOCKS_EXCLUDED(rpc_clients_mtx_);
/**
@@ -109,7 +105,8 @@ public:
RpcClientSharedPtr getRpcClient(const std::string& target_host, bool need_heartbeat = true) override
LOCKS_EXCLUDED(rpc_clients_mtx_);
- static SendReceipt processSendResponse(const rmq::MessageQueue& message_queue, const SendMessageResponse& response,
+ static SendReceipt processSendResponse(const rmq::MessageQueue& message_queue,
+ const SendMessageResponse& response,
std::error_code& ec);
// only for test
@@ -120,13 +117,17 @@ public:
void addClientObserver(std::weak_ptr<Client> client) override;
- void queryAssignment(const std::string& target, const Metadata& metadata, const QueryAssignmentRequest& request,
+ void queryAssignment(const std::string& target,
+ const Metadata& metadata,
+ const QueryAssignmentRequest& request,
std::chrono::milliseconds timeout,
const std::function<void(const std::error_code&, const QueryAssignmentResponse&)>& cb) override;
- void receiveMessage(const std::string& target, const Metadata& metadata, const ReceiveMessageRequest& request,
- std::chrono::milliseconds timeout, ReceiveMessageCallback cb) override
- LOCKS_EXCLUDED(rpc_clients_mtx_);
+ void receiveMessage(const std::string& target,
+ const Metadata& metadata,
+ const ReceiveMessageRequest& request,
+ std::chrono::milliseconds timeout,
+ ReceiveMessageCallback cb) override LOCKS_EXCLUDED(rpc_clients_mtx_);
/**
* Translate protobuf message struct to domain model.
@@ -143,11 +144,16 @@ public:
* @param target_host Target broker host address.
* @param request Ack message request.
*/
- void ack(const std::string& target_host, const Metadata& metadata, const AckMessageRequest& request,
- std::chrono::milliseconds timeout, const std::function<void(const std::error_code&)>& cb) override;
-
- void changeInvisibleDuration(const std::string& target_host, const Metadata& metadata,
- const ChangeInvisibleDurationRequest&, std::chrono::milliseconds timeout,
+ void ack(const std::string& target_host,
+ const Metadata& metadata,
+ const AckMessageRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code&)>& cb) override;
+
+ void changeInvisibleDuration(const std::string& target_host,
+ const Metadata& metadata,
+ const ChangeInvisibleDurationRequest&,
+ std::chrono::milliseconds timeout,
const std::function<void(const std::error_code&)>&) override;
void forwardMessageToDeadLetterQueue(const std::string& target_host,
@@ -170,11 +176,14 @@ public:
* @param timeout
* @param cb
*/
- void endTransaction(const std::string& target_host, const Metadata& metadata, const EndTransactionRequest& request,
+ void endTransaction(const std::string& target_host,
+ const Metadata& metadata,
+ const EndTransactionRequest& request,
std::chrono::milliseconds timeout,
const std::function<void(const std::error_code&, const EndTransactionResponse&)>& cb) override;
- std::error_code notifyClientTermination(const std::string& target_host, const Metadata& metadata,
+ std::error_code notifyClientTermination(const std::string& target_host,
+ const Metadata& metadata,
const NotifyClientTerminationRequest& request,
std::chrono::milliseconds timeout) override;
@@ -182,7 +191,9 @@ public:
trace_ = trace;
}
- void heartbeat(const std::string& target_host, const Metadata& metadata, const HeartbeatRequest& request,
+ void heartbeat(const std::string& target_host,
+ const Metadata& metadata,
+ const HeartbeatRequest& request,
std::chrono::milliseconds timeout,
const std::function<void(const std::error_code&, const HeartbeatResponse&)>& cb) override;
@@ -193,8 +204,6 @@ public:
private:
void doHeartbeat();
- void logStats();
-
SchedulerSharedPtr scheduler_;
static const char* HEARTBEAT_TASK_NAME;
@@ -208,15 +217,12 @@ private:
absl::Mutex clients_mtx_;
absl::flat_hash_map<std::string, std::shared_ptr<RpcClient>> rpc_clients_ GUARDED_BY(rpc_clients_mtx_);
- absl::Mutex rpc_clients_mtx_; // protects rpc_clients_
+ absl::Mutex rpc_clients_mtx_; // protects rpc_clients_
std::uint32_t heartbeat_task_id_{0};
- std::uint32_t stats_task_id_{0};
std::unique_ptr<ThreadPoolImpl> callback_thread_pool_;
- Histogram latency_histogram_;
-
absl::flat_hash_set<std::string> exporter_endpoint_set_ GUARDED_BY(exporter_endpoint_set_mtx_);
absl::Mutex exporter_endpoint_set_mtx_;
diff --git a/cpp/src/main/cpp/client/include/HeartbeatDataCallback.h b/cpp/src/main/cpp/client/include/HeartbeatDataCallback.h
deleted file mode 100644
index f4466e2..0000000
--- a/cpp/src/main/cpp/client/include/HeartbeatDataCallback.h
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.
- */
-#pragma once
-
-#include "RpcClient.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-/**
- * This callback will be invoked to collect and prepare heartbeat data, which will be sent to brokers.
- */
-class HeartbeatDataCallback {
-public:
- virtual void onHeartbeatDataCallback(HeartbeatRequest& request) = 0;
-};
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/client/include/Topic.h b/cpp/src/main/cpp/client/include/Topic.h
deleted file mode 100644
index 258abab..0000000
--- a/cpp/src/main/cpp/client/include/Topic.h
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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.
- */
-#pragma once
-
-#include <string>
-#include <unordered_map>
-
-#include "rocketmq/RocketMQ.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-class Topic {
-public:
- Topic(std::string resource_namespace, std::string name)
- : resource_namespace_(std::move(resource_namespace)), name_(std::move(name)) {
- }
-
- const std::string& resourceNamespace() const {
- return resource_namespace_;
- }
-
- const std::string& name() const {
- return name_;
- }
-
- bool operator==(const Topic& other) const {
- return resource_namespace_ == other.resource_namespace_ && name_ == other.name_;
- }
-
- bool operator<(const Topic& other) const {
- if (resource_namespace_ < other.resource_namespace_) {
- return true;
- } else if (resource_namespace_ > other.resource_namespace_) {
- return false;
- }
-
- if (name_ < other.name_) {
- return true;
- } else if (name_ > other.name_) {
- return false;
- }
- return false;
- }
-
-private:
- std::string resource_namespace_;
- std::string name_;
-};
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/client/include/TopicRouteChangeCallback.h b/cpp/src/main/cpp/client/include/TopicRouteChangeCallback.h
deleted file mode 100644
index 0490808..0000000
--- a/cpp/src/main/cpp/client/include/TopicRouteChangeCallback.h
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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.
- */
-#pragma once
-
-#include "absl/strings/string_view.h"
-
-#include "TopicRouteData.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-class TopicRouteChangeCallback {
-public:
- virtual void onTopicRouteChange(absl::string_view topic, const TopicRouteDataPtr& topic_route_data) = 0;
-};
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/rocketmq/ClientImpl.cpp b/cpp/src/main/cpp/rocketmq/ClientImpl.cpp
index 61e6ed9..190b9c2 100644
--- a/cpp/src/main/cpp/rocketmq/ClientImpl.cpp
+++ b/cpp/src/main/cpp/rocketmq/ClientImpl.cpp
@@ -32,7 +32,6 @@
#include <utility>
#include "ClientManagerFactory.h"
-#include "HttpClientImpl.h"
#include "InvocationContext.h"
#include "LoggerImpl.h"
#include "MessageExt.h"
diff --git a/cpp/src/main/cpp/rocketmq/DynamicNameServerResolver.cpp b/cpp/src/main/cpp/rocketmq/DynamicNameServerResolver.cpp
deleted file mode 100644
index f9981bc..0000000
--- a/cpp/src/main/cpp/rocketmq/DynamicNameServerResolver.cpp
+++ /dev/null
@@ -1,129 +0,0 @@
-/*
- * 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 "DynamicNameServerResolver.h"
-
-#include <atomic>
-#include <chrono>
-#include <cstdint>
-#include <functional>
-#include <memory>
-
-#include "absl/strings/str_join.h"
-
-#include "LoggerImpl.h"
-#include "SchedulerImpl.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-DynamicNameServerResolver::DynamicNameServerResolver(absl::string_view endpoint,
- std::chrono::milliseconds refresh_interval)
- : endpoint_(endpoint.data(), endpoint.length()), scheduler_(std::make_shared<SchedulerImpl>(1)),
- refresh_interval_(refresh_interval) {
- absl::string_view remains;
- if (absl::StartsWith(endpoint_, "https://")) {
- ssl_ = true;
- remains = absl::StripPrefix(endpoint_, "https://");
- } else {
- remains = absl::StripPrefix(endpoint_, "http://");
- }
-
- std::int32_t port = 80;
- if (ssl_) {
- port = 443;
- }
-
- absl::string_view host;
- if (absl::StrContains(remains, ':')) {
- std::vector<absl::string_view> segments = absl::StrSplit(remains, ':');
- host = segments[0];
- remains = absl::StripPrefix(remains, host);
- remains = absl::StripPrefix(remains, ":");
-
- segments = absl::StrSplit(remains, '/');
- if (!absl::SimpleAtoi(segments[0], &port)) {
- SPDLOG_WARN("Failed to parse port of name-server-list discovery service endpoint");
- abort();
- }
- remains = absl::StripPrefix(remains, segments[0]);
- } else {
- std::vector<absl::string_view> segments = absl::StrSplit(remains, '/');
- host = segments[0];
- remains = absl::StripPrefix(remains, host);
- }
-
- top_addressing_ = absl::make_unique<TopAddressing>(std::string(host.data(), host.length()), port,
- std::string(remains.data(), remains.length()));
-}
-
-std::string DynamicNameServerResolver::resolve() {
- bool fetch_immediately = false;
- {
- absl::MutexLock lk(&name_server_list_mtx_);
- if (name_server_list_.empty()) {
- fetch_immediately = true;
- }
- }
-
- if (fetch_immediately) {
- fetch();
- }
-
- {
- absl::MutexLock lk(&name_server_list_mtx_);
- return naming_scheme_.buildAddress(name_server_list_);
- }
-}
-
-void DynamicNameServerResolver::fetch() {
- std::weak_ptr<DynamicNameServerResolver> ptr(shared_from_this());
- auto callback = [ptr](bool success, const std::vector<std::string>& name_server_list) {
- if (success && !name_server_list.empty()) {
- std::shared_ptr<DynamicNameServerResolver> resolver = ptr.lock();
- if (resolver) {
- resolver->onNameServerListFetched(name_server_list);
- }
- }
- };
- top_addressing_->fetchNameServerAddresses(callback);
-}
-
-void DynamicNameServerResolver::onNameServerListFetched(const std::vector<std::string>& name_server_list) {
- if (!name_server_list.empty()) {
- absl::MutexLock lk(&name_server_list_mtx_);
- if (name_server_list_ != name_server_list) {
- SPDLOG_INFO("Name server list changed. {} --> {}", absl::StrJoin(name_server_list_, ";"),
- absl::StrJoin(name_server_list, ";"));
- name_server_list_ = name_server_list;
- }
- }
-}
-
-void DynamicNameServerResolver::injectHttpClient(std::unique_ptr<HttpClient> http_client) {
- top_addressing_->injectHttpClient(std::move(http_client));
-}
-
-void DynamicNameServerResolver::start() {
- scheduler_->start();
- scheduler_->schedule(std::bind(&DynamicNameServerResolver::fetch, this), "DynamicNameServerResolver",
- std::chrono::milliseconds(0), refresh_interval_);
-}
-
-void DynamicNameServerResolver::shutdown() {
- scheduler_->shutdown();
-}
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/rocketmq/Producer.cpp b/cpp/src/main/cpp/rocketmq/Producer.cpp
index b2bc5e0..cf2b4da 100644
--- a/cpp/src/main/cpp/rocketmq/Producer.cpp
+++ b/cpp/src/main/cpp/rocketmq/Producer.cpp
@@ -21,7 +21,6 @@
#include <system_error>
#include <utility>
-#include "DynamicNameServerResolver.h"
#include "LoggerImpl.h"
#include "MixAll.h"
#include "ProducerImpl.h"
diff --git a/cpp/src/main/cpp/rocketmq/PushConsumer.cpp b/cpp/src/main/cpp/rocketmq/PushConsumer.cpp
index daabf01..17ea8ca 100644
--- a/cpp/src/main/cpp/rocketmq/PushConsumer.cpp
+++ b/cpp/src/main/cpp/rocketmq/PushConsumer.cpp
@@ -17,7 +17,6 @@
#include <chrono>
#include <memory>
-#include "DynamicNameServerResolver.h"
#include "PushConsumerImpl.h"
#include "StaticNameServerResolver.h"
#include "rocketmq/PushConsumer.h"
diff --git a/cpp/src/main/cpp/rocketmq/include/ConsumeMessageServiceImpl.h b/cpp/src/main/cpp/rocketmq/include/ConsumeMessageServiceImpl.h
index 632a388..91979d2 100644
--- a/cpp/src/main/cpp/rocketmq/include/ConsumeMessageServiceImpl.h
+++ b/cpp/src/main/cpp/rocketmq/include/ConsumeMessageServiceImpl.h
@@ -22,7 +22,6 @@
#include <system_error>
#include "ConsumeMessageService.h"
-#include "RateLimiter.h"
#include "ThreadPool.h"
#include "absl/container/flat_hash_map.h"
#include "rocketmq/State.h"
diff --git a/cpp/src/main/cpp/rocketmq/include/DynamicNameServerResolver.h b/cpp/src/main/cpp/rocketmq/include/DynamicNameServerResolver.h
deleted file mode 100644
index 5bc2c53..0000000
--- a/cpp/src/main/cpp/rocketmq/include/DynamicNameServerResolver.h
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * 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.
- */
-#pragma once
-
-#include <atomic>
-#include <chrono>
-#include <cstdint>
-#include <cstdlib>
-#include <memory>
-
-#include "absl/base/thread_annotations.h"
-#include "absl/memory/memory.h"
-#include "absl/strings/numbers.h"
-#include "absl/strings/str_split.h"
-#include "absl/strings/string_view.h"
-#include "absl/synchronization/mutex.h"
-
-#include "NameServerResolver.h"
-#include "NamingScheme.h"
-#include "Scheduler.h"
-#include "TopAddressing.h"
-
-ROCKETMQ_NAMESPACE_BEGIN
-
-class DynamicNameServerResolver : public NameServerResolver,
- public std::enable_shared_from_this<DynamicNameServerResolver> {
-public:
- DynamicNameServerResolver(absl::string_view endpoint, std::chrono::milliseconds refresh_interval);
-
- void start() override;
-
- void shutdown() override;
-
- std::string resolve() override LOCKS_EXCLUDED(name_server_list_mtx_);
-
- void injectHttpClient(std::unique_ptr<HttpClient> http_client);
-
-private:
- std::string endpoint_;
-
- SchedulerSharedPtr scheduler_;
- std::chrono::milliseconds refresh_interval_;
-
- void fetch();
-
- void onNameServerListFetched(const std::vector<std::string>& name_server_list) LOCKS_EXCLUDED(name_server_list_mtx_);
-
- std::vector<std::string> name_server_list_ GUARDED_BY(name_server_list_mtx_);
- absl::Mutex name_server_list_mtx_;
-
- std::atomic<std::uint32_t> index_{0};
-
- bool ssl_{false};
- std::unique_ptr<TopAddressing> top_addressing_;
-
- NamingScheme naming_scheme_;
-};
-
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/third_party/cpp_httplib.BUILD b/cpp/third_party/cpp_httplib.BUILD
deleted file mode 100644
index cf425f7..0000000
--- a/cpp/third_party/cpp_httplib.BUILD
+++ /dev/null
@@ -1,17 +0,0 @@
-load("@rules_cc//cc:defs.bzl", "cc_library")
-
-cc_library(
- name = "cpp_httplib",
- hdrs = [
- "httplib.h",
- ],
- visibility = [
- "//visibility:public",
- ],
- deps = [
- "//external:madler_zlib",
- ],
- defines = [
- "CPPHTTPLIB_ZLIB_SUPPORT",
- ],
-)
\ No newline at end of file
diff --git a/cpp/third_party/curl.BUILD b/cpp/third_party/curl.BUILD
deleted file mode 100644
index 4b39657..0000000
--- a/cpp/third_party/curl.BUILD
+++ /dev/null
@@ -1,35 +0,0 @@
-load("@org_apache_rocketmq//bazel:curl.bzl", "CURL_COPTS")
-
-package(features = ["no_copts_tokenization"])
-
-config_setting(
- name = "windows",
- values = {"cpu": "x64_windows"},
- visibility = ["//visibility:private"],
-)
-
-config_setting(
- name = "osx",
- values = {"cpu": "darwin"},
- visibility = ["//visibility:private"],
-)
-
-cc_library(
- name = "curl",
- srcs = glob([
- "lib/**/*.c",
- ]),
- hdrs = glob([
- "include/curl/*.h",
- "lib/**/*.h",
- ]),
- copts = CURL_COPTS + [
- "-DOS=\"os\"",
- "-DCURL_EXTERN_SYMBOL=__attribute__((__visibility__(\"default\")))",
- ],
- includes = [
- "include/",
- "lib/",
- ],
- visibility = ["//visibility:public"],
-)