diff --git a/BUILD b/BUILD index ad3919a8..12015972 100644 --- a/BUILD +++ b/BUILD @@ -194,30 +194,6 @@ cc_library( visibility = ["//visibility:public"], ) -cc_library( - name = 'kafka_message', - hdrs = [ - 'src/factory/KafkaTaskImpl.inl', - 'src/protocol/KafkaDataTypes.h', - 'src/protocol/KafkaMessage.h', - 'src/protocol/KafkaResult.h', - 'src/protocol/kafka_parser.h', - ], - includes = [ - 'src/factory', - 'src/protocol', - ], - srcs = [ - 'src/factory/KafkaTaskImpl.cc', - 'src/protocol/KafkaMessage.cc', - 'src/protocol/KafkaResult.cc', - ], - copts = ['-fno-rtti'], - deps = [ - ':common', - ], -) - cc_library( name = 'kafka', hdrs = [ @@ -237,10 +213,12 @@ cc_library( 'src/client/WFKafkaClient.cc', 'src/protocol/KafkaDataTypes.cc', 'src/protocol/kafka_parser.c', + 'src/factory/KafkaTaskImpl.cc', + 'src/protocol/KafkaMessage.cc', + 'src/protocol/KafkaResult.cc', ], deps = [ ':common', - ':kafka_message', ], visibility = ["//visibility:public"], linkopts = [ @@ -274,37 +252,37 @@ cc_library( ) cc_binary( - name = 'helloworld', - srcs = ['tutorial/tutorial-00-helloworld.cc'], - deps = [':http'], + name = 'helloworld', + srcs = ['tutorial/tutorial-00-helloworld.cc'], + deps = [':http'], ) cc_binary( - name = 'wget', - srcs = ['tutorial/tutorial-01-wget.cc'], - deps = [':http'], + name = 'wget', + srcs = ['tutorial/tutorial-01-wget.cc'], + deps = [':http'], ) cc_binary( - name = 'redis_cli', - srcs = ['tutorial/tutorial-02-redis_cli.cc'], - deps = [':redis'], + name = 'redis_cli', + srcs = ['tutorial/tutorial-02-redis_cli.cc'], + deps = [':redis'], ) cc_binary( - name = 'wget_to_redis', - srcs = ['tutorial/tutorial-03-wget_to_redis.cc'], - deps = [':http', 'redis'], + name = 'wget_to_redis', + srcs = ['tutorial/tutorial-03-wget_to_redis.cc'], + deps = [':http', 'redis'], ) cc_binary( - name = 'http_echo_server', - srcs = ['tutorial/tutorial-04-http_echo_server.cc'], - deps = [':http'], + name = 'http_echo_server', + srcs = ['tutorial/tutorial-04-http_echo_server.cc'], + deps = [':http'], ) cc_binary( - name = 'http_proxy', - srcs = ['tutorial/tutorial-05-http_proxy.cc'], - deps = [':http'], + name = 'http_proxy', + srcs = ['tutorial/tutorial-05-http_proxy.cc'], + deps = [':http'], ) cc_binary( @@ -314,21 +292,21 @@ cc_binary( ) cc_binary( - name = 'sort_task', - srcs = ['tutorial/tutorial-07-sort_task.cc'], - deps = [':common'], + name = 'sort_task', + srcs = ['tutorial/tutorial-07-sort_task.cc'], + deps = [':common'], ) cc_binary( - name = 'matrix_multiply', - srcs = ['tutorial/tutorial-08-matrix_multiply.cc'], - deps = [':common'], + name = 'matrix_multiply', + srcs = ['tutorial/tutorial-08-matrix_multiply.cc'], + deps = [':common'], ) cc_binary( - name = 'http_file_server', - srcs = ['tutorial/tutorial-09-http_file_server.cc'], - deps = [':http'], + name = 'http_file_server', + srcs = ['tutorial/tutorial-09-http_file_server.cc'], + deps = [':http'], ) cc_library( @@ -338,45 +316,45 @@ cc_library( ) cc_binary( - name = 'server', - srcs = [ - 'tutorial/tutorial-10-user_defined_protocol/server.cc', - 'tutorial/tutorial-10-user_defined_protocol/message.cc', - ], - deps = [':common', ':user_hdrs'], + name = 'server', + srcs = [ + 'tutorial/tutorial-10-user_defined_protocol/server.cc', + 'tutorial/tutorial-10-user_defined_protocol/message.cc', + ], + deps = [':common', ':user_hdrs'], ) cc_binary( - name = 'client', - srcs = [ - 'tutorial/tutorial-10-user_defined_protocol/client.cc', - 'tutorial/tutorial-10-user_defined_protocol/message.cc', - ], - deps = [':common', ':user_hdrs'], + name = 'client', + srcs = [ + 'tutorial/tutorial-10-user_defined_protocol/client.cc', + 'tutorial/tutorial-10-user_defined_protocol/message.cc', + ], + deps = [':common', ':user_hdrs'], ) cc_binary( - name = 'graph_task', - srcs = ['tutorial/tutorial-11-graph_task.cc'], - deps = [':http'], + name = 'graph_task', + srcs = ['tutorial/tutorial-11-graph_task.cc'], + deps = [':http'], ) cc_binary( - name = 'mysql_cli', - srcs = ['tutorial/tutorial-12-mysql_cli.cc'], - deps = [':mysql'], + name = 'mysql_cli', + srcs = ['tutorial/tutorial-12-mysql_cli.cc'], + deps = [':mysql'], ) cc_binary( - name = 'kafka_cli', - srcs = ['tutorial/tutorial-13-kafka_cli.cc'], - deps = [':kafka', ':workflow_hdrs'], + name = 'kafka_cli', + srcs = ['tutorial/tutorial-13-kafka_cli.cc'], + deps = [':kafka', ':workflow_hdrs'], ) cc_binary( - name = 'consul_cli', - srcs = ['tutorial/tutorial-14-consul_cli.cc'], - deps = [':consul'], + name = 'consul_cli', + srcs = ['tutorial/tutorial-14-consul_cli.cc'], + deps = [':consul'], ) cc_binary( @@ -407,9 +385,9 @@ cc_binary( ) cc_binary( - name = 'redis_subscriber', - srcs = ['tutorial/tutorial-18-redis_subscriber.cc'], - deps = [':redis'], + name = 'redis_subscriber', + srcs = ['tutorial/tutorial-18-redis_subscriber.cc'], + deps = [':redis'], ) cc_binary( diff --git a/src/factory/CMakeLists.txt b/src/factory/CMakeLists.txt index c8efcaf7..6a247ba8 100644 --- a/src/factory/CMakeLists.txt +++ b/src/factory/CMakeLists.txt @@ -33,5 +33,4 @@ if (KAFKA STREQUAL "y") KafkaTaskImpl.cc ) add_library("factory_kafka" OBJECT ${SRC}) - set_property(SOURCE KafkaTaskImpl.cc APPEND PROPERTY COMPILE_OPTIONS "-fno-rtti") endif () diff --git a/src/factory/xmake.lua b/src/factory/xmake.lua index e1e76b2c..9e7aa559 100644 --- a/src/factory/xmake.lua +++ b/src/factory/xmake.lua @@ -13,7 +13,6 @@ target("kafka_factory") if has_config("kafka") then add_files("KafkaTaskImpl.cc") set_kind("object") - add_cxxflags("-fno-rtti") add_deps("factory") add_packages("zlib", "snappy", "zstd", "lz4") else diff --git a/src/include/workflow/dns_types.h b/src/include/workflow/dns_types.h new file mode 120000 index 00000000..666230ef --- /dev/null +++ b/src/include/workflow/dns_types.h @@ -0,0 +1 @@ +../../protocol/dns_types.h \ No newline at end of file diff --git a/src/kernel/Communicator.cc b/src/kernel/Communicator.cc index b32f22be..665e9333 100644 --- a/src/kernel/Communicator.cc +++ b/src/kernel/Communicator.cc @@ -97,12 +97,96 @@ static int __bind_sockaddr(int sockfd, const struct sockaddr *addr, return 0; } +<<<<<<< HEAD +======= +static int __create_ssl(SSL_CTX *ssl_ctx, struct CommConnEntry *entry) +{ + BIO *bio = BIO_new_socket(entry->sockfd, BIO_NOCLOSE); + + if (bio) + { + entry->ssl = SSL_new(ssl_ctx); + if (entry->ssl) + { + SSL_set_bio(entry->ssl, bio, bio); + return 0; + } + + BIO_free(bio); + } + + return -1; +} + +#define SSL_WRITEV_BUFSIZE 2048 + +static int __ssl_writev(SSL *ssl, struct iovec vectors[], int cnt) +{ + if (vectors[0].iov_len < SSL_WRITEV_BUFSIZE && cnt > 1) + { + char *p = (char *)SSL_get_app_data(ssl); + size_t nleft = SSL_WRITEV_BUFSIZE; + size_t n; + int i; + + if (!p) + { + p = (char *)malloc(SSL_WRITEV_BUFSIZE); + if (!p) + return -1; + + if (SSL_set_app_data(ssl, p) <= 0) + { + free(p); + return -1; + } + } + + n = vectors[0].iov_len; + memcpy(p, vectors[0].iov_base, n); + vectors[0].iov_base = p; + + p += n; + nleft -= n; + for (i = 1; i < cnt; i++) + { + if (vectors[i].iov_len < nleft) + n = vectors[i].iov_len; + else + n = nleft; + + memcpy(p, vectors[i].iov_base, n); + vectors[i].iov_base = (char *)vectors[i].iov_base + n; + vectors[i].iov_len -= n; + + p += n; + nleft -= n; + if (nleft == 0) + break; + } + + vectors[0].iov_len = SSL_WRITEV_BUFSIZE - nleft; + } + + return SSL_write(ssl, vectors[0].iov_base, vectors[0].iov_len); +} + +>>>>>>> f20375dbc48c543ec675560e134c375118d0438b static void __release_conn(struct CommConnEntry *entry) { delete entry->conn; if (!entry->service) pthread_mutex_destroy(&entry->mutex); +<<<<<<< HEAD +======= + if (entry->ssl) + { + free(SSL_get_app_data(entry->ssl)); + SSL_free(entry->ssl); + } + +>>>>>>> f20375dbc48c543ec675560e134c375118d0438b close(entry->sockfd); free(entry); } @@ -386,20 +470,39 @@ int Communicator::send_message_sync(struct iovec vectors[], int cnt, while (cnt > 0) { +<<<<<<< HEAD n = writev(entry->sockfd, vectors, cnt <= IOV_MAX ? cnt : IOV_MAX); if (n < 0) return errno == EAGAIN ? cnt : -1; +======= + if (!entry->ssl) + { + n = writev(entry->sockfd, vectors, cnt <= IOV_MAX ? cnt : IOV_MAX); + if (n < 0) + return errno == EAGAIN ? cnt : -1; + } + else if (vectors->iov_len > 0) + { + n = __ssl_writev(entry->ssl, vectors, cnt); + if (n <= 0) + return cnt; + } + else + n = 0; +>>>>>>> f20375dbc48c543ec675560e134c375118d0438b for (i = 0; i < cnt; i++) { if ((size_t)n >= vectors[i].iov_len) n -= vectors[i].iov_len; - else + else if (n > 0) { vectors[i].iov_base = (char *)vectors[i].iov_base + n; vectors[i].iov_len -= n; - break; + return cnt - i; } + else + break; } vectors += i; diff --git a/src/protocol/CMakeLists.txt b/src/protocol/CMakeLists.txt index 97881f77..c3f43e54 100644 --- a/src/protocol/CMakeLists.txt +++ b/src/protocol/CMakeLists.txt @@ -42,6 +42,4 @@ if (KAFKA STREQUAL "y") KafkaResult.cc ) add_library("protocol_kafka" OBJECT ${SRC}) - set_property(SOURCE KafkaMessage.cc APPEND PROPERTY COMPILE_OPTIONS "-fno-rtti") - set_property(SOURCE KafkaResult.cc APPEND PROPERTY COMPILE_OPTIONS "-fno-rtti") endif () diff --git a/src/protocol/KafkaDataTypes.cc b/src/protocol/KafkaDataTypes.cc index 870366bb..9a0be8ae 100644 --- a/src/protocol/KafkaDataTypes.cc +++ b/src/protocol/KafkaDataTypes.cc @@ -664,4 +664,12 @@ const KafkaMeta *get_meta(const char *topic, KafkaMetaList *meta_list) return NULL; } +KafkaSnappySink::~KafkaSnappySink() +{ +} + +KafkaSnappySource::~KafkaSnappySource() +{ +} + } /* namespace protocol */ diff --git a/src/protocol/KafkaDataTypes.h b/src/protocol/KafkaDataTypes.h index 325b332b..ca6013a0 100644 --- a/src/protocol/KafkaDataTypes.h +++ b/src/protocol/KafkaDataTypes.h @@ -1609,6 +1609,9 @@ public: private: KafkaBuffer *buffer; + +public: + virtual ~KafkaSnappySink(); }; class KafkaSnappySource : public snappy::Source @@ -1643,6 +1646,9 @@ private: KafkaBuffer *buffer; size_t buf_size; size_t pos; + +public: + virtual ~KafkaSnappySource(); }; } diff --git a/src/protocol/KafkaResult.cc b/src/protocol/KafkaResult.cc index f573ab5b..6292ded4 100644 --- a/src/protocol/KafkaResult.cc +++ b/src/protocol/KafkaResult.cc @@ -27,12 +27,6 @@ enum KAFKA_STATUS_END, }; -KafkaResult::KafkaResult() -{ - this->resp_vec = NULL; - this->resp_num = 0; -} - KafkaResult::~KafkaResult() { delete []this->resp_vec; diff --git a/src/protocol/KafkaResult.h b/src/protocol/KafkaResult.h index d2d16318..3ce575d7 100644 --- a/src/protocol/KafkaResult.h +++ b/src/protocol/KafkaResult.h @@ -43,7 +43,11 @@ public: void set_resp(KafkaResponse&& resp, size_t i); public: - KafkaResult(); + KafkaResult() + { + this->resp_vec = NULL; + this->resp_num = 0; + } virtual ~KafkaResult(); diff --git a/src/protocol/xmake.lua b/src/protocol/xmake.lua index 51a55edd..721b4f70 100644 --- a/src/protocol/xmake.lua +++ b/src/protocol/xmake.lua @@ -35,23 +35,14 @@ target("protocol") set_kind("object") add_deps("basic_protocol", "mysql_protocol", "redis_protocol") -target("kafka_message") - if has_config("kafka") then - add_files("KafkaMessage.cc", - "KafkaResult.cc") - set_kind("object") - add_cxxflags("-fno-rtti") - add_packages("lz4", "zstd", "zlib", "snappy") - else - set_kind("phony") - end - target("kafka_protocol") if has_config("kafka") then set_kind("object") add_files("kafka_parser.c", - "KafkaDataTypes.cc") - add_deps("kafka_message", "protocol") + "KafkaDataTypes.cc", + "KafkaMessage.cc", + "KafkaResult.cc") + add_deps("basic_protocol") add_packages("zlib", "snappy", "zstd", "lz4") else set_kind("phony")