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 34edf6e4fe [INLONG-8905][SDK] Reformat the code for dataproxy cpp sdk
(#8906)
34edf6e4fe is described below
commit 34edf6e4fe190c1f1439685b661a755d24ce5f0e
Author: doleyzi <[email protected]>
AuthorDate: Wed Sep 13 11:37:50 2023 +0800
[INLONG-8905][SDK] Reformat the code for dataproxy cpp sdk (#8906)
---
.../dataproxy-sdk-cpp/README.md | 70 ++--
.../release/conf/config_example.json | 60 ++--
.../dataproxy-sdk-cpp/release/demo/send_demo.cc | 101 +++---
.../dataproxy-sdk-cpp/release/inc/api_code.h | 46 +--
.../dataproxy-sdk-cpp/release/inc/inlong_api.h | 5 +-
.../dataproxy-sdk-cpp/release/inc/sdk_conf.h | 53 ++-
.../dataproxy-sdk-cpp/release/inc/sdk_msg.h | 52 ++-
.../dataproxy-sdk-cpp/src/config/ini_help.cc | 372 ++++++++++-----------
.../dataproxy-sdk-cpp/src/config/ini_help.h | 98 +++---
.../dataproxy-sdk-cpp/src/config/sdk_conf.cc | 42 +--
.../dataproxy-sdk-cpp/src/core/api_imp.cc | 18 +-
.../dataproxy-sdk-cpp/src/core/api_imp.h | 13 +-
.../dataproxy-sdk-cpp/src/core/inlong_api.cc | 11 +-
.../dataproxy-sdk-cpp/src/group/recv_group.cc | 36 +-
.../dataproxy-sdk-cpp/src/group/recv_group.h | 9 +-
.../dataproxy-sdk-cpp/src/group/send_group.cc | 3 +-
.../dataproxy-sdk-cpp/src/group/send_group.h | 2 +-
.../dataproxy-sdk-cpp/src/manager/proxy_manager.cc | 18 +-
.../dataproxy-sdk-cpp/src/manager/proxy_manager.h | 6 +-
.../dataproxy-sdk-cpp/src/manager/send_manager.cc | 8 +-
.../dataproxy-sdk-cpp/src/manager/send_manager.h | 2 +-
.../dataproxy-sdk-cpp/src/protocol/msg_protocol.cc | 34 +-
.../dataproxy-sdk-cpp/src/protocol/msg_protocol.h | 34 +-
.../dataproxy-sdk-cpp/src/utils/capi_constant.h | 13 +-
.../dataproxy-sdk-cpp/src/utils/read_write_mutex.h | 2 +-
.../dataproxy-sdk-cpp/src/utils/send_buffer.h | 17 +-
.../dataproxy-sdk-cpp/src/utils/utils.h | 90 ++---
.../dataproxy-sdk-cpp/third_party/CMakeLists.txt | 76 ++---
28 files changed, 637 insertions(+), 654 deletions(-)
diff --git a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/README.md
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/README.md
index 644186ea09..c136444328 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/README.md
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/README.md
@@ -43,44 +43,52 @@ Go to the dataproxy-sdk-cpp root, and run
Refer to `release/conf/config_example.json`.
-|name|default value|description|
-|:---|:---|:---|
-|thread_num|10|number of network sending threads|
-|inlong_group_ids|""|the list of inlong_group_id, seperated by commas, such as
"b_inlong_group_test_01, b_inlong_group_test_02"|
-|enable_groupId_isolation|false|whether different groupid data using different
buffer pools inside the sdk|
-|buffer_num_per_groupId|5|number of buffer pools of each groupid|
-|enable_pack|true|whether multiple messages are packed while sending to
dataproxy|
-|pack_size|4096|byte, pack messages and send to dataproxy when the data in
buffer pool exceeds this value|
-|ext_pack_size|16384|byte, maximum length of a message|
-|enable_zip|true|whether zip data while sending to dataproxy|
-|min_ziplen|512|byte, minimum zip len|
-|enable_retry|true|whether do resend while failed to send data|
-|retry_ms|3000|millisecond, resend interval|
-|retry_num|3|maximum resend times|
-|max_active_proxy|3|maximum number of established connections with dataproxy|
-|max_buf_pool|50 `*`1024`*` 1024|byte, the size of buffer pool|
-|log_num|10|maximum number of log files|
-|log_size|10|MB, maximum size of one log file|
-|log_level|2|log level: trace(4)>debug(3)>info(2)>warn(1)>error(0)|
-|log_file_type|2|type of log output: 2->file, 1->console|
-|log_path|./logs/|log path|
-|proxy_update_interval|10|interval of requesting and updating dataproxy lists
from manager|
-|proxy_cfg_preurl|"http://127.0.0.1:8099/inlong/manager/openapi/dataproxy/getIpList";|the
url of manager openapi|
-|need_auth|false|whether need authentication while interacting with manager|
-|auth_id|""|authenticate id if need authentication|
-|auth_key|""|authenticate key if need authentication|
+| name | default value
| description
|
+|:-------------------------|:-------------------------------------------------------------------|:-----------------------------------------------------------------------------------------------------------|
+| thread_num | 10
| number of network sending threads
|
+| inlong_group_ids | ""
| the list of inlong_group_id, seperated by commas, such as
"b_inlong_group_test_01, b_inlong_group_test_02" |
+| enable_groupId_isolation | false
| whether different groupid data using different buffer pools
inside the sdk |
+| buffer_num_per_groupId | 5
| number of buffer pools of each groupid
|
+| enable_pack | true
| whether multiple messages are packed while sending to
dataproxy |
+| pack_size | 4096
| byte, pack messages and send to dataproxy when the data in
buffer pool exceeds this value |
+| ext_pack_size | 16384
| byte, maximum length of a message
|
+| enable_zip | true
| whether zip data while sending to dataproxy
|
+| min_ziplen | 512
| byte, minimum zip len
|
+| enable_retry | true
| whether do resend while failed to send data
|
+| retry_ms | 3000
| millisecond, resend interval
|
+| retry_num | 3
| maximum resend times
|
+| max_active_proxy | 3
| maximum number of established connections with dataproxy
|
+| max_buf_pool | 50 `*`1024`*` 1024
| byte, the size of buffer pool
|
+| log_num | 10
| maximum number of log files
|
+| log_size | 10
| MB, maximum size of one log file
|
+| log_level | 2
| log level: trace(4)>debug(3)>info(2)>warn(1)>error(0)
|
+| log_file_type | 2
| type of log output: 2->file, 1->console
|
+| log_path | ./logs/
| log path
|
+| proxy_update_interval | 10
| interval of requesting and updating dataproxy lists from
manager |
+| manager_url |
"http://127.0.0.1:8099/inlong/manager/openapi/dataproxy/getIpList"; | the url of
manager openapi
|
+| need_auth | false
| whether need authentication while interacting with manager
|
+| auth_id | ""
| authenticate id if need authentication
|
+| auth_key | ""
| authenticate key if need authentication
|
## Usage
1. First, init dataproxy-sdk, there are two ways you can choose:
-- A) `int32_t tc_api_init(const char* config_file)`. Here, `config_file` is
the path of your config file, and absolute path is recommended. Note that only
once called is needed in one process.
-- B) `int32_t tc_api_init(ClientConfig& client_config)`. Here, `client_config`
is the pointer of a `ClientConfig` object.
-2. Then, send data: `int32_t tc_api_send(const char* inlong_group_id, const
char* inlong_stream_id, const char* msg, int32_t msg_len, UserCallBack
call_back = NULL)`. If you set `call_back`, it will be callbacked if your data
failed to send. See the signature of `UserCallBack` in `release/inc/user_msg.h`.
+- A) `int32_t tc_api_init(const char* config_file)`. Here, `config_file` is
the path of your config file, and absolute
+ path is recommended. Note that only once called is needed in one process.
+- B) `int32_t tc_api_init(ClientConfig& client_config)`. Here, `client_config`
is the pointer of a `ClientConfig`
+ object.
-3. Finally, close sdk if no more data to be sent: `int32_t
tc_api_close(int32_t max_waitms)`. Here, `max_waitms` is the interval of
waiting data in memory to be sent.
+2. Then, send
+ data: `int32_t tc_api_send(const char* inlong_group_id, const char*
inlong_stream_id, const char* msg, int32_t msg_len, UserCallBack call_back =
NULL)`.
+ If you set `call_back`, it will be callbacked if your data failed to send.
See the signature of `UserCallBack`
+ in `release/inc/user_msg.h`.
-4. Note, the above functions return 0 if success, otherwise it means failure.
As for other return results, please refer to `SDKInvalidResult` in
`release/inc/tc_api.h`.
+3. Finally, close sdk if no more data to be sent: `int32_t
tc_api_close(int32_t max_waitms)`. Here, `max_waitms` is the
+ interval of waiting data in memory to be sent.
+
+4. Note, the above functions return 0 if success, otherwise it means failure.
As for other return results, please refer
+ to `SDKInvalidResult` in `release/inc/tc_api.h`.
## Demo
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/conf/config_example.json
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/conf/config_example.json
index c4ae2630de..3c94362bb9 100644
---
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/conf/config_example.json
+++
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/conf/config_example.json
@@ -1,32 +1,32 @@
{
- "init-param": {
- "thread_num": 5,
- "inlong_group_ids": "b_inlong_group_test_01, b_inlong_group_test_02",
- "enable_groupId_isolation": false,
- "shared_buf_num": 5,
- "buffer_num_per_groupId": 5,
- "enable_pack": true,
- "pack_size": 4096,
- "ext_pack_size": 4096,
- "enable_zip": true,
- "min_ziplen": 300,
- "enable_retry": true,
- "retry_ms": 3000,
- "retry_num": 3,
- "max_active_proxy": 4,
- "max_buf_pool": 548576000,
- "msg_type": 7,
- "log_num": 10,
- "log_size": 10,
- "log_level": 3,
- "log_file_type": 2,
- "log_path": "./",
- "enable_hb": true,
- "hb_interval": 60,
- "proxy_update_interval": 10,
- "manager_url":
"http://127.0.0.1:8099/inlong/manager/openapi/dataproxy/getIpList";,
- "need_auth": true,
- "auth_id": "admin",
- "auth_key": "87haw3VYTPqK5fK0"
- }
+ "init-param": {
+ "thread_num": 5,
+ "inlong_group_ids": "b_inlong_group_test_01, b_inlong_group_test_02",
+ "enable_groupId_isolation": false,
+ "shared_buf_num": 5,
+ "buffer_num_per_groupId": 5,
+ "enable_pack": true,
+ "pack_size": 4096,
+ "ext_pack_size": 4096,
+ "enable_zip": true,
+ "min_ziplen": 300,
+ "enable_retry": true,
+ "retry_ms": 3000,
+ "retry_num": 3,
+ "max_active_proxy": 4,
+ "max_buf_pool": 548576000,
+ "msg_type": 7,
+ "log_num": 10,
+ "log_size": 10,
+ "log_level": 3,
+ "log_file_type": 2,
+ "log_path": "./",
+ "enable_hb": true,
+ "hb_interval": 60,
+ "proxy_update_interval": 10,
+ "manager_url":
"http://127.0.0.1:8099/inlong/manager/openapi/dataproxy/getIpList";,
+ "need_auth": true,
+ "auth_id": "admin",
+ "auth_key": "87haw3VYTPqK5fK0"
+ }
}
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/demo/send_demo.cc
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/demo/send_demo.cc
index 1c07fb99ac..a4a783e0dc 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/demo/send_demo.cc
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/demo/send_demo.cc
@@ -17,72 +17,69 @@
* under the License.
*/
+#include "inlong_api.h"
#include <chrono>
#include <iostream>
#include <string>
-#include "inlong_api.h"
using namespace std;
using namespace inlong;
// user set call_back func
-int call_back_func(const char *inlong_group_id, const char *inlong_stream_id,
const char *msg, int msg_len, long report_time, const char *ip)
-{
- cout << "******this is call back, print info******" << endl;
- cout << "inlong_group_id: " << inlong_group_id << ", inlong_stream_id: "
<< inlong_stream_id << endl;
- cout << "msg_len: " << msg_len << ", msg content: " << msg << endl;
- cout << "report_time: " << report_time << ", client ip: " << ip << endl;
- cout << "******call back end******" << endl;
+int call_back_func(const char *inlong_group_id, const char *inlong_stream_id,
+ const char *msg, int msg_len, long report_time,
+ const char *ip) {
+ cout << "******this is call back, print info******" << endl;
+ cout << "inlong_group_id: " << inlong_group_id
+ << ", inlong_stream_id: " << inlong_stream_id << endl;
+ cout << "msg_len: " << msg_len << ", msg content: " << msg << endl;
+ cout << "report_time: " << report_time << ", client ip: " << ip << endl;
+ cout << "******call back end******" << endl;
- return 0;
+ return 0;
}
-int main(int argc, char const *argv[])
-{
- if ( argc <2)
- {
- cout << "USAGE: ./send_demo ../config/config_example.json" << endl;
- return 0;
- }
- InLongApi inlong_api;
- // step1. init
- if (inlong_api.InitApi(argv[1]))
- {
- cout << "init error" << endl;
- return 0;
- }
- cout << "---->start sdk successfully" << endl;
+int main(int argc, char const *argv[]) {
+ if (argc < 2) {
+ cout << "USAGE: ./send_demo ../config/config_example.json" << endl;
+ return 0;
+ }
+ InLongApi inlong_api;
+ // step1. init
+ if (inlong_api.InitApi(argv[1])) {
+ cout << "init error" << endl;
+ return 0;
+ }
+ cout << "---->start sdk successfully" << endl;
- int count = 1000;
- string inlong_group_id = "test_cpp_sdk_20230404";
- string inlong_stream_id = "stream1";
- if ( 4 == argc) {
- inlong_group_id = argv[2];
- inlong_stream_id = argv[3];
- }
- cout << "inlong_group_id:
"<<inlong_group_id<<",inlong_stream_id:"<<inlong_stream_id << endl;
- string msg = "this is a test ttttttttttttttt; eiwhgreuhg
jfdiowaehgorerlea; test end";
+ int count = 1000;
+ string inlong_group_id = "test_cpp_sdk_20230404";
+ string inlong_stream_id = "stream1";
+ if (4 == argc) {
+ inlong_group_id = argv[2];
+ inlong_stream_id = argv[3];
+ }
+ cout << "inlong_group_id: " << inlong_group_id
+ << ",inlong_stream_id:" << inlong_stream_id << endl;
+ string msg =
+ "this is a test ttttttttttttttt; eiwhgreuhg jfdiowaehgorerlea; test end";
- // step2. send
- cout << "---->start tc_api_send" << endl;
- for (size_t i = 0; i < count; i++)
- {
- if (inlong_api.Send(inlong_group_id.c_str(), inlong_stream_id.c_str(),
msg.c_str(), msg.length(), call_back_func))
- {
- cout << "tc_api_send error;"
- << " ";
- }
+ // step2. send
+ cout << "---->start tc_api_send" << endl;
+ for (size_t i = 0; i < count; i++) {
+ if (inlong_api.Send(inlong_group_id.c_str(), inlong_stream_id.c_str(),
+ msg.c_str(), msg.length(), call_back_func)) {
+ cout << "tc_api_send error;"
+ << " ";
}
+ }
- // step3. close
- if (inlong_api.CloseApi(1000))
- {
- cout << "close sdk error" << endl;
- }
- else
- {
- cout << "---->close sdk successfully" << endl;
- }
+ // step3. close
+ if (inlong_api.CloseApi(10000)) {
+ cout << "close sdk error" << endl;
+ } else {
+ cout << "---->close sdk successfully" << endl;
+ }
- return 0;
+ return 0;
}
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/api_code.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/api_code.h
index ee9640f396..227f6f9e24 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/api_code.h
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/api_code.h
@@ -1,23 +1,23 @@
/*
-* 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.
-*/
+ * 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_CODE_H
#define INLONG_SDK_API_CODE_H
-namespace inlong{
+namespace inlong {
enum SdkCode {
kSuccess = 0,
kMultiInit = 4,
@@ -34,15 +34,15 @@ enum SdkCode {
kFailWriteToBuf = 15,
kErrorCURL = 16,
kErrorParseJson = 17,
- kFailGetRevGroup = 18 ,
- kFailGetBusConf = 19 ,
- kFailSendBus = 20 ,
+ kFailGetRevGroup = 18,
+ kFailGetProxyConf = 19,
+ kFailSendProxy = 20,
kFailSParseResponse = 21,
- kSendBeforeInit= 22,
- kFailMallocBuf= 23,
- kMsgSizeLargerThanPackSize= 24,
+ kSendBeforeInit = 22,
+ kFailMallocBuf = 23,
+ kMsgSizeLargerThanPackSize = 24,
kSendBufferFull = 25
};
}
-#endif // INLONG_SDK_API_CODE_H
+#endif // INLONG_SDK_API_CODE_H
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
index 458155d9bc..66a1a7ce39 100644
--- 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
@@ -28,9 +28,8 @@
namespace inlong {
-using UserCallBack =
- std::function<int32_t(const char *, const char *, const char *, int32_t,
- const int64_t, const char *)>;
+typedef int (*UserCallBack)(const char *, const char *, const char *, int32_t,
+ const int64_t, const char *);
class ApiImp;
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/sdk_conf.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/sdk_conf.h
index 1beb05ea22..d9bf4b16c3 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/sdk_conf.h
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/sdk_conf.h
@@ -1,21 +1,21 @@
/**
-* 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.
-*/
+ * 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 CAPI_BASE_CLIENT_CONFIG_H_
#define CAPI_BASE_CLIENT_CONFIG_H_
@@ -36,21 +36,22 @@ private:
void InitCacheParam(const rapidjson::Value &doc);
void InitZipParam(const rapidjson::Value &doc);
void InitLogParam(const rapidjson::Value &doc);
- void InitBusParam(const rapidjson::Value &doc);
+ void InitManagerParam(const rapidjson::Value &doc);
void InitTcpParam(const rapidjson::Value &doc);
void InitAuthParm(const rapidjson::Value &doc);
void OthersParam(const rapidjson::Value &doc);
public:
// cache parameter
- std::vector<std::string> inlong_group_ids_; // Initialize the inlong groupid
collection
- uint32_t recv_buf_size_; // Receive buf size, tid granularity
- uint32_t send_buf_size_; // Send buf size, bid granularity
+ std::vector<std::string>
+ inlong_group_ids_; // Initialize the inlong groupid collection
+ uint32_t recv_buf_size_; // Receive buf size, tid granularity
+ uint32_t send_buf_size_; // Send buf size, bid granularity
// thread parameters
- uint32_t per_groupid_thread_nums_; // Sending thread per groupid
- uint32_t dispatch_interval_zip_; // Compression thread distribution interval
- uint32_t dispatch_interval_send_; // sending thread sending interval
+ uint32_t per_groupid_thread_nums_; // Sending thread per groupid
+ uint32_t dispatch_interval_zip_; // Compression thread distribution
interval
+ uint32_t dispatch_interval_send_; // sending thread sending interval
// Packaging parameters
bool enable_pack_;
@@ -107,9 +108,7 @@ public:
bool ParseConfig(const std::string &config_path);
void ShowClientConfig();
- inline bool enableChar() const {
- return (((extend_field_)&0x4) >> 2);
- }
+ inline bool enableChar() const { return (((extend_field_)&0x4) >> 2); }
inline bool enableTraceIP() const { return (((extend_field_)&0x2) >> 1); }
// data type message datlen|data
inline bool isNormalDataPackFormat() const {
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/sdk_msg.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/sdk_msg.h
index 1114eb17db..ae77b7efcf 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/sdk_msg.h
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/release/inc/sdk_msg.h
@@ -1,29 +1,30 @@
/*
-* 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.
+ * 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 SDK_USER_MSG_H_
#define SDK_USER_MSG_H_
-#include <stdint.h>
#include <functional>
#include <memory>
+#include <stdint.h>
#include <string>
namespace inlong {
-using UserCallBack = std::function<int32_t(const char*, const char*, const
char*, int32_t, const int64_t, const char*)>;
+typedef int (*UserCallBack)(const char *, const char *, const char *, int32_t,
+ const int64_t, const char *);
struct SdkMsg {
std::string msg_;
@@ -35,18 +36,15 @@ struct SdkMsg {
std::string user_client_ip_;
std::string data_pack_format_attr_;
- SdkMsg(const std::string& mmsg, const std::string& mclient_ip, int64_t
mreport_time, UserCallBack mcb,
- const std::string& attr, const std::string& u_ip, int64_t u_time)
- : msg_(mmsg),
- client_ip_(mclient_ip),
- report_time_(mreport_time),
- cb_(mcb),
- user_report_time_(u_time),
- user_client_ip_(u_ip),
- data_pack_format_attr_(attr){}
+ SdkMsg(const std::string &mmsg, const std::string &mclient_ip,
+ int64_t mreport_time, UserCallBack mcb, const std::string &attr,
+ const std::string &u_ip, int64_t u_time)
+ : msg_(mmsg), client_ip_(mclient_ip), report_time_(mreport_time),
+ cb_(mcb), user_report_time_(u_time), user_client_ip_(u_ip),
+ data_pack_format_attr_(attr) {}
};
using SdkMsgPtr = std::shared_ptr<SdkMsg>;
-} // namespace inlong
+} // namespace inlong
-#endif // SDK_USER_MSG_H_
\ No newline at end of file
+#endif // SDK_USER_MSG_H_
\ No newline at end of file
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/ini_help.cc
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/ini_help.cc
index e8fa8fe466..ba13b6aa01 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/ini_help.cc
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/ini_help.cc
@@ -25,8 +25,7 @@
#include <stdio.h>
#include <stdlib.h>
-namespace inlong
-{
+namespace inlong {
/**
* @description: parse line content
@@ -35,258 +34,241 @@ namespace inlong
* @param {string*} key
* @param {string*} value
*/
-bool IniFile::parse(const std::string& content, std::string* key, std::string*
value) { return split(content, "=", key, value); }
-
-int IniFile::updateSection(const std::string& cleanLine,
- const std::string& comment,
- const std::string& rightComment,
- IniSection** section)
-{
- IniSection* newSection;
- // find ']'
- size_t index = cleanLine.find_first_of(']');
- if (index == std::string::npos)
- {
- err_msg_ = std::string("no matched ] found");
- return 1;
- }
+bool IniFile::parse(const std::string &content, std::string *key,
+ std::string *value) {
+ return split(content, "=", key, value);
+}
- int len = index - 1;
+int IniFile::updateSection(const std::string &cleanLine,
+ const std::string &comment,
+ const std::string &rightComment,
+ IniSection **section) {
+ IniSection *newSection;
+ // find ']'
+ size_t index = cleanLine.find_first_of(']');
+ if (index == std::string::npos) {
+ err_msg_ = std::string("no matched ] found");
+ return 1;
+ }
- if (len <= 0)
- {
- err_msg_ = std::string("section name is empty");
- return 1;
- }
+ int len = index - 1;
- // set section name
- std::string s(cleanLine, 1, len);
+ if (len <= 0) {
+ err_msg_ = std::string("section name is empty");
+ return 1;
+ }
- trim(s);
+ // set section name
+ std::string s(cleanLine, 1, len);
- //check section
- if (getSection(s) != NULL)
- {
- err_msg_ = std::string("section ") + s + std::string("already exist");
- return 1;
- }
+ trim(s);
+
+ // check section
+ if (getSection(s) != NULL) {
+ err_msg_ = std::string("section ") + s + std::string("already exist");
+ return 1;
+ }
- newSection = new IniSection();
- newSection->name = s;
- newSection->comment = comment;
- newSection->rightComment = rightComment;
+ newSection = new IniSection();
+ newSection->name = s;
+ newSection->comment = comment;
+ newSection->rightComment = rightComment;
- sections_.push_back(newSection);
+ sections_.push_back(newSection);
- *section = newSection;
+ *section = newSection;
- return 0;
+ return 0;
}
-int IniFile::addKV(const std::string& cleanLine,
- const std::string& comment,
- const std::string& rightComment,
- IniSection* section)
-{
- std::string key, value;
-
- if (!parse(cleanLine, &key, &value))
- {
- err_msg_ = std::string("parse line failed:") + cleanLine;
- return 1;
- }
+int IniFile::addKV(const std::string &cleanLine, const std::string &comment,
+ const std::string &rightComment, IniSection *section) {
+ std::string key, value;
+
+ if (!parse(cleanLine, &key, &value)) {
+ err_msg_ = std::string("parse line failed:") + cleanLine;
+ return 1;
+ }
- Iterm item;
- item.key = key;
- item.value = value;
- item.comment = comment;
- item.rightComment = rightComment;
+ Iterm item;
+ item.key = key;
+ item.value = value;
+ item.comment = comment;
+ item.rightComment = rightComment;
- section->items.push_back(item);
+ section->items.push_back(item);
- return 0;
+ return 0;
}
-int IniFile::load(const std::string& fileName)
-{
- int err;
- std::string line;
- std::string comment;
- std::string rightComment;
- IniSection* currSection = NULL; //init a section
-
- close();
-
- ini_file_name_ = fileName;
- std::ifstream istream(ini_file_name_);
- if (!istream.is_open())
- {
- err_msg_ = std::string("open") + ini_file_name_ + std::string(" file
failed");
- return 1;
+int IniFile::load(const std::string &fileName) {
+ int err;
+ std::string line;
+ std::string comment;
+ std::string rightComment;
+ IniSection *currSection = NULL; // init a section
+
+ close();
+
+ ini_file_name_ = fileName;
+ std::ifstream istream(ini_file_name_);
+ if (!istream.is_open()) {
+ err_msg_ =
+ std::string("open") + ini_file_name_ + std::string(" file failed");
+ return 1;
+ }
+
+ // add new section
+ currSection = new IniSection();
+ currSection->name = "";
+ sections_.push_back(currSection);
+
+ // read line
+ while (std::getline(istream, line)) {
+ trim(line);
+
+ // skip empty
+ if (line.length() <= 0) {
+ comment += delim;
+ continue;
}
- // add new section
- currSection = new IniSection();
- currSection->name = "";
- sections_.push_back(currSection);
-
- // read line
- while (std::getline(istream, line))
- {
- trim(line);
-
- // skip empty
- if (line.length() <= 0)
- {
- comment += delim;
- continue;
- }
-
- // whether contains section or key
- // find '['
- if (line[0] == '[') { err = updateSection(line, comment, rightComment,
&currSection); }
- else
- {
- err = addKV(line, comment, rightComment, currSection);
- }
-
- if (err != 0)
- {
- istream.close();
- return err;
- }
-
- // clear
- comment = "";
- rightComment = "";
+ // whether contains section or key
+ // find '['
+ if (line[0] == '[') {
+ err = updateSection(line, comment, rightComment, &currSection);
+ } else {
+ err = addKV(line, comment, rightComment, currSection);
}
- istream.close();
+ if (err != 0) {
+ istream.close();
+ return err;
+ }
- return 0;
+ // clear
+ comment = "";
+ rightComment = "";
+ }
+
+ istream.close();
+
+ return 0;
}
-IniSection* IniFile::getSection(const std::string& section)
-{
- for (SectionIterator it = sections_.begin(); it != sections_.end(); ++it)
- {
- if ((*it)->name == section) { return *it; }
+IniSection *IniFile::getSection(const std::string §ion) {
+ for (SectionIterator it = sections_.begin(); it != sections_.end(); ++it) {
+ if ((*it)->name == section) {
+ return *it;
}
+ }
- return NULL;
+ return NULL;
}
-int IniFile::getString(const std::string& section, const std::string& key,
std::string* value)
-{
- return getValue(section, key, value);
+int IniFile::getString(const std::string §ion, const std::string &key,
+ std::string *value) {
+ return getValue(section, key, value);
}
-int IniFile::getInt(const std::string& section, const std::string& key, int*
intValue)
-{
- int err;
- std::string strValue;
+int IniFile::getInt(const std::string §ion, const std::string &key,
+ int *intValue) {
+ int err;
+ std::string strValue;
- err = getValue(section, key, &strValue);
+ err = getValue(section, key, &strValue);
- *intValue = atoi(strValue.c_str());
+ *intValue = atoi(strValue.c_str());
- return err;
+ return err;
}
-int IniFile::getValue(const std::string& section, const std::string& key,
std::string* value)
-{
- std::string comment;
- return getValue(section, key, value, &comment);
+int IniFile::getValue(const std::string §ion, const std::string &key,
+ std::string *value) {
+ std::string comment;
+ return getValue(section, key, value, &comment);
}
-int IniFile::getValue(const std::string& section, const std::string& key,
std::string* value, std::string* comment)
-{
- IniSection* sect = getSection(section);
+int IniFile::getValue(const std::string §ion, const std::string &key,
+ std::string *value, std::string *comment) {
+ IniSection *sect = getSection(section);
- if (sect == NULL)
- {
- err_msg_ = std::string("not find the section ") + section;
- return 1;
- }
+ if (sect == NULL) {
+ err_msg_ = std::string("not find the section ") + section;
+ return 1;
+ }
- for (IniSection::ItermIterator it = sect->begin(); it != sect->end(); ++it)
- {
- if (it->key == key)
- {
- *value = it->value;
- *comment = it->comment;
- return 0;
- }
+ for (IniSection::ItermIterator it = sect->begin(); it != sect->end(); ++it) {
+ if (it->key == key) {
+ *value = it->value;
+ *comment = it->comment;
+ return 0;
}
+ }
- err_msg_ = std::string("not find the key ") + key;
- return 1;
+ err_msg_ = std::string("not find the key ") + key;
+ return 1;
}
-void IniFile::close()
-{
- ini_file_name_ = "";
+void IniFile::close() {
+ ini_file_name_ = "";
- for (SectionIterator it = sections_.begin(); it != sections_.end(); ++it)
- {
- delete (*it); // clear section
- }
+ for (SectionIterator it = sections_.begin(); it != sections_.end(); ++it) {
+ delete (*it); // clear section
+ }
- sections_.clear();
+ sections_.clear();
}
-void IniFile::trim(std::string& str)
-{
- int len = str.length();
+void IniFile::trim(std::string &str) {
+ int len = str.length();
- int i = 0;
+ int i = 0;
- while ((i < len) && isspace(str[i]) && (str[i] != '\0'))
- {
- i++;
- }
+ while ((i < len) && isspace(str[i]) && (str[i] != '\0')) {
+ i++;
+ }
- if (i != 0) { str = std::string(str, i, len - i); }
+ if (i != 0) {
+ str = std::string(str, i, len - i);
+ }
- len = str.length();
+ len = str.length();
- for (i = len - 1; i >= 0; --i)
- {
- if (!isspace(str[i])) { break; }
+ for (i = len - 1; i >= 0; --i) {
+ if (!isspace(str[i])) {
+ break;
}
+ }
- str = std::string(str, 0, i + 1);
+ str = std::string(str, 0, i + 1);
}
-bool IniFile::split(const std::string& str, const std::string& sep,
std::string* pleft, std::string* pright)
-{
- size_t pos = str.find(sep);
- std::string left, right;
+bool IniFile::split(const std::string &str, const std::string &sep,
+ std::string *pleft, std::string *pright) {
+ size_t pos = str.find(sep);
+ std::string left, right;
- if (pos != std::string::npos)
- {
- left = std::string(str, 0, pos);
- right = std::string(str, pos + 1);
+ if (pos != std::string::npos) {
+ left = std::string(str, 0, pos);
+ right = std::string(str, pos + 1);
- trim(left);
- trim(right);
+ trim(left);
+ trim(right);
- *pleft = left;
- *pright = right;
- return true;
- }
- else
- {
- left = str;
- right = "";
+ *pleft = left;
+ *pright = right;
+ return true;
+ } else {
+ left = str;
+ right = "";
- trim(left);
+ trim(left);
- *pleft = left;
- *pright = right;
- return false;
- }
+ *pleft = left;
+ *pright = right;
+ return false;
+ }
}
-
-} // namespace inlong
+} // namespace inlong
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/ini_help.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/ini_help.h
index d846c09da9..e0558a18e0 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/ini_help.h
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/ini_help.h
@@ -25,70 +25,60 @@
#include <string>
#include <vector>
-namespace inlong
-{
+namespace inlong {
const char delim[] = "\n";
-struct Iterm
-{
- std::string key;
- std::string value;
- std::string comment;
- std::string rightComment;
+struct Iterm {
+ std::string key;
+ std::string value;
+ std::string comment;
+ std::string rightComment;
};
-struct IniSection
-{
- using ItermIterator = std::vector<Iterm>::iterator;
- ItermIterator begin()
- {
- return items.begin();
- }
+struct IniSection {
+ using ItermIterator = std::vector<Iterm>::iterator;
+ ItermIterator begin() { return items.begin(); }
- ItermIterator end()
- {
- return items.end();
- }
+ ItermIterator end() { return items.end(); }
- std::string name;
- std::string comment;
- std::string rightComment;
- std::vector<Iterm> items;
+ std::string name;
+ std::string comment;
+ std::string rightComment;
+ std::vector<Iterm> items;
};
-class IniFile
-{
- public:
- IniFile(){}
- ~IniFile() { close(); }
+class IniFile {
+public:
+ IniFile() {}
+ ~IniFile() { close(); }
- int load(const std::string& fileName);
- int getString(const std::string& section, const std::string& key,
std::string* value);
- int getInt(const std::string& section, const std::string& key, int* value);
+ int load(const std::string &fileName);
+ int getString(const std::string §ion, const std::string &key,
+ std::string *value);
+ int getInt(const std::string §ion, const std::string &key, int *value);
- private:
- IniSection* getSection(const std::string& section = "");
- static void trim(std::string& str);
- int updateSection(const std::string& cleanLine,
- const std::string& comment,
- const std::string& rightComment,
- IniSection** section);
- int addKV(const std::string& cleanLine,
- const std::string& comment,
- const std::string& rightComment,
- IniSection* section);
- void close();
- bool split(const std::string& str, const std::string& sep, std::string*
left, std::string* right);
- bool parse(const std::string& content, std::string* key, std::string*
value);
- int getValue(const std::string& section, const std::string& key,
std::string* value);
- int getValue(const std::string& section, const std::string& key,
std::string* value, std::string* comment);
+private:
+ IniSection *getSection(const std::string §ion = "");
+ static void trim(std::string &str);
+ int updateSection(const std::string &cleanLine, const std::string &comment,
+ const std::string &rightComment, IniSection **section);
+ int addKV(const std::string &cleanLine, const std::string &comment,
+ const std::string &rightComment, IniSection *section);
+ void close();
+ bool split(const std::string &str, const std::string &sep, std::string *left,
+ std::string *right);
+ bool parse(const std::string &content, std::string *key, std::string *value);
+ int getValue(const std::string §ion, const std::string &key,
+ std::string *value);
+ int getValue(const std::string §ion, const std::string &key,
+ std::string *value, std::string *comment);
- private:
- using SectionIterator = std::vector<IniSection*>::iterator;
- std::vector<IniSection*> sections_;
- std::string ini_file_name_;
- std::string err_msg_; // save err msg
+private:
+ using SectionIterator = std::vector<IniSection *>::iterator;
+ std::vector<IniSection *> sections_;
+ std::string ini_file_name_;
+ std::string err_msg_; // save err msg
};
-} // namespace inlong
+} // namespace inlong
-#endif // INLONG_SDK_BASE_INI_HELP_H_
\ No newline at end of file
+#endif // INLONG_SDK_BASE_INI_HELP_H_
\ No newline at end of file
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/sdk_conf.cc
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/sdk_conf.cc
index d08d262067..b327e305e3 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/sdk_conf.cc
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/config/sdk_conf.cc
@@ -60,7 +60,7 @@ bool SdkConfig::ParseConfig(const std::string &config_path) {
InitCacheParam(doc);
InitZipParam(doc);
InitLogParam(doc);
- InitBusParam(doc);
+ InitManagerParam(doc);
InitTcpParam(doc);
OthersParam(doc);
@@ -98,11 +98,11 @@ void SdkConfig::defaultInit() {
// manager parameters
manager_url_ = constants::kManagerURL;
- enable_manager_url_from_cluster_ = constants::kEnableBusURLFromCluster;
- manager_cluster_url_ = constants::kBusClusterURL;
- manager_update_interval_ = constants::kBusUpdateInterval;
- manager_url_timeout_ = constants::kBusURLTimeout;
- max_proxy_num_ = constants::kMaxBusNum;
+ enable_manager_url_from_cluster_ = constants::kEnableManagerFromCluster;
+ manager_cluster_url_ = constants::kManagerClusterURL;
+ manager_update_interval_ = constants::kManagerUpdateInterval;
+ manager_url_timeout_ = constants::kManagerTimeout;
+ max_proxy_num_ = constants::kMaxProxyNum;
local_ip_ = constants::kSerIP;
local_port_ = constants::kSerPort;
@@ -242,7 +242,7 @@ void SdkConfig::InitLogParam(const rapidjson::Value &doc) {
log_path_ = constants::kLogPath;
}
}
-void SdkConfig::InitBusParam(const rapidjson::Value &doc) {
+void SdkConfig::InitManagerParam(const rapidjson::Value &doc) {
// manager url
if (doc.HasMember("manager_url") && doc["manager_url"].IsString()) {
const rapidjson::Value &obj = doc["manager_url"];
@@ -250,41 +250,41 @@ void SdkConfig::InitBusParam(const rapidjson::Value &doc)
{
} else {
manager_url_ = constants::kManagerURL;
}
- // bus_cluster_URL_
+ // manager cluster url
if (doc.HasMember("manager_cluster_url") &&
doc["manager_cluster_url"].IsString()) {
const rapidjson::Value &obj = doc["manager_cluster_url"];
manager_cluster_url_ = obj.GetString();
} else {
- manager_cluster_url_ = constants::kBusClusterURL;
+ manager_cluster_url_ = constants::kManagerClusterURL;
}
- // enable_bus_URL_from_cluster
+ // enable manager from cluster
if (doc.HasMember("enable_manager_url_from_cluster") &&
doc["enable_manager_url_from_cluster"].IsBool()) {
const rapidjson::Value &obj = doc["enable_manager_url_from_cluster"];
enable_manager_url_from_cluster_ = obj.GetBool();
} else {
- enable_manager_url_from_cluster_ = constants::kEnableBusURLFromCluster;
+ enable_manager_url_from_cluster_ = constants::kEnableManagerFromCluster;
}
- // bus_update_interval
+ // manager update interval
if (doc.HasMember("manager_update_interval") &&
doc["manager_update_interval"].IsInt() &&
doc["manager_update_interval"].GetInt() > 0) {
const rapidjson::Value &obj = doc["manager_update_interval"];
manager_update_interval_ = obj.GetInt();
} else {
- manager_update_interval_ = constants::kBusUpdateInterval;
+ manager_update_interval_ = constants::kManagerUpdateInterval;
}
- // bus_URL_timeout
+ // manager timeout
if (doc.HasMember("manager_url_timeout") &&
doc["manager_url_timeout"].IsInt() &&
doc["manager_url_timeout"].GetInt() > 0) {
const rapidjson::Value &obj = doc["manager_url_timeout"];
manager_url_timeout_ = obj.GetInt();
} else {
- manager_url_timeout_ = constants::kBusURLTimeout;
+ manager_url_timeout_ = constants::kManagerTimeout;
}
- // msg_type
+ // msg type
if (doc.HasMember("msg_type") && doc["msg_type"].IsInt() &&
doc["msg_type"].GetInt() > 0 && doc["msg_type"].GetInt() < 9) {
const rapidjson::Value &obj = doc["msg_type"];
@@ -292,13 +292,13 @@ void SdkConfig::InitBusParam(const rapidjson::Value &doc)
{
} else {
msg_type_ = constants::kMsgType;
}
- // max_active_bus_num
- if (doc.HasMember("max_tcp_num") && doc["max_tcp_num"].IsInt() &&
- doc["max_tcp_num"].GetInt() > 0) {
- const rapidjson::Value &obj = doc["max_tcp_num"];
+ // max proxy num
+ if (doc.HasMember("max_proxy_num") && doc["max_proxy_num"].IsInt() &&
+ doc["max_proxy_num"].GetInt() > 0) {
+ const rapidjson::Value &obj = doc["max_proxy_num"];
max_proxy_num_ = obj.GetInt();
} else {
- max_proxy_num_ = constants::kMaxBusNum;
+ max_proxy_num_ = constants::kMaxProxyNum;
}
if (doc.HasMember("inlong_group_ids") && doc["inlong_group_ids"].IsString())
{
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
index 0a7b8d944c..785f648226 100644
--- 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
@@ -42,13 +42,13 @@ int32_t ApiImp::InitApi(const char *config_file_path) {
return DoInit();
}
-int32_t ApiImp::Send(const char *business_id, const char *table_id,
+int32_t ApiImp::Send(const char *inlong_group_id, const char *inlong_stream_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) {
+ if (inlong_group_id == nullptr || inlong_stream_id == nullptr ||
+ msg == nullptr || msg_len <= 0) {
return SdkCode::kInvalidInput;
}
@@ -57,7 +57,7 @@ int32_t ApiImp::Send(const char *business_id, const char
*table_id,
}
int64_t msg_time = Utils::getCurrentMsTime();
- return this->SendBase(business_id, table_id, local_ip_, msg_time,
+ return this->SendBase(inlong_group_id, inlong_stream_id, local_ip_, msg_time,
{msg, msg_len}, call_back);
}
@@ -94,16 +94,17 @@ int32_t ApiImp::CloseApi(int32_t max_waitms) {
}
int32_t ApiImp::DoInit() {
- LOG_INFO("tdbus sdk cpp start Init, version:" << constants::kVersion);
+ LOG_INFO(
+ "inlong dataproxy sdk cpp start Init, version:" << constants::kVersion);
signal(SIGPIPE, SIG_IGN);
- LOG_INFO("tdbus_sdk_cpp Init complete!");
+ LOG_INFO("inlong dataproxy cpp sdk Init complete!");
ProxyManager::GetInstance()->Init();
for (int i = 0; i < SdkConfig::getInstance()->inlong_group_ids_.size(); i++)
{
- LOG_INFO("DoInit CheckBidConf inlong_group_id:"
+ LOG_INFO("DoInit CheckConf inlong_group_id:"
<< SdkConfig::getInstance()->inlong_group_ids_[i]);
ProxyManager::GetInstance()->CheckBidConf(
SdkConfig::getInstance()->inlong_group_ids_[i], false);
@@ -151,7 +152,8 @@ int32_t ApiImp::InitManager() {
init_succeed_ = true;
return SdkCode::kSuccess;
}
-int32_t ApiImp::AddBid(const std::vector<std::string> &inlong_group_ids) {
+int32_t
+ApiImp::AddInLongGroupId(const std::vector<std::string> &inlong_group_ids) {
if (inited_ == false) {
return SdkCode::kSendBeforeInit;
}
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
index 80372e7c8d..0b19ad4d98 100644
--- 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
@@ -26,9 +26,9 @@
#include <functional>
namespace inlong {
-using UserCallBack =
- std::function<int32_t(const char *, const char *, const char *, int32_t,
- const int64_t, const char *)>;
+
+typedef int (*UserCallBack)(const char *, const char *, const char *, int32_t,
+ const int64_t, const char *);
class ApiImp {
public:
@@ -36,12 +36,13 @@ public:
~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 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);
- int32_t AddBid(const std::vector<std::string> &bids);
+ int32_t AddInLongGroupId(const std::vector<std::string> &inlong_group_ids);
private:
int32_t DoInit();
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
index cef90fdb46..14a5f91207 100644
--- 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
@@ -26,16 +26,17 @@ 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::Send(const char *inlong_group_id,
+ const char *inlong_stream_id, const char *msg,
+ int32_t msg_len, UserCallBack call_back) {
+ return api_impl_->Send(inlong_group_id, inlong_stream_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);
+ return api_impl_->AddInLongGroupId(groupids);
}
} // namespace inlong
\ No newline at end of file
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/recv_group.cc
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/recv_group.cc
index 4e58fe9be4..94b0f8e848 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/recv_group.cc
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/recv_group.cc
@@ -17,18 +17,20 @@
#include "recv_group.h"
+#include "../protocol/msg_protocol.h"
#include "../utils/utils.h"
#include "api_code.h"
-#include "../protocol/msg_protocol.h"
#include <cstdlib>
#include <functional>
namespace inlong {
const uint32_t ATTR_LENGTH = 10;
const uint32_t DEFAULT_PACK_ATTR = 400;
-RecvGroup::RecvGroup(const std::string &inlong_group_id, const std::string
&inlong_stream_id,
+RecvGroup::RecvGroup(const std::string &inlong_group_id,
+ const std::string &inlong_stream_id,
std::shared_ptr<SendManager> send_manager)
- : cur_len_(0), inlong_group_id_(inlong_group_id),
inlong_stream_id_(inlong_stream_id), groupId_num_(0), streamId_num_(0),
+ : cur_len_(0), inlong_group_id_(inlong_group_id),
+ inlong_stream_id_(inlong_stream_id), groupId_num_(0), streamId_num_(0),
msg_type_(SdkConfig::getInstance()->msg_type_),
data_capacity_(SdkConfig::getInstance()->buf_size_),
send_manager_(send_manager) {
@@ -38,7 +40,8 @@ RecvGroup::RecvGroup(const std::string &inlong_group_id,
const std::string &inlo
pack_buf_ = new char[data_capacity_];
memset(pack_buf_, 0x0, data_capacity_);
- topic_desc_ = "groupId=" + inlong_group_id_ + "&streamId=" +
inlong_stream_id_;
+ topic_desc_ =
+ "groupId=" + inlong_group_id_ + "&streamId=" + inlong_stream_id_;
data_time_ = 0;
last_pack_time_ = Utils::getCurrentMsTime();
max_recv_size_ = SdkConfig::getInstance()->recv_buf_size_;
@@ -138,8 +141,8 @@ void RecvGroup::AddMsg(const std::string &msg, std::string
client_ip,
"__addcol1__reptime=" + Utils::getFormatTime(data_time_) +
"&__addcol2__ip=" + client_ip;
msgs_.push(std::make_shared<SdkMsg>(msg, client_ip, data_time_, call_back,
- data_pack_format_attr, user_client_ip,
- user_report_time));
+ data_pack_format_attr, user_client_ip,
+ user_report_time));
cur_len_ += msg.size() + ATTR_LENGTH;
}
@@ -189,8 +192,7 @@ bool RecvGroup::PackMsg(std::vector<SdkMsgPtr> &msgs, char
*pack_data,
}
if (msg_type_ >= constants::kBinPackMethod) {
- char *bodyBegin = pack_data + sizeof(BinaryMsgHead) +
- sizeof(uint32_t);
+ char *bodyBegin = pack_data + sizeof(BinaryMsgHead) + sizeof(uint32_t);
uint32_t body_len = 0;
std::string snappy_res;
@@ -207,8 +209,7 @@ bool RecvGroup::PackMsg(std::vector<SdkMsgPtr> &msgs, char
*pack_data,
memcpy(bodyBegin, pack_buf_, body_len);
real_msg_type = msg_type_;
}
- *(uint32_t *)(&(pack_data[sizeof(BinaryMsgHead)])) =
- htonl(body_len);
+ *(uint32_t *)(&(pack_data[sizeof(BinaryMsgHead)])) = htonl(body_len);
bodyBegin += body_len;
@@ -235,7 +236,8 @@ bool RecvGroup::PackMsg(std::vector<SdkMsgPtr> &msgs, char
*pack_data,
attr = "node1ip=" + SdkConfig::getInstance()->local_ip_ +
"&rtime1=" + std::to_string(Utils::getCurrentMsTime());
else
- attr = groupId_streamId_char + "&node1ip=" +
SdkConfig::getInstance()->local_ip_ +
+ attr = groupId_streamId_char +
+ "&node1ip=" + SdkConfig::getInstance()->local_ip_ +
"&rtime1=" + std::to_string(Utils::getCurrentMsTime());
} else {
attr = topic_desc_;
@@ -281,12 +283,9 @@ bool RecvGroup::PackMsg(std::vector<SdkMsgPtr> &msgs, char
*pack_data,
memcpy(bodyBegin, snappy_res.data(), body_len);
} else {
body_len = idx;
- memcpy(
- bodyBegin, pack_buf_,
- body_len);
+ memcpy(bodyBegin, pack_buf_, body_len);
}
- *(uint32_t *)(&(pack_data[sizeof(ProtocolMsgHead)])) =
- htonl(body_len);
+ *(uint32_t *)(&(pack_data[sizeof(ProtocolMsgHead)])) = htonl(body_len);
bodyBegin += body_len;
// attr
@@ -369,8 +368,9 @@ RecvGroup::BuildSendBuf(std::vector<SdkMsgPtr> &msgs) {
void RecvGroup::CallbalkToUsr(std::vector<SdkMsgPtr> &msgs) {
for (auto &it : msgs) {
if (it->cb_) {
- it->cb_(inlong_group_id_.data(), inlong_stream_id_.data(),
it->msg_.data(), it->msg_.size(),
- it->user_report_time_, it->user_client_ip_.data());
+ it->cb_(inlong_group_id_.data(), inlong_stream_id_.data(),
+ it->msg_.data(), it->msg_.size(), it->user_report_time_,
+ it->user_client_ip_.data());
}
}
}
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/recv_group.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/recv_group.h
index 3624e26c60..f9150f3602 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/recv_group.h
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/recv_group.h
@@ -29,7 +29,6 @@
#include "../utils/noncopyable.h"
#include "sdk_conf.h"
-
namespace inlong {
class RecvGroup {
private:
@@ -67,11 +66,9 @@ public:
int32_t SendData(const std::string &msg, const std::string &groupId,
const std::string &streamId, const std::string &client_ip,
uint64_t report_time, UserCallBack call_back);
- bool PackMsg(std::vector<SdkMsgPtr> &msgs, char *pack_data,
- uint32_t &out_len, uint32_t uniq_id);
- void DispatchMsg(
- bool
- exit);
+ bool PackMsg(std::vector<SdkMsgPtr> &msgs, char *pack_data, uint32_t
&out_len,
+ uint32_t uniq_id);
+ void DispatchMsg(bool exit);
char *data() const { return pack_buf_; }
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/send_group.cc
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/send_group.cc
index 57f1c574b4..b4d001caab 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/send_group.cc
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/send_group.cc
@@ -121,7 +121,8 @@ void SendGroup::UpdateConf(std::error_code error) {
ClearOldTcpClients();
ProxyInfoVec new_proxy_info;
- if (ProxyManager::GetInstance()->GetProxy(group_id_, new_proxy_info) !=
kSuccess ||
+ if (ProxyManager::GetInstance()->GetProxy(group_id_, new_proxy_info) !=
+ kSuccess ||
new_proxy_info.empty()) {
update_conf_timer_->expires_after(std::chrono::milliseconds(kTimerMinute));
update_conf_timer_->async_wait(
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/send_group.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/send_group.h
index b1659e3631..88dc90771c 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/send_group.h
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/group/send_group.h
@@ -20,9 +20,9 @@
#ifndef INLONG_SDK_SEND_GROUP_H
#define INLONG_SDK_SEND_GROUP_H
+#include "../client/tcp_client.h"
#include "../config/proxy_info.h"
#include "../utils/send_buffer.h"
-#include "../client/tcp_client.h"
#include <queue>
namespace inlong {
const int kTimerMiSeconds = 10;
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/proxy_manager.cc
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/proxy_manager.cc
index 142cbe5fbc..ea0b3f2ef5 100644
---
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/proxy_manager.cc
+++
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/proxy_manager.cc
@@ -68,7 +68,7 @@ void ProxyManager::DoUpdate() {
LOG_INFO("start ProxyManager DoUpdate.");
if (groupid_2_cluster_map_.empty()) {
- LOG_INFO("empty groupid, no need to DoUpdate buslist");
+ LOG_INFO("empty groupid, no need to DoUpdate proxy list");
update_mutex_.unlock();
return;
}
@@ -216,16 +216,17 @@ int32_t ProxyManager::GetProxy(const std::string &groupid,
auto it = groupid_2_proxy_map_.find(groupid);
if (it == groupid_2_proxy_map_.end()) {
LOG_ERROR("GetProxyByGroupid failed . Groupid " << groupid);
- return SdkCode::kFailGetBusConf;
+ return SdkCode::kFailGetProxyConf;
}
proxy_info_vec = it->second;
return SdkCode::kSuccess;
}
-int32_t ProxyManager::CheckBidConf(const std::string &groupid, bool is_inited)
{
+int32_t ProxyManager::CheckBidConf(const std::string &inlong_group_id,
+ bool is_inited) {
{
unique_read_lock<read_write_mutex> rdlck(groupid_2_cluster_rwmutex_);
- auto it = groupid_2_cluster_map_.find(groupid);
+ auto it = groupid_2_cluster_map_.find(inlong_group_id);
if (it != groupid_2_cluster_map_.end()) {
return SdkCode::kSuccess;
}
@@ -233,10 +234,11 @@ int32_t ProxyManager::CheckBidConf(const std::string
&groupid, bool is_inited) {
{
unique_write_lock<read_write_mutex> wtlck(groupid_2_cluster_rwmutex_);
- groupid_2_cluster_map_.emplace(groupid, -1);
+ groupid_2_cluster_map_.emplace(inlong_group_id, -1);
}
- LOG_INFO("CheckProxyConf groupid:" << groupid << ",isInited :" << is_inited);
+ LOG_INFO("CheckProxyConf groupid:" << inlong_group_id
+ << ",isInited :" << is_inited);
if (is_inited) {
std::unique_lock<std::mutex> con_lck(cond_mutex_);
update_flag_ = true;
@@ -248,9 +250,9 @@ int32_t ProxyManager::CheckBidConf(const std::string
&groupid, bool is_inited) {
return SdkCode::kSuccess;
}
-bool ProxyManager::IsBusExist(const std::string &groupid) {
+bool ProxyManager::IsExist(const std::string &inlong_group_id) {
unique_read_lock<read_write_mutex> rdlck(groupid_2_proxy_map_rwmutex_);
- auto it = groupid_2_proxy_map_.find(groupid);
+ auto it = groupid_2_proxy_map_.find(inlong_group_id);
if (it == groupid_2_proxy_map_.end()) {
return false;
}
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/proxy_manager.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/proxy_manager.h
index 25d78beb13..b3f46bc68d 100644
---
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/proxy_manager.h
+++
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/proxy_manager.h
@@ -46,18 +46,18 @@ private:
volatile bool inited_ = false;
int32_t ParseAndGet(const std::string &groupid, const std::string &meta_data,
- ProxyInfoVec &bus_info_vec);
+ ProxyInfoVec &proxy_info_vec);
public:
ProxyManager(){};
~ProxyManager();
static ProxyManager *GetInstance() { return instance_; }
- int32_t CheckBidConf(const std::string &groupid, bool is_inited);
+ int32_t CheckBidConf(const std::string &inlong_group_id, bool is_inited);
void Update();
void DoUpdate();
void Init();
int32_t GetProxy(const std::string &groupid, ProxyInfoVec &proxy_info_vec);
- bool IsBusExist(const std::string &groupid);
+ bool IsExist(const std::string &inlong_group_id);
};
} // namespace inlong
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/send_manager.cc
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/send_manager.cc
index f883bf44d7..24b74c2bf2 100644
---
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/send_manager.cc
+++
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/send_manager.cc
@@ -40,12 +40,12 @@ SendGroupPtr SendManager::GetSendGroup(const std::string
&group_id) {
return send_group_ptr;
}
-bool SendManager::AddSendGroup(const std::string &group_id) {
- if (!ProxyManager::GetInstance()->IsBusExist(group_id)) {
- LOG_ERROR("bus is not exist." << group_id);
+bool SendManager::AddSendGroup(const std::string &inlong_group_id) {
+ if (!ProxyManager::GetInstance()->IsExist(inlong_group_id)) {
+ LOG_ERROR("inlong_group_id is not exist." << inlong_group_id);
return false;
}
- DoAddSendGroup(group_id);
+ DoAddSendGroup(inlong_group_id);
return false;
}
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/send_manager.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/send_manager.h
index d6c42044f4..6ee55d4fb4 100644
---
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/send_manager.h
+++
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/manager/send_manager.h
@@ -40,7 +40,7 @@ public:
SendManager();
virtual ~SendManager(){};
SendGroupPtr GetSendGroup(const std::string &group_id);
- bool AddSendGroup(const std::string &group_id);
+ bool AddSendGroup(const std::string &inlong_group_id);
};
} // namespace inlong
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/protocol/msg_protocol.cc
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/protocol/msg_protocol.cc
index 8d561a78ec..28bfed473c 100644
---
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/protocol/msg_protocol.cc
+++
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/protocol/msg_protocol.cc
@@ -1,21 +1,21 @@
/**
-* 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.
-*/
+ * 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 "msg_protocol.h"
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/protocol/msg_protocol.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/protocol/msg_protocol.h
index ab1cfcb806..a17fc84f72 100644
---
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/protocol/msg_protocol.h
+++
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/protocol/msg_protocol.h
@@ -1,21 +1,21 @@
/**
-* 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.
-*/
+ * 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_MSG_PROTOCOL_H
#define INLONG_SDK_MSG_PROTOCOL_H
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/capi_constant.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/capi_constant.h
index 89f10e5656..089bd0bca4 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/capi_constant.h
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/capi_constant.h
@@ -57,12 +57,13 @@ static const char kLogPath[] = "./";
static const char kManagerURL[] =
"http://127.0.0.1:8099/inlong/manager/openapi/dataproxy/getIpList";;
-static const bool kEnableBusURLFromCluster = false;
-static const char kBusClusterURL[] =
- "127.0.0.1/heartbeat/tdbus_ip_v2?cluster_id=0&net_tag=all";
-static const uint32_t kBusUpdateInterval = 2;
-static const uint32_t kBusURLTimeout = 5;
-static const uint32_t kMaxBusNum = 200;
+static const bool kEnableManagerFromCluster = false;
+static const char kManagerClusterURL[] =
+ "http://127.0.0.1:8099/heartbeat/";
+ "dataproxy_ip_v2?cluster_id=0&net_tag=normal";
+static const uint32_t kManagerUpdateInterval = 2;
+static const uint32_t kManagerTimeout = 5;
+static const uint32_t kMaxProxyNum = 200;
static const bool kEnableTCPNagle = true;
static const uint32_t kTcpIdleTime = 600000;
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/read_write_mutex.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/read_write_mutex.h
index ef73450d49..4ddabfbbce 100644
---
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/read_write_mutex.h
+++
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/read_write_mutex.h
@@ -128,6 +128,6 @@ private:
_ReadWriteLock *m_ptr_rw_lock = nullptr;
};
-} // namespace dataproxy_sdk
+} // namespace inlong
#endif // INLONG_SDK_READ_WRITE_MUTEX_H
\ No newline at end of file
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/send_buffer.h
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/send_buffer.h
index 16bde14941..66e93d67ce 100644
--- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/send_buffer.h
+++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/src/utils/send_buffer.h
@@ -54,8 +54,8 @@ private:
public:
SendBuffer(uint32_t size)
: uniq_id_(0), is_used_(false), is_packed_(false), size_(size),
- msg_cnt_(0), len_(0), inlong_group_id_(), inlong_stream_id_(),
first_send_time_(0),
- latest_send_time_(0) {
+ msg_cnt_(0), len_(0), inlong_group_id_(), inlong_stream_id_(),
+ first_send_time_(0), latest_send_time_(0) {
content_ = new char[size];
if (content_) {
memset(content_, 0x0, size);
@@ -74,8 +74,12 @@ public:
void setLen(const uint32_t len) { len_ = len; }
std::string bid() { return inlong_group_id_; }
std::string tid() { return inlong_stream_id_; }
- void setInlongGroupId(const std::string &inlong_group_id) { inlong_group_id_
= inlong_group_id; }
- void setStreamId(const std::string &inlong_stream_id) { inlong_stream_id_ =
inlong_stream_id; }
+ void setInlongGroupId(const std::string &inlong_group_id) {
+ inlong_group_id_ = inlong_group_id;
+ }
+ void setStreamId(const std::string &inlong_stream_id) {
+ inlong_stream_id_ = inlong_stream_id;
+ }
void setUniqId(const uint32_t &uniq_id) { uniq_id_ = uniq_id; }
@@ -83,8 +87,9 @@ public:
void doUserCallBack() {
for (auto it : user_msg_vector_) {
if (it->cb_) {
- it->cb_(inlong_group_id_.data(), inlong_stream_id_.data(),
it->msg_.data(), it->msg_.size(),
- it->user_report_time_, it->user_client_ip_.data());
+ it->cb_(inlong_group_id_.data(), inlong_stream_id_.data(),
+ it->msg_.data(), it->msg_.size(), it->user_report_time_,
+ it->user_client_ip_.data());
}
}
}
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 c1d731f4dc..153dd588df 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
@@ -1,88 +1,88 @@
/**
-* 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.
-*/
+ * 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_UTILS_H
#define INLONG_SDK_UTILS_H
+#include "atomic.h"
#include <snappy.h>
#include <stdint.h>
+#include <string>
#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
-#include <string>
#include <utility>
#include <vector>
-#include "atomic.h"
namespace inlong {
extern AtomicUInt g_send_msgid;
extern AtomicInt user_exit_flag;
-struct HttpRequest
-{
+struct HttpRequest {
std::string url;
uint32_t timeout;
bool need_auth;
std::string auth_id;
std::string auth_key;
std::string post_data;
-
};
class Utils {
- private:
+private:
static char snowflake_id[35];
static uint16_t sequence;
static uint64_t last_msstamp;
- public:
+public:
static void taskWaitTime(int32_t sec);
static uint64_t getCurrentMsTime();
static uint64_t getCurrentWsTime();
static std::string getFormatTime(uint64_t date_time);
- static size_t zipData(const char* input, uint32_t input_len, std::string&
zip_res);
- static char* getSnowflakeId();
- static bool getFirstIpAddr(std::string& local_host);
- inline static bool isLegalTime(uint64_t report_time)
- {
+ static size_t zipData(const char *input, uint32_t input_len,
+ std::string &zip_res);
+ static char *getSnowflakeId();
+ static bool getFirstIpAddr(std::string &local_host);
+ inline static bool isLegalTime(uint64_t report_time) {
return ((report_time > 1435101567000LL) && (report_time <
4103101567000LL));
}
static bool bindCPU(int32_t cpu_id);
- static int32_t requestUrl(const std::string& url, std::string& urlByDNS,
std::string& res,
- uint32_t timeout);
- static int32_t requestUrl(std::string& res, const HttpRequest* request);
- static bool readFile(const std::string& file_path,
- std::string& content);
+ static int32_t requestUrl(const std::string &url, std::string &urlByDNS,
+ std::string &res, uint32_t timeout);
+ static int32_t requestUrl(std::string &res, const HttpRequest *request);
+ static bool readFile(const std::string &file_path, std::string &content);
- static int32_t splitOperate(const std::string& source,
std::vector<std::string>& result,
- const std::string& delimiter);
- static std::string getVectorStr(std::vector<std::string>& vs);
+ static int32_t splitOperate(const std::string &source,
+ std::vector<std::string> &result,
+ const std::string &delimiter);
+ static std::string getVectorStr(std::vector<std::string> &vs);
- static std::string GenBasicAuthCredential(const std::string& id, const
std::string& key);
- static std::string Base64Encode(const std::string& data);
+ static std::string GenBasicAuthCredential(const std::string &id,
+ const std::string &key);
+ static std::string Base64Encode(const std::string &data);
private:
- static size_t getUrlResponse(void* buffer, size_t size, size_t count, void*
response);
+ static size_t getUrlResponse(void *buffer, size_t size, size_t count,
+ void *response);
static int64_t waitNextMills(int64_t last_ms);
- static std::string trim(const std::string& source);
- static bool parseHost(const std::string& host, std::string& ip);
- static bool getUrlByDNS(const std::string& url, std::string& ipUrl);
+ static std::string trim(const std::string &source);
+ static bool parseHost(const std::string &host, std::string &ip);
+ static bool getUrlByDNS(const std::string &url, std::string &ipUrl);
};
-} // namespace inlong
+} // namespace inlong
-#endif // INLONG_SDK_UTILS_H
\ No newline at end of file
+#endif // INLONG_SDK_UTILS_H
\ No newline at end of file
diff --git
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/third_party/CMakeLists.txt
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/third_party/CMakeLists.txt
index eadbe8f97f..9c0355bd7f 100644
---
a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/third_party/CMakeLists.txt
+++
b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-cpp/third_party/CMakeLists.txt
@@ -26,54 +26,54 @@ set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-error")
include(ExternalProject)
ExternalProject_Add(
- snappy_proj
- URL https://github.com/google/snappy/archive/1.1.8.tar.gz
- CMAKE_ARGS
- -DCMAKE_INSTALL_PREFIX=${PROJECT_SOURCE_DIR}
- -DSNAPPY_BUILD_TESTS=OFF
- -DHAVE_LIBLZO2=OFF
- -DCMAKE_CXX_FLAGS=-fPIC
- TEST_BEFORE_INSTALL 0
- BUILD_IN_SOURCE 1
+ snappy_proj
+ URL https://github.com/google/snappy/archive/1.1.8.tar.gz
+ CMAKE_ARGS
+ -DCMAKE_INSTALL_PREFIX=${PROJECT_SOURCE_DIR}
+ -DSNAPPY_BUILD_TESTS=OFF
+ -DHAVE_LIBLZO2=OFF
+ -DCMAKE_CXX_FLAGS=-fPIC
+ TEST_BEFORE_INSTALL 0
+ BUILD_IN_SOURCE 1
)
ExternalProject_Add(
- curl_proj
- URL
https://github.com/curl/curl/releases/download/curl-7_78_0/curl-7.78.0.tar.gz
- CONFIGURE_COMMAND ./configure --prefix=${PROJECT_SOURCE_DIR}
--without-zlib --without-ssl --disable-shared --disable-ldap && make install
- TEST_BEFORE_INSTALL 0
- BUILD_IN_SOURCE 1
+ curl_proj
+ URL
https://github.com/curl/curl/releases/download/curl-7_78_0/curl-7.78.0.tar.gz
+ CONFIGURE_COMMAND ./configure --prefix=${PROJECT_SOURCE_DIR}
--without-zlib --without-ssl --disable-shared --disable-ldap && make install
+ TEST_BEFORE_INSTALL 0
+ BUILD_IN_SOURCE 1
)
ExternalProject_Add(
- rapidjson
- PREFIX "rapidjson"
- URL "https://github.com/Tencent/rapidjson/archive/v1.1.0.tar.gz";
- INSTALL_DIR ${PROJECT_SOURCE_DIR}
- CMAKE_ARGS
- -DRAPIDJSON_BUILD_TESTS=OFF
- -DRAPIDJSON_BUILD_DOC=OFF
- -DRAPIDJSON_BUILD_EXAMPLES=OFF
- CONFIGURE_COMMAND ""
- BUILD_COMMAND ""
- INSTALL_COMMAND ""
- UPDATE_COMMAND ""
+ rapidjson
+ PREFIX "rapidjson"
+ URL "https://github.com/Tencent/rapidjson/archive/v1.1.0.tar.gz";
+ INSTALL_DIR ${PROJECT_SOURCE_DIR}
+ CMAKE_ARGS
+ -DRAPIDJSON_BUILD_TESTS=OFF
+ -DRAPIDJSON_BUILD_DOC=OFF
+ -DRAPIDJSON_BUILD_EXAMPLES=OFF
+ CONFIGURE_COMMAND ""
+ BUILD_COMMAND ""
+ INSTALL_COMMAND ""
+ UPDATE_COMMAND ""
)
ExternalProject_Add(
- asio
- SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/asio
- BINARY_DIR ${CMAKE_CURRENT_SOURCE_DIR}/asio/asio/
- URL https://github.com/chriskohlhoff/asio/archive/asio-1-28-0.tar.gz
- CONFIGURE_COMMAND ./autogen.sh
- COMMAND ./configure --prefix=${PROJECT_SOURCE_DIR}/ CFLAGS=-std=c++11
CPPFLAGS=-std=c++11 CXXFLAGS=-std=c++11 --with-openssl=${PROJECT_SOURCE_DIR}
+ asio
+ SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/asio
+ BINARY_DIR ${CMAKE_CURRENT_SOURCE_DIR}/asio/asio/
+ URL https://github.com/chriskohlhoff/asio/archive/asio-1-28-0.tar.gz
+ CONFIGURE_COMMAND ./autogen.sh
+ COMMAND ./configure --prefix=${PROJECT_SOURCE_DIR}/ CFLAGS=-std=c++11
CPPFLAGS=-std=c++11 CXXFLAGS=-std=c++11 --with-openssl=${PROJECT_SOURCE_DIR}
)
ExternalProject_Add(
- openssl
- URL https://www.openssl.org/source/openssl-1.1.1q.tar.gz
- CONFIGURE_COMMAND ./config --prefix=${PROJECT_SOURCE_DIR}
- BUILD_COMMAND make
- INSTALL_COMMAND make install
- BUILD_IN_SOURCE 1
+ openssl
+ URL https://www.openssl.org/source/openssl-1.1.1q.tar.gz
+ CONFIGURE_COMMAND ./config --prefix=${PROJECT_SOURCE_DIR}
+ BUILD_COMMAND make
+ INSTALL_COMMAND make install
+ BUILD_IN_SOURCE 1
)
\ No newline at end of file
