This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 1bcd8341b0 [INLONG-8887][SDK] Optimize api framework for dataproxy cpp
sdk (#8888)
1bcd8341b0 is described below
commit 1bcd8341b06e3e0cba88a875dcc26dd08353b8a8
Author: doleyzi <[email protected]>
AuthorDate: Mon Sep 11 17:43:21 2023 +0800
[INLONG-8887][SDK] Optimize api framework for dataproxy cpp sdk (#8888)
---
.../dataproxy-sdk-cpp/release/inc/inlong_api.h | 55 +++++++
.../dataproxy-sdk-cpp/src/core/api_imp.cc | 165 +++++++++++++++++++++
.../dataproxy-sdk-cpp/src/core/api_imp.h | 70 +++++++++
.../dataproxy-sdk-cpp/src/core/inlong_api.cc | 41 +++++
.../dataproxy-sdk-cpp/src/utils/atomic.h | 71 +++++++++
.../dataproxy-sdk-cpp/src/utils/utils.h | 2 +-
6 files changed, 403 insertions(+), 1 deletion(-)
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/inlong_api.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/inlong_api.h
new file mode 100644
index 0000000000..23ca465359
--- /dev/null
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/inlong_api.h
@@ -0,0 +1,55 @@
+/**
+ * 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.
+ */
+
+#ifndef INLONG_SDK_API_H
+#define INLONG_SDK_API_H
+
+#include <clocale>
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <vector>
+
+namespace inlong {
+
+using UserCallBack =
+ std::function<int32_t(const char *, const char *, const char *, int32_t,
+ const int64_t, const char *)>;
+
+class InLongApiImp;
+
+class InLongApi {
+public:
+ InLongApi();
+ ~InLongApi();
+ int32_t InitApi(const char *config_path);
+
+ int32_t AddBid(const std::vector<std::string> &groupids);
+
+ int32_t Send(const char *inlong_group_id, const char *inlong_stream_id,
+ const char *msg, int32_t msg_len,
+ UserCallBack call_back = nullptr);
+
+ int32_t CloseApi(int32_t max_waitms);
+
+private:
+ std::shared_ptr<InLongApiImp> api_impl_;
+};
+} // namespace inlong
+#endif // INLONG_SDK_API_H
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/core/api_imp.cc
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/core/api_imp.cc
new file mode 100644
index 0000000000..b80f1cc40f
--- /dev/null
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/core/api_imp.cc
@@ -0,0 +1,165 @@
+/*
+ * 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 "api_imp.h"
+#include "../manager/proxy_manager.h"
+#include "../utils/logger.h"
+#include "../utils/utils.h"
+#include "api_code.h"
+#include <iostream>
+#include <signal.h>
+
+namespace inlong {
+int32_t ApiImp::InitApi(const char *config_file_path) {
+ if (!__sync_bool_compare_and_swap(&inited_, false, true)) {
+ return SdkCode::kMultiInit;
+ }
+
+ user_exit_flag_.getAndSet(0);
+
+ if (!SdkConfig::getInstance()->ParseConfig(config_file_path)) {
+ LOG_ERROR("ParseConfig error ");
+ return SdkCode::kErrorInit;
+ }
+ max_msg_length_ = std::min(SdkConfig::getInstance()->max_msg_size_,
+ SdkConfig::getInstance()->pack_size_);
+ local_ip_ = SdkConfig::getInstance()->ser_ip_;
+
+ return DoInit();
+}
+
+int32_t ApiImp::Send(const char *business_id, const char *table_id,
+ const char *msg, int32_t msg_len, UserCallBack call_back)
{
+ if (msg_len > max_msg_length_) {
+ return SdkCode::kMsgTooLong;
+ }
+ if (business_id == nullptr || table_id == nullptr || msg == nullptr ||
+ msg_len <= 0) {
+ return SdkCode::kInvalidInput;
+ }
+
+ if (inited_ == false) {
+ return SdkCode::kSendBeforeInit;
+ }
+
+ int64_t msg_time = Utils::getCurrentMsTime();
+ return this->SendBase(business_id, table_id, local_ip_, msg_time,
+ {msg, msg_len}, call_back);
+}
+
+int32_t ApiImp::SendBase(const std::string inlong_group_id,
+ const std::string inlong_stream_id,
+ const std::string client_ip, int64_t report_time,
+ const std::string msg, UserCallBack call_back) {
+ int32_t check_ret = CheckData(inlong_group_id, inlong_stream_id, msg);
+ if (check_ret != SdkCode::kSuccess) {
+ return check_ret;
+ }
+
+ ProxyManager::GetInstance()->CheckBidConf(inlong_group_id, true);
+
+ auto recv_group =
+ recv_manager_->GetRecvGroup(inlong_group_id, inlong_stream_id);
+ if (recv_group == nullptr) {
+ LOG_ERROR("fail to get pack queue, inlong_group_id:%s, inlong_stream_id:%s"
+ << inlong_group_id.c_str() << inlong_stream_id.c_str());
+ return SdkCode::kFailGetRevGroup;
+ }
+
+ return recv_group->SendData(msg, inlong_group_id, inlong_stream_id,
client_ip,
+ report_time, call_back);
+}
+
+int32_t ApiImp::CloseApi(int32_t max_waitms) {
+ if (!__sync_bool_compare_and_swap(&init_flag_, false, true)) {
+ LOG_ERROR("sdk has been closed! .");
+ return SdkCode::kMultiExits;
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(max_waitms));
+ return SdkCode::kSuccess;
+}
+
+int32_t ApiImp::DoInit() {
+ LOG_INFO(
+ "tdbus sdk cpp start Init, version:" << constants::kTDBusCAPIVersion);
+
+ signal(SIGPIPE, SIG_IGN);
+
+ LOG_INFO("tdbus_sdk_cpp Init complete!");
+
+ ProxyManager::GetInstance()->Init();
+
+ for (int i = 0; i < SdkConfig::getInstance()->inlong_group_ids_.size(); i++)
{
+ LOG_INFO("DoInit CheckBidConf inlong_group_id:"
+ << SdkConfig::getInstance()->inlong_group_ids_[i]);
+ ProxyManager::GetInstance()->CheckBidConf(
+ SdkConfig::getInstance()->inlong_group_ids_[i], false);
+ }
+
+ return InitManager();
+}
+
+int32_t ApiImp::CheckData(const std::string inlong_group_id,
+ const std::string inlong_stream_id,
+ const std::string msg) {
+ if (init_succeed_ == 0 || user_exit_flag_.get() == 1) {
+ LOG_ERROR("capi has been closed, Init first and then send");
+ return SdkCode::kSendAfterClose;
+ }
+
+ if (msg.empty() || inlong_group_id.empty() || inlong_stream_id.empty()) {
+ LOG_ERROR("invalid input, inlong_group_id"
+ << inlong_group_id << " inlong_stream_id" << inlong_stream_id
+ << "msg" << msg);
+ return SdkCode::kInvalidInput;
+ }
+
+ if (msg.size() > SdkConfig::getInstance()->max_msg_size_) {
+ LOG_ERROR("msg len is too long, cur msg_len"
+ << msg.size() << " ext_pack_size"
+ << SdkConfig::getInstance()->max_msg_size_);
+ return SdkCode::kMsgTooLong;
+ }
+
+ return SdkCode::kSuccess;
+}
+int32_t ApiImp::InitManager() {
+ send_manager_ = std::make_shared<SendManager>();
+ if (!send_manager_) {
+ LOG_ERROR("fail to Init global buffer pools");
+ return SdkCode::kErrorInit;
+ }
+
+ recv_manager_ = std::make_shared<RecvManager>(send_manager_);
+ if (!recv_manager_) {
+ LOG_ERROR("fail to Init global packqueue");
+ return SdkCode::kErrorInit;
+ }
+ init_succeed_ = true;
+ return SdkCode::kSuccess;
+}
+int32_t ApiImp::AddBid(const std::vector<std::string> &inlong_group_ids) {
+ if (inited_ == false) {
+ return SdkCode::kSendBeforeInit;
+ }
+ for (auto inlong_group_id : inlong_group_ids) {
+ ProxyManager::GetInstance()->CheckBidConf(inlong_group_id, false);
+ }
+}
+ApiImp::ApiImp() = default;
+ApiImp::~ApiImp() = default;
+} // namespace inlong
\ No newline at end of file
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/core/api_imp.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/core/api_imp.h
new file mode 100644
index 0000000000..80372e7c8d
--- /dev/null
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/core/api_imp.h
@@ -0,0 +1,70 @@
+/*
+ * 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.
+ */
+
+#ifndef INLONG_SDK_API_IMP_H
+#define INLONG_SDK_API_IMP_H
+
+#include "../manager/recv_manager.h"
+#include "../manager/send_manager.h"
+#include "../utils/atomic.h"
+#include "sdk_conf.h"
+#include <cstdint>
+#include <functional>
+
+namespace inlong {
+using UserCallBack =
+ std::function<int32_t(const char *, const char *, const char *, int32_t,
+ const int64_t, const char *)>;
+
+class ApiImp {
+public:
+ ApiImp();
+ ~ApiImp();
+ int32_t InitApi(const char *config_file_path);
+
+ int32_t Send(const char *business_id, const char *table_id, const char *msg,
+ int32_t msg_len, UserCallBack call_back = nullptr);
+
+ int32_t CloseApi(int32_t max_waitms);
+
+ int32_t AddBid(const std::vector<std::string> &bids);
+
+private:
+ int32_t DoInit();
+ int32_t InitManager();
+ int32_t SendBase(const std::string inlong_group_id,
+ const std::string inlong_stream_id,
+ const std::string client_ip, int64_t report_time,
+ const std::string msg, UserCallBack call_back);
+
+ int32_t CheckData(const std::string inlong_group_id,
+ const std::string inlong_stream_id, const std::string msg);
+
+ AtomicInt user_exit_flag_{0};
+ volatile bool init_flag_ = false;
+ volatile bool inited_ = false;
+ volatile bool init_succeed_ = false;
+ AtomicInt buf_full_{0};
+ uint32_t max_msg_length_;
+ std::string local_ip_;
+
+ std::shared_ptr<RecvManager> recv_manager_;
+ std::shared_ptr<SendManager> send_manager_;
+};
+
+} // namespace inlong
+#endif // INLONG_SDK_API_IMP_H
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/core/inlong_api.cc
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/core/inlong_api.cc
new file mode 100644
index 0000000000..0a8b75fa07
--- /dev/null
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/core/inlong_api.cc
@@ -0,0 +1,41 @@
+/*
+ * 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 "inlong_api.h"
+#include "../core/api_imp.h"
+namespace inlong {
+
+InLongApi::InLongApi() { api_impl_ = std::make_shared<InLongApiImp>(); };
+InLongApi::~InLongApi() { api_impl_->CloseApi(10); }
+
+int32_t InLongApi::InitApi(const char *config_path) {
+ return api_impl_->InitApi(config_path);
+}
+
+int32_t InLongApi::Send(const char *business_id, const char *table_id,
+ const char *msg, int32_t msg_len,
+ UserCallBack call_back) {
+ return api_impl_->Send(business_id, table_id, msg, msg_len, call_back);
+}
+
+int32_t InLongApi::CloseApi(int32_t max_waitms) {
+ return api_impl_->CloseApi(max_waitms);
+}
+int32_t InLongApi::AddBid(const std::vector<std::string> &groupids) {
+ return api_impl_->AddBid(groupids);
+}
+} // namespace inlong
\ No newline at end of file
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/atomic.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/atomic.h
new file mode 100644
index 0000000000..3a1782cabd
--- /dev/null
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/atomic.h
@@ -0,0 +1,71 @@
+/**
+ * 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.
+ */
+
+#ifndef INLONG_SDK_ATOMIC_H
+#define INLONG_SDK_ATOMIC_H
+
+#include <stdint.h>
+
+#include "noncopyable.h"
+namespace inlong {
+template <typename T> class AtomicIntegerT : noncopyable {
+private:
+ volatile T value_;
+
+public:
+ AtomicIntegerT() : value_(0) {}
+
+ explicit AtomicIntegerT(T value) : value_(value) {}
+
+ // if value_ equals oldval, update it as newval and return true;
+ // otherwise, return false
+ inline bool compareAndSwap(T oldval, T newval) {
+ return __sync_bool_compare_and_swap(&value_, oldval, newval);
+ }
+
+ inline T get() { return __sync_val_compare_and_swap(&value_, 0, 0); }
+
+ inline T getAndAdd(T x) { return __sync_fetch_and_add(&value_, x); }
+
+ inline T getAndIncrease() { return __sync_fetch_and_add(&value_, 1); }
+
+ inline T addAndGet(T x) { return getAndAdd(x) + x; }
+
+ inline T incrementAndGet() { return addAndGet(1); }
+
+ inline T decrementAndGet() { return addAndGet(-1); }
+
+ inline void add(T x) { getAndAdd(x); }
+
+ inline void increment() { incrementAndGet(); }
+
+ inline void decrement() { decrementAndGet(); }
+
+ inline T getAndSet(T newValue) {
+ return __sync_lock_test_and_set(&value_, newValue);
+ }
+};
+
+using AtomicInt = AtomicIntegerT<int32_t>;
+
+using AtomicUInt = AtomicIntegerT<uint32_t>;
+
+} // namespace inlong
+
+#endif // INLONG_SDK_ATOMIC_H
\ No newline at end of file
diff --git a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/utils.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/utils.h
index 4ee96262a0..5915ae63e1 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/utils.h
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/utils.h
@@ -30,7 +30,7 @@
#include <vector>
#include "snappy.h"
-namespace dataproxy_sdk {
+namespace inlong {
using PAIR = std::pair<std::string, int32_t>;
struct HttpRequest {
std::string url;
