From f95e88b04cba6b0b4198396b420e92889c79cfc6 Mon Sep 17 00:00:00 2001 From: XieHan Date: Mon, 29 Aug 2022 18:05:33 +0800 Subject: [PATCH 1/2] Add creating client task with sockaddr --- src/factory/WFTaskFactory.h | 6 ++++++ src/factory/WFTaskFactory.inl | 15 ++++++++++++++- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/src/factory/WFTaskFactory.h b/src/factory/WFTaskFactory.h index 42bbcc80..a9141258 100644 --- a/src/factory/WFTaskFactory.h +++ b/src/factory/WFTaskFactory.h @@ -362,6 +362,12 @@ public: int retry_max, std::function callback); + static T *create_client_task(TransportType type, + const struct sockaddr *addr, + socklen_t addrlen, + int retry_max, + std::function callback); + public: static T *create_server_task(CommService *service, std::function& process); diff --git a/src/factory/WFTaskFactory.inl b/src/factory/WFTaskFactory.inl index 8a9b0227..e6aa65fa 100644 --- a/src/factory/WFTaskFactory.inl +++ b/src/factory/WFTaskFactory.inl @@ -311,7 +311,6 @@ void WFComplexClientTask::init(TransportType type, auto params = WFGlobal::get_global_settings()->endpoint_params; struct addrinfo addrinfo = { }; addrinfo.ai_family = addr->sa_family; - addrinfo.ai_socktype = SOCK_STREAM; addrinfo.ai_addr = (struct sockaddr *)addr; addrinfo.ai_addrlen = addrlen; @@ -618,6 +617,20 @@ WFNetworkTaskFactory::create_client_task(TransportType type, return task; } +template +WFNetworkTask * +WFNetworkTaskFactory::create_client_task(TransportType type, + const struct sockaddr *addr, + socklen_t addrlen, + int retry_max, + std::function *)> callback) +{ + auto *task = new WFComplexClientTask(retry_max, std::move(callback)); + + task->init(type, addr, addrlen, ""); + return task; +} + template WFNetworkTask * WFNetworkTaskFactory::create_server_task(CommService *service, From 6fd0d0c385c442a1682fa9710d12828a22168600 Mon Sep 17 00:00:00 2001 From: XieHan Date: Mon, 29 Aug 2022 18:55:11 +0800 Subject: [PATCH 2/2] Add unix domain socket example --- tutorial/CMakeLists.txt | 8 ++ .../client-uds.cc | 117 ++++++++++++++++++ .../server-uds.cc | 92 ++++++++++++++ 3 files changed, 217 insertions(+) create mode 100644 tutorial/tutorial-10-user_defined_protocol/client-uds.cc create mode 100644 tutorial/tutorial-10-user_defined_protocol/server-uds.cc diff --git a/tutorial/CMakeLists.txt b/tutorial/CMakeLists.txt index ad9a27f3..c9db7511 100644 --- a/tutorial/CMakeLists.txt +++ b/tutorial/CMakeLists.txt @@ -94,13 +94,21 @@ endif () set(DIR10 tutorial-10-user_defined_protocol) add_executable(server ${DIR10}/server.cc ${DIR10}/message.cc) add_executable(client ${DIR10}/client.cc ${DIR10}/message.cc) +add_executable(server-uds ${DIR10}/server-uds.cc ${DIR10}/message.cc) +add_executable(client-uds ${DIR10}/client-uds.cc ${DIR10}/message.cc) target_link_libraries(server ${WORKFLOW_LIB}) target_link_libraries(client ${WORKFLOW_LIB}) +target_link_libraries(server-uds ${WORKFLOW_LIB}) +target_link_libraries(client-uds ${WORKFLOW_LIB}) set_target_properties(server PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${PROJECT_SOURCE_DIR}/${DIR10}) set_target_properties(client PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${PROJECT_SOURCE_DIR}/${DIR10}) +set_target_properties(server-uds PROPERTIES + RUNTIME_OUTPUT_DIRECTORY ${PROJECT_SOURCE_DIR}/${DIR10}) +set_target_properties(client-uds PROPERTIES + RUNTIME_OUTPUT_DIRECTORY ${PROJECT_SOURCE_DIR}/${DIR10}) set(DIR16 tutorial-16-graceful_restart) add_executable(bootstrap ${DIR16}/bootstrap.c) diff --git a/tutorial/tutorial-10-user_defined_protocol/client-uds.cc b/tutorial/tutorial-10-user_defined_protocol/client-uds.cc new file mode 100644 index 00000000..5c2f42ac --- /dev/null +++ b/tutorial/tutorial-10-user_defined_protocol/client-uds.cc @@ -0,0 +1,117 @@ +/* + Copyright (c) 2022 Sogou, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + + Author: Xie Han (xiehan@sogou-inc.com;63350856@qq.com) +*/ + +#include +#include +#include +#include +#include "workflow/Workflow.h" +#include "workflow/WFTaskFactory.h" +#include "workflow/WFFacilities.h" +#include "message.h" + +using WFTutorialTask = WFNetworkTask; +using tutorial_callback_t = std::function; + +using namespace protocol; + +class MyFactory : public WFTaskFactory +{ +public: + static WFTutorialTask *create_tutorial_task(const struct sockaddr *addr, + socklen_t addrlen, + int retry_max, + tutorial_callback_t callback) + { + using NTF = WFNetworkTaskFactory; + WFTutorialTask *task = NTF::create_client_task(TT_TCP, addr, addrlen, + retry_max, + std::move(callback)); + task->set_keep_alive(30 * 1000); + return task; + } +}; + +int main(int argc, char *argv[]) +{ + const char *path; + std::string host; + + if (argc != 2) + { + fprintf(stderr, "USAGE: %s \n", argv[0]); + exit(1); + } + + path = argv[1]; + + auto&& create = [path](WFRepeaterTask *)->SubTask *{ + char buf[1024]; + printf("Input next request string (Ctrl-D to exit): "); + *buf = '\0'; + scanf("%1023s", buf); + size_t body_size = strlen(buf); + if (body_size == 0) + { + printf("\n"); + return NULL; + } + + struct sockaddr_un sun = { }; + sun.sun_family = AF_UNIX; + strncpy(sun.sun_path, path, sizeof sun.sun_path); + WFTutorialTask *task = MyFactory::create_tutorial_task( + (struct sockaddr *)&sun, sizeof sun, + 0, + [](WFTutorialTask *task) { + int state = task->get_state(); + int error = task->get_error(); + TutorialResponse *resp = task->get_resp(); + void *body; + size_t body_size; + + if (state == WFT_STATE_SUCCESS) + { + resp->get_message_body_nocopy(&body, &body_size); + printf("Server Response: %.*s\n", (int)body_size, (char *)body); + } + else + { + const char *str = WFGlobal::get_error_string(state, error); + fprintf(stderr, "Error: %s\n", str); + } + }); + + task->get_req()->set_message_body(buf, body_size); + task->get_resp()->set_size_limit(4 * 1024); + return task; + }; + + WFFacilities::WaitGroup wait_group(1); + + WFRepeaterTask *repeater; + repeater = WFTaskFactory::create_repeater_task(std::move(create), nullptr); + Workflow::start_series_work(repeater, [&wait_group](const SeriesWork *) { + wait_group.done(); + }); + + wait_group.wait(); + return 0; +} + diff --git a/tutorial/tutorial-10-user_defined_protocol/server-uds.cc b/tutorial/tutorial-10-user_defined_protocol/server-uds.cc new file mode 100644 index 00000000..7dc905de --- /dev/null +++ b/tutorial/tutorial-10-user_defined_protocol/server-uds.cc @@ -0,0 +1,92 @@ +/* + Copyright (c) 2022 Sogou, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + + Author: Xie Han (xiehan@sogou-inc.com;63350856@qq.com) +*/ + +#include +#include +#include +#include +#include +#include +#include "workflow/Workflow.h" +#include "workflow/WFTaskFactory.h" +#include "workflow/WFServer.h" +#include "workflow/WFFacilities.h" +#include "message.h" + +using WFTutorialTask = WFNetworkTask; +using WFTutorialServer = WFServer; + +using namespace protocol; + +void process(WFTutorialTask *task) +{ + TutorialRequest *req = task->get_req(); + TutorialResponse *resp = task->get_resp(); + void *body; + size_t size; + size_t i; + + req->get_message_body_nocopy(&body, &size); + for (i = 0; i < size; i++) + ((char *)body)[i] = toupper(((char *)body)[i]); + + resp->set_message_body(body, size); +} + +static WFFacilities::WaitGroup wait_group(1); + +void sig_handler(int signo) +{ + wait_group.done(); +} + +int main(int argc, char *argv[]) +{ + struct sockaddr_un sun = { }; + + if (argc != 2) + { + fprintf(stderr, "USAGE %s \n", argv[0]); + exit(1); + } + + sun.sun_family = AF_UNIX; + strncpy(sun.sun_path, argv[1], sizeof sun.sun_path); + + signal(SIGINT, sig_handler); + + struct WFServerParams params = SERVER_PARAMS_DEFAULT; + params.request_size_limit = 4 * 1024; + + WFTutorialServer server(¶ms, process); + if (server.start((struct sockaddr *)&sun, sizeof sun) == 0) + { + wait_group.wait(); + server.stop(); + } + else + { + perror("server.start"); + exit(1); + } + + return 0; +} +