-
Notifications
You must be signed in to change notification settings - Fork 240
/
Copy pathWebBinaryProxy.cpp
132 lines (113 loc) · 7.75 KB
/
WebBinaryProxy.cpp
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
#include <brynet/base/AppStatus.hpp>
#include <brynet/net/AsyncConnector.hpp>
#include <brynet/net/ListenThread.hpp>
#include <brynet/net/http/HttpService.hpp>
#include <brynet/net/http/WebSocketFormat.hpp>
#include <iostream>
#include <string>
#include <thread>
using namespace std;
using namespace brynet;
using namespace brynet::net;
int main(int argc, char** argv)
{
if (argc != 4)
{
fprintf(stderr, "Usage: <listen port> <backend ip> <backend port>");
exit(-1);
}
int bindPort = atoi(argv[1]);
string backendIP = argv[2];
int backendPort = atoi(argv[3]);
auto tcpService = IOThreadTcpService::Create();
tcpService->startWorkerThread(std::thread::hardware_concurrency());
auto asyncConnector = AsyncConnector::Create();
asyncConnector->startWorkerThread();
auto listenThread = ListenThread::Create(false,
"0.0.0.0",
bindPort,
[tcpService, asyncConnector, backendIP, backendPort](TcpSocket::Ptr socket) {
auto status = std::make_shared<bool>(true);
auto enterCallback = [tcpService, asyncConnector, backendIP, backendPort, status](const TcpConnection::Ptr& session) {
std::shared_ptr<TcpConnection::Ptr> shareBackendSession = std::make_shared<TcpConnection::Ptr>(nullptr);
std::shared_ptr<std::vector<string>> cachePacket = std::make_shared<std::vector<std::string>>();
auto enterCallback = [tcpService, session, shareBackendSession, cachePacket, status](TcpSocket::Ptr socket) {
auto enterCallback = [=](const TcpConnection::Ptr& backendSession) {
if (!*status) /*if http client already close*/
{
backendSession->postDisConnect();
return;
}
*shareBackendSession = backendSession;
for (auto& p : *cachePacket)
{
backendSession->send(p);
}
cachePacket->clear();
backendSession->setDisConnectCallback([=](const TcpConnection::Ptr& backendSession) {
(void) backendSession;
*shareBackendSession = nullptr;
if (*status)
{
session->postDisConnect();
}
});
backendSession->setDataCallback([=](brynet::base::BasePacketReader& reader) {
/* receive data from backend server, then send to http client */
session->send(reader.begin(), reader.size());
reader.consumeAll();
});
};
ConnectionOption option;
option.enterCallback.emplace_back(enterCallback);
option.maxRecvBufferSize = 32 * 1024;
tcpService->addTcpConnection(std::move(socket), option);
};
/* new connect to backend server */
ConnectOption option;
option.ip = backendIP;
option.port = backendPort;
option.timeout = std::chrono::seconds(10);
option.completedCallback = enterCallback;
asyncConnector->asyncConnect(option);
session->setDataCallback([=](brynet::base::BasePacketReader& reader) {
TcpConnection::Ptr backendSession = *shareBackendSession;
if (backendSession == nullptr)
{
/*cache it*/
cachePacket->push_back(std::string(reader.begin(), reader.size()));
}
else
{
/* receive data from http client, then send to backend server */
backendSession->send(reader.begin(), reader.size());
}
reader.consumeAll();
});
session->setDisConnectCallback([=](const TcpConnection::Ptr& session) {
/*if http client close, then close it's backend server */
TcpConnection::Ptr backendSession = *shareBackendSession;
if (backendSession != nullptr)
{
backendSession->postDisConnect();
}
*status = false;
});
};
ConnectionOption option;
option.enterCallback.emplace_back(enterCallback);
option.maxRecvBufferSize = 32 * 1024;
tcpService->addTcpConnection(std::move(socket), option);
});
// listen for front http client
listenThread->startListen();
while (true)
{
std::this_thread::sleep_for(std::chrono::seconds(1));
if (brynet::base::app_kbhit())
{
break;
}
}
return 0;
}