kapitan/IXWebSocket
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

(ixwebsocket) add WebSocketProxyServer, from ws. Still need to make the interface better.

Browse Source
This commit is contained in:
Benjamin Sergeant 2020-07-25 11:26:06 -07:00
parent 9a47ec1217
commit 0388459bd0
9 changed files with 146 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CMakeLists.txt
View File

@ -53,6 +53,7 @@ set( IXWEBSOCKET_SOURCES
ixwebsocket/IXWebSocketPerMessageDeflate.cpp
ixwebsocket/IXWebSocketPerMessageDeflateCodec.cpp
ixwebsocket/IXWebSocketPerMessageDeflateOptions.cpp
ixwebsocket/IXWebSocketProxyServer.cpp
ixwebsocket/IXWebSocketServer.cpp
ixwebsocket/IXWebSocketTransport.cpp
)
@ -96,6 +97,7 @@ set( IXWEBSOCKET_HEADERS
ixwebsocket/IXWebSocketPerMessageDeflate.h
ixwebsocket/IXWebSocketPerMessageDeflateCodec.h
ixwebsocket/IXWebSocketPerMessageDeflateOptions.h
ixwebsocket/IXWebSocketProxyServer.h
ixwebsocket/IXWebSocketSendInfo.h
ixwebsocket/IXWebSocketServer.h
ixwebsocket/IXWebSocketTransport.h

4
docs/CHANGELOG.md
View File

@ -1,6 +1,10 @@
# Changelog
All changes to this project will be documented in this file.
## [9.10.7] - 2020-07-25
(ixwebsocket) add WebSocketProxyServer, from ws. Still need to make the interface better.
## [9.10.6] - 2020-07-24
(ws) port broadcast_server sub-command to the new server API

117
ixwebsocket/IXWebSocketProxyServer.cpp Normal file
View File

@ -0,0 +1,117 @@
/*
* IXWebSocketProxyServer.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*/
#include "IXWebSocketProxyServer.h"
#include "IXWebSocketServer.h"
#include <sstream>
namespace ix
{
class ProxyConnectionState : public ix::ConnectionState
{
public:
ProxyConnectionState()
: _connected(false)
{
}
ix::WebSocket& webSocket()
{
return _serverWebSocket;
}
bool isConnected()
{
return _connected;
}
void setConnected()
{
_connected = true;
}
private:
ix::WebSocket _serverWebSocket;
bool _connected;
};
int websocket_proxy_server_main(int port,
const std::string& hostname,
const ix::SocketTLSOptions& tlsOptions,
const std::string& remoteUrl,
bool verbose)
{
ix::WebSocketServer server(port, hostname);
server.setTLSOptions(tlsOptions);
auto factory = []() -> std::shared_ptr<ix::ConnectionState> {
return std::make_shared<ProxyConnectionState>();
};
server.setConnectionStateFactory(factory);
server.setOnConnectionCallback(
[remoteUrl, verbose](std::shared_ptr<ix::WebSocket> webSocket,
std::shared_ptr<ConnectionState> connectionState,
std::unique_ptr<ConnectionInfo> connectionInfo) {
auto state = std::dynamic_pointer_cast<ProxyConnectionState>(connectionState);
auto remoteIp = connectionInfo->remoteIp;
// Server connection
state->webSocket().setOnMessageCallback(
[webSocket, state, remoteIp, verbose](const WebSocketMessagePtr& msg) {
if (msg->type == ix::WebSocketMessageType::Close)
{
state->setTerminated();
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
webSocket->send(msg->str, msg->binary);
}
});
// Client connection
webSocket->setOnMessageCallback(
[state, remoteUrl, verbose](const WebSocketMessagePtr& msg) {
if (msg->type == ix::WebSocketMessageType::Open)
{
// Connect to the 'real' server
std::string url(remoteUrl);
url += msg->openInfo.uri;
state->webSocket().setUrl(url);
state->webSocket().disableAutomaticReconnection();
state->webSocket().start();
// we should sleep here for a bit until we've established the
// connection with the remote server
while (state->webSocket().getReadyState() != ReadyState::Open)
{
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
state->webSocket().close(msg->closeInfo.code, msg->closeInfo.reason);
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
state->webSocket().send(msg->str, msg->binary);
}
});
});
auto res = server.listen();
if (!res.first)
{
return 1;
}
server.start();
server.wait();
return 0;
}
} // namespace ix

20
ixwebsocket/IXWebSocketProxyServer.h Normal file
View File

@ -0,0 +1,20 @@
/*
* IXWebSocketProxyServer.h
* Author: Benjamin Sergeant
* Copyright (c) 2019-2020 Machine Zone, Inc. All rights reserved.
*/
#pragma once
#include "IXSocketTLSOptions.h"
#include <cstdint>
#include <stddef.h>
#include <string>
namespace ix
{
int websocket_proxy_server_main(int port,
const std::string& hostname,
const ix::SocketTLSOptions& tlsOptions,
const std::string& remoteUrl,
bool verbose);
} // namespace ix

2
ixwebsocket/IXWebSocketVersion.h
View File

@ -6,4 +6,4 @@
#pragma once
#define IX_WEBSOCKET_VERSION "9.10.6"
#define IX_WEBSOCKET_VERSION "9.10.7"

1
ws/CMakeLists.txt
View File

@ -66,7 +66,6 @@ add_executable(ws
ws_cobra_publish.cpp
ws_httpd.cpp
ws_autobahn.cpp
ws_proxy_server.cpp
ws_sentry_minidump_upload.cpp
ws_dns_lookup.cpp
ws.cpp)

3
ws/ws.cpp
View File

@ -22,6 +22,7 @@
#include <ixwebsocket/IXNetSystem.h>
#include <ixwebsocket/IXSocket.h>
#include <ixwebsocket/IXUserAgent.h>
#include <ixwebsocket/IXWebSocketProxyServer.h>
#include <spdlog/sinks/basic_file_sink.h>
#include <spdlog/spdlog.h>
#include <sstream>
@ -657,7 +658,7 @@ int main(int argc, char** argv)
}
else if (app.got_subcommand("proxy_server"))
{
ret = ix::ws_proxy_server_main(port, hostname, tlsOptions, remoteHost, verbose);
ret = ix::websocket_proxy_server_main(port, hostname, tlsOptions, remoteHost, verbose);
}
else if (app.got_subcommand("upload_minidump"))
{

6
ws/ws.h
View File

@ -116,12 +116,6 @@ namespace ix
int ws_redis_server_main(int port, const std::string& hostname);
int ws_proxy_server_main(int port,
const std::string& hostname,
const ix::SocketTLSOptions& tlsOptions,
const std::string& remoteHost,
bool verbose);
int ws_sentry_minidump_upload(const std::string& metadataPath,
const std::string& minidump,
const std::string& project,

176
ws/ws_proxy_server.cpp
View File

@ -1,176 +0,0 @@
/*
* ws_proxy_server.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*/
#include <ixwebsocket/IXWebSocketServer.h>
#include <spdlog/spdlog.h>
#include <sstream>
namespace ix
{
class ProxyConnectionState : public ix::ConnectionState
{
public:
ProxyConnectionState()
: _connected(false)
{
}
ix::WebSocket& webSocket()
{
return _serverWebSocket;
}
bool isConnected()
{
return _connected;
}
void setConnected()
{
_connected = true;
}
private:
ix::WebSocket _serverWebSocket;
bool _connected;
};
int ws_proxy_server_main(int port,
const std::string& hostname,
const ix::SocketTLSOptions& tlsOptions,
const std::string& remoteUrl,
bool verbose)
{
spdlog::info("Listening on {}:{}", hostname, port);
ix::WebSocketServer server(port, hostname);
server.setTLSOptions(tlsOptions);
auto factory = []() -> std::shared_ptr<ix::ConnectionState> {
return std::make_shared<ProxyConnectionState>();
};
server.setConnectionStateFactory(factory);
server.setOnConnectionCallback([remoteUrl,
verbose](std::shared_ptr<ix::WebSocket> webSocket,
std::shared_ptr<ConnectionState> connectionState,
std::unique_ptr<ConnectionInfo> connectionInfo) {
auto state = std::dynamic_pointer_cast<ProxyConnectionState>(connectionState);
auto remoteIp = connectionInfo->remoteIp;
// Server connection
state->webSocket().setOnMessageCallback([webSocket, state, remoteIp, verbose](
const WebSocketMessagePtr& msg) {
if (msg->type == ix::WebSocketMessageType::Open)
{
spdlog::info("New connection to remote server");
spdlog::info("remote ip: {}", remoteIp);
spdlog::info("id: {}", state->getId());
spdlog::info("Uri: {}", msg->openInfo.uri);
spdlog::info("Headers:");
for (auto it : msg->openInfo.headers)
{
spdlog::info("{}: {}", it.first, it.second);
}
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
spdlog::info("Closed remote server connection: client id {} code {} reason {}",
state->getId(),
msg->closeInfo.code,
msg->closeInfo.reason);
state->setTerminated();
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
spdlog::error("Connection error: {}", msg->errorInfo.reason);
spdlog::error("#retries: {}", msg->errorInfo.retries);
spdlog::error("Wait time(ms): {}", msg->errorInfo.wait_time);
spdlog::error("HTTP Status: {}", msg->errorInfo.http_status);
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
spdlog::info("Received {} bytes from server", msg->wireSize);
if (verbose)
{
spdlog::info("payload {}", msg->str);
}
webSocket->send(msg->str, msg->binary);
}
});
// Client connection
webSocket->setOnMessageCallback(
[state, remoteUrl, verbose](const WebSocketMessagePtr& msg) {
if (msg->type == ix::WebSocketMessageType::Open)
{
spdlog::info("New connection from client");
spdlog::info("id: {}", state->getId());
spdlog::info("Uri: {}", msg->openInfo.uri);
spdlog::info("Headers:");
for (auto it : msg->openInfo.headers)
{
spdlog::info("{}: {}", it.first, it.second);
}
// Connect to the 'real' server
std::string url(remoteUrl);
url += msg->openInfo.uri;
state->webSocket().setUrl(url);
state->webSocket().disableAutomaticReconnection();
state->webSocket().start();
// we should sleep here for a bit until we've established the
// connection with the remote server
while (state->webSocket().getReadyState() != ReadyState::Open)
{
spdlog::info("waiting for server connection establishment");
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
spdlog::info("server connection established");
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
spdlog::info("Closed client connection: client id {} code {} reason {}",
state->getId(),
msg->closeInfo.code,
msg->closeInfo.reason);
state->webSocket().close(msg->closeInfo.code, msg->closeInfo.reason);
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
spdlog::error("Connection error: {}", msg->errorInfo.reason);
spdlog::error("#retries: {}", msg->errorInfo.retries);
spdlog::error("Wait time(ms): {}", msg->errorInfo.wait_time);
spdlog::error("HTTP Status: {}", msg->errorInfo.http_status);
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
spdlog::info("Received {} bytes from client", msg->wireSize);
if (verbose)
{
spdlog::info("payload {}", msg->str);
}
state->webSocket().send(msg->str, msg->binary);
}
});
});
auto res = server.listen();
if (!res.first)
{
spdlog::info(res.second);
return 1;
}
server.start();
server.wait();
return 0;
}
} // namespace ix