GithubStar/workflow
1
0
Fork 0
mirror of https://github.com/sogou/workflow.git synced 2026-10-24 16:33:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1032 from Barenboim/master

Browse Source 
Add creating client task with sockaddr and Add UDS example
This commit is contained in:
xiehan
2022-08-29 19:09:59 +08:00
committed by GitHub
parent 116e6772cd 6fd0d0c385
commit 1787544c9c
5 changed files with 237 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/factory/WFTaskFactory.h
+6
View File
@@ -362,6 +362,12 @@ public:
int retry_max,
std::function<void (T *)> callback);
static T *create_client_task(TransportType type,
const struct sockaddr *addr,
socklen_t addrlen,
int retry_max,
std::function<void (T *)> callback);
public:
static T *create_server_task(CommService *service,
std::function<void (T *)>& process);
src/factory/WFTaskFactory.inl
+14 -1
View File
@@ -311,7 +311,6 @@ void WFComplexClientTask<REQ, RESP, CTX>::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<REQ, RESP>::create_client_task(TransportType type,
return task;
}
template<class REQ, class RESP>
WFNetworkTask<REQ, RESP> *
WFNetworkTaskFactory<REQ, RESP>::create_client_task(TransportType type,
const struct sockaddr *addr,
socklen_t addrlen,
int retry_max,
std::function<void (WFNetworkTask<REQ, RESP> *)> callback)
{
auto *task = new WFComplexClientTask<REQ, RESP>(retry_max, std::move(callback));
task->init(type, addr, addrlen, "");
return task;
}
template<class REQ, class RESP>
WFNetworkTask<REQ, RESP> *
WFNetworkTaskFactory<REQ, RESP>::create_server_task(CommService *service,
tutorial/CMakeLists.txt
+8
View File
@@ -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)
tutorial/tutorial-10-user_defined_protocol/client-uds.cc
+117
View File
@@ -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 <sys/un.h>
#include <string.h>
#include <stdio.h>
#include <string.h>
#include "workflow/Workflow.h"
#include "workflow/WFTaskFactory.h"
#include "workflow/WFFacilities.h"
#include "message.h"
using WFTutorialTask = WFNetworkTask<protocol::TutorialRequest,
protocol::TutorialResponse>;
using tutorial_callback_t = std::function<void (WFTutorialTask *)>;
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<TutorialRequest, TutorialResponse>;
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 <path>\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;
}
tutorial/tutorial-10-user_defined_protocol/server-uds.cc
+92
View File
@@ -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 <sys/un.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <ctype.h>
#include <signal.h>
#include "workflow/Workflow.h"
#include "workflow/WFTaskFactory.h"
#include "workflow/WFServer.h"
#include "workflow/WFFacilities.h"
#include "message.h"
using WFTutorialTask = WFNetworkTask<protocol::TutorialRequest,
protocol::TutorialResponse>;
using WFTutorialServer = WFServer<protocol::TutorialRequest,
protocol::TutorialResponse>;
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 <path>\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(&params, process);
if (server.start((struct sockaddr *)&sun, sizeof sun) == 0)
{
wait_group.wait();
server.stop();
}
else
{
perror("server.start");
exit(1);
}
return 0;
}