-
Notifications
You must be signed in to change notification settings - Fork 4k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Batch create and accept stream (#2754)
* feat: batch create and accept stream * fix protobuf 22.5 compilation error related to thread_local in MacOS * refine style * modify code based on the code review feedback and add more tests modify code based on the code review feedback and add more tests
- Loading branch information
Showing
16 changed files
with
649 additions
and
67 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,140 @@ | ||
| # 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. | ||
|
|
||
| cmake_minimum_required(VERSION 2.8.10) | ||
| project(streaming_batch_echo_c++ C CXX) | ||
|
|
||
| option(LINK_SO "Whether examples are linked dynamically" OFF) | ||
|
|
||
| execute_process( | ||
| COMMAND bash -c "find ${PROJECT_SOURCE_DIR}/../.. -type d -regex \".*output/include$\" | head -n1 | xargs dirname | tr -d '\n'" | ||
| OUTPUT_VARIABLE OUTPUT_PATH | ||
| ) | ||
|
|
||
| set(CMAKE_PREFIX_PATH ${OUTPUT_PATH}) | ||
|
|
||
| include(FindThreads) | ||
| include(FindProtobuf) | ||
| protobuf_generate_cpp(PROTO_SRC PROTO_HEADER echo.proto) | ||
| # include PROTO_HEADER | ||
| include_directories(${CMAKE_CURRENT_BINARY_DIR}) | ||
|
|
||
| # Search for libthrift* by best effort. If it is not found and brpc is | ||
| # compiled with thrift protocol enabled, a link error would be reported. | ||
| find_library(THRIFT_LIB NAMES thrift) | ||
| if (NOT THRIFT_LIB) | ||
| set(THRIFT_LIB "") | ||
| endif() | ||
|
|
||
| find_path(BRPC_INCLUDE_PATH NAMES brpc/server.h) | ||
| if(LINK_SO) | ||
| find_library(BRPC_LIB NAMES brpc) | ||
| else() | ||
| find_library(BRPC_LIB NAMES libbrpc.a brpc) | ||
| endif() | ||
| if((NOT BRPC_INCLUDE_PATH) OR (NOT BRPC_LIB)) | ||
| message(FATAL_ERROR "Fail to find brpc") | ||
| endif() | ||
| include_directories(${BRPC_INCLUDE_PATH}) | ||
|
|
||
| find_path(GFLAGS_INCLUDE_PATH gflags/gflags.h) | ||
| find_library(GFLAGS_LIBRARY NAMES gflags libgflags) | ||
| if((NOT GFLAGS_INCLUDE_PATH) OR (NOT GFLAGS_LIBRARY)) | ||
| message(FATAL_ERROR "Fail to find gflags") | ||
| endif() | ||
| include_directories(${GFLAGS_INCLUDE_PATH}) | ||
|
|
||
| execute_process( | ||
| COMMAND bash -c "grep \"namespace [_A-Za-z0-9]\\+ {\" ${GFLAGS_INCLUDE_PATH}/gflags/gflags_declare.h | head -1 | awk '{print $2}' | tr -d '\n'" | ||
| OUTPUT_VARIABLE GFLAGS_NS | ||
| ) | ||
| if(${GFLAGS_NS} STREQUAL "GFLAGS_NAMESPACE") | ||
| execute_process( | ||
| COMMAND bash -c "grep \"#define GFLAGS_NAMESPACE [_A-Za-z0-9]\\+\" ${GFLAGS_INCLUDE_PATH}/gflags/gflags_declare.h | head -1 | awk '{print $3}' | tr -d '\n'" | ||
| OUTPUT_VARIABLE GFLAGS_NS | ||
| ) | ||
| endif() | ||
| if(CMAKE_SYSTEM_NAME STREQUAL "Darwin") | ||
| include(CheckFunctionExists) | ||
| CHECK_FUNCTION_EXISTS(clock_gettime HAVE_CLOCK_GETTIME) | ||
| if(NOT HAVE_CLOCK_GETTIME) | ||
| set(DEFINE_CLOCK_GETTIME "-DNO_CLOCK_GETTIME_IN_MAC") | ||
| endif() | ||
| endif() | ||
|
|
||
| set(CMAKE_CPP_FLAGS "${DEFINE_CLOCK_GETTIME} -DGFLAGS_NS=${GFLAGS_NS}") | ||
| set(CMAKE_CXX_FLAGS "${CMAKE_CPP_FLAGS} -DNDEBUG -O2 -D__const__=__unused__ -pipe -W -Wall -Wno-unused-parameter -fPIC -fno-omit-frame-pointer") | ||
|
|
||
| if(CMAKE_VERSION VERSION_LESS "3.1.3") | ||
| if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU") | ||
| set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11") | ||
| endif() | ||
| if(CMAKE_CXX_COMPILER_ID STREQUAL "Clang") | ||
| set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11") | ||
| endif() | ||
| else() | ||
| set(CMAKE_CXX_STANDARD 11) | ||
| set(CMAKE_CXX_STANDARD_REQUIRED ON) | ||
| endif() | ||
|
|
||
| find_path(LEVELDB_INCLUDE_PATH NAMES leveldb/db.h) | ||
| find_library(LEVELDB_LIB NAMES leveldb) | ||
| if ((NOT LEVELDB_INCLUDE_PATH) OR (NOT LEVELDB_LIB)) | ||
| message(FATAL_ERROR "Fail to find leveldb") | ||
| endif() | ||
| include_directories(${LEVELDB_INCLUDE_PATH}) | ||
|
|
||
| if(CMAKE_SYSTEM_NAME STREQUAL "Darwin") | ||
| set(OPENSSL_ROOT_DIR | ||
| "/usr/local/opt/openssl" # Homebrew installed OpenSSL | ||
| ) | ||
| endif() | ||
|
|
||
| find_package(OpenSSL) | ||
| include_directories(${OPENSSL_INCLUDE_DIR}) | ||
|
|
||
| set(DYNAMIC_LIB | ||
| ${CMAKE_THREAD_LIBS_INIT} | ||
| ${GFLAGS_LIBRARY} | ||
| ${PROTOBUF_LIBRARIES} | ||
| ${LEVELDB_LIB} | ||
| ${OPENSSL_CRYPTO_LIBRARY} | ||
| ${OPENSSL_SSL_LIBRARY} | ||
| ${THRIFT_LIB} | ||
| dl | ||
| ) | ||
|
|
||
| if(CMAKE_SYSTEM_NAME STREQUAL "Darwin") | ||
| set(DYNAMIC_LIB ${DYNAMIC_LIB} | ||
| pthread | ||
| "-framework CoreFoundation" | ||
| "-framework CoreGraphics" | ||
| "-framework CoreData" | ||
| "-framework CoreText" | ||
| "-framework Security" | ||
| "-framework Foundation" | ||
| "-Wl,-U,_MallocExtension_ReleaseFreeMemory" | ||
| "-Wl,-U,_ProfilerStart" | ||
| "-Wl,-U,_ProfilerStop" | ||
| "-Wl,-U,__Z13GetStackTracePPvii") | ||
| endif() | ||
|
|
||
| add_executable(streaming_batch_echo_client client.cpp ${PROTO_SRC} ${PROTO_HEADER}) | ||
| add_executable(streaming_batch_echo_server server.cpp ${PROTO_SRC} ${PROTO_HEADER}) | ||
|
|
||
| target_link_libraries(streaming_batch_echo_client ${BRPC_LIB} ${DYNAMIC_LIB}) | ||
| target_link_libraries(streaming_batch_echo_server ${BRPC_LIB} ${DYNAMIC_LIB}) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,118 @@ | ||
| // 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. | ||
|
|
||
| // A client sending requests to server in batch every 1 second. | ||
|
|
||
| #include <gflags/gflags.h> | ||
| #include <butil/logging.h> | ||
| #include <brpc/channel.h> | ||
| #include <brpc/stream.h> | ||
| #include "echo.pb.h" | ||
|
|
||
| DEFINE_bool(send_attachment, true, "Carry attachment along with requests"); | ||
| DEFINE_string(connection_type, "", "Connection type. Available values: single, pooled, short"); | ||
| DEFINE_string(server, "0.0.0.0:8001", "IP Address of server"); | ||
| DEFINE_int32(timeout_ms, 100, "RPC timeout in milliseconds"); | ||
| DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)"); | ||
|
|
||
| class StreamClientReceiver : public brpc::StreamInputHandler { | ||
| public: | ||
| virtual int on_received_messages(brpc::StreamId id, | ||
| butil::IOBuf *const messages[], | ||
| size_t size) { | ||
| std::ostringstream os; | ||
| for (size_t i = 0; i < size; ++i) { | ||
| os << "msg[" << i << "]=" << *messages[i]; | ||
| } | ||
| LOG(INFO) << "Received from Stream=" << id << ": " << os.str(); | ||
| return 0; | ||
| } | ||
| virtual void on_idle_timeout(brpc::StreamId id) { | ||
| LOG(INFO) << "Stream=" << id << " has no data transmission for a while"; | ||
| } | ||
| virtual void on_closed(brpc::StreamId id) { | ||
| LOG(INFO) << "Stream=" << id << " is closed"; | ||
| } | ||
|
|
||
| virtual void on_finished(brpc::StreamId id, int32_t finish_code) { | ||
| LOG(INFO) << "Stream=" << id << " is finished, code " << finish_code; | ||
| } | ||
| }; | ||
|
|
||
| int main(int argc, char* argv[]) { | ||
| // Parse gflags. We recommend you to use gflags as well. | ||
| GFLAGS_NS::ParseCommandLineFlags(&argc, &argv, true); | ||
|
|
||
| // A Channel represents a communication line to a Server. Notice that | ||
| // Channel is thread-safe and can be shared by all threads in your program. | ||
| brpc::Channel channel; | ||
|
|
||
| // Initialize the channel, NULL means using default options. | ||
| brpc::ChannelOptions options; | ||
| options.protocol = brpc::PROTOCOL_BAIDU_STD; | ||
| options.connection_type = FLAGS_connection_type; | ||
| options.timeout_ms = FLAGS_timeout_ms/*milliseconds*/; | ||
| options.max_retry = FLAGS_max_retry; | ||
| if (channel.Init(FLAGS_server.c_str(), NULL) != 0) { | ||
| LOG(ERROR) << "Fail to initialize channel"; | ||
| return -1; | ||
| } | ||
|
|
||
| // Normally, you should not call a Channel directly, but instead construct | ||
| // a stub Service wrapping it. stub can be shared by all threads as well. | ||
| example::EchoService_Stub stub(&channel); | ||
| StreamClientReceiver receiver; | ||
| brpc::Controller cntl; | ||
| brpc::StreamIds streams; | ||
| brpc::StreamOptions stream_options; | ||
| stream_options.handler = &receiver; | ||
| if (brpc::StreamCreate(streams, 3, cntl, &stream_options) != 0) { | ||
| LOG(ERROR) << "Fail to create stream"; | ||
| return -1; | ||
| } | ||
| for(size_t i = 0; i < streams.size(); ++i) { | ||
| LOG(INFO) << "Created Stream=" << streams[i]; | ||
| } | ||
| example::EchoRequest request; | ||
| example::EchoResponse response; | ||
| request.set_message("I'm a RPC to connect stream"); | ||
| stub.Echo(&cntl, &request, &response, NULL); | ||
| if (cntl.Failed()) { | ||
| LOG(ERROR) << "Fail to connect stream, " << cntl.ErrorText(); | ||
| return -1; | ||
| } | ||
|
|
||
| while (!brpc::IsAskedToQuit()) { | ||
| butil::IOBuf msg1; | ||
| msg1.append("abcdefghijklmnopqrstuvwxyz"); | ||
| CHECK_EQ(0, brpc::StreamWrite(streams[0], msg1)); | ||
| butil::IOBuf msg2; | ||
| msg2.append("0123456789"); | ||
| CHECK_EQ(0, brpc::StreamWrite(streams[1], msg2)); | ||
| sleep(1); | ||
| butil::IOBuf msg3; | ||
| msg3.append("hello world"); | ||
| CHECK_EQ(0, brpc::StreamWrite(streams[2], msg3)); | ||
| sleep(1); | ||
| } | ||
|
|
||
| CHECK_EQ(0, brpc::StreamClose(streams[0])); | ||
| CHECK_EQ(0, brpc::StreamClose(streams[1])); | ||
| CHECK_EQ(0, brpc::StreamClose(streams[2])); | ||
| LOG(INFO) << "EchoClient is going to quit"; | ||
| return 0; | ||
| } |
Oops, something went wrong.