(socket servers) merge the ConnectionInfo class with the ConnectionState one, which simplify all the server apis
This commit is contained in:
@ -1,25 +0,0 @@
|
||||
/*
|
||||
* IXConnectionInfo.h
|
||||
* Author: Benjamin Sergeant
|
||||
* Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <string>
|
||||
|
||||
namespace ix
|
||||
{
|
||||
struct ConnectionInfo
|
||||
{
|
||||
std::string remoteIp;
|
||||
int remotePort;
|
||||
|
||||
ConnectionInfo(const std::string& r = std::string(), int p = 0)
|
||||
: remoteIp(r)
|
||||
, remotePort(p)
|
||||
{
|
||||
;
|
||||
}
|
||||
};
|
||||
} // namespace ix
|
@ -50,4 +50,24 @@ namespace ix
|
||||
_onSetTerminatedCallback();
|
||||
}
|
||||
}
|
||||
|
||||
const std::string& ConnectionState::getRemoteIp()
|
||||
{
|
||||
return _remoteIp;
|
||||
}
|
||||
|
||||
int ConnectionState::getRemotePort()
|
||||
{
|
||||
return _remotePort;
|
||||
}
|
||||
|
||||
void ConnectionState::setRemoteIp(const std::string& remoteIp)
|
||||
{
|
||||
_remoteIp = remoteIp;
|
||||
}
|
||||
|
||||
void ConnectionState::setRemotePort(int remotePort)
|
||||
{
|
||||
_remotePort = remotePort;
|
||||
}
|
||||
} // namespace ix
|
||||
|
@ -28,11 +28,17 @@ namespace ix
|
||||
void setTerminated();
|
||||
bool isTerminated() const;
|
||||
|
||||
const std::string& getRemoteIp();
|
||||
int getRemotePort();
|
||||
|
||||
static std::shared_ptr<ConnectionState> createConnectionState();
|
||||
|
||||
private:
|
||||
void setOnSetTerminatedCallback(const OnSetTerminatedCallback& callback);
|
||||
|
||||
void setRemoteIp(const std::string& remoteIp);
|
||||
void setRemotePort(int remotePort);
|
||||
|
||||
protected:
|
||||
std::atomic<bool> _terminated;
|
||||
std::string _id;
|
||||
@ -40,6 +46,9 @@ namespace ix
|
||||
|
||||
static std::atomic<uint64_t> _globalId;
|
||||
|
||||
std::string _remoteIp;
|
||||
int _remotePort;
|
||||
|
||||
friend class SocketServer;
|
||||
};
|
||||
} // namespace ix
|
||||
|
@ -120,8 +120,7 @@ namespace ix
|
||||
}
|
||||
|
||||
void HttpServer::handleConnection(std::unique_ptr<Socket> socket,
|
||||
std::shared_ptr<ConnectionState> connectionState,
|
||||
std::unique_ptr<ConnectionInfo> connectionInfo)
|
||||
std::shared_ptr<ConnectionState> connectionState)
|
||||
{
|
||||
_connectedClientsCount++;
|
||||
|
||||
@ -130,8 +129,7 @@ namespace ix
|
||||
|
||||
if (std::get<0>(ret))
|
||||
{
|
||||
auto response =
|
||||
_onConnectionCallback(std::get<2>(ret), connectionState, std::move(connectionInfo));
|
||||
auto response = _onConnectionCallback(std::get<2>(ret), connectionState);
|
||||
if (!Http::sendResponse(response, socket))
|
||||
{
|
||||
logError("Cannot send response");
|
||||
@ -151,8 +149,7 @@ namespace ix
|
||||
{
|
||||
setOnConnectionCallback(
|
||||
[this](HttpRequestPtr request,
|
||||
std::shared_ptr<ConnectionState> /*connectionState*/,
|
||||
std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr {
|
||||
std::shared_ptr<ConnectionState> connectionState) -> HttpResponsePtr {
|
||||
std::string uri(request->uri);
|
||||
if (uri.empty() || uri == "/")
|
||||
{
|
||||
@ -184,8 +181,8 @@ namespace ix
|
||||
|
||||
// Log request
|
||||
std::stringstream ss;
|
||||
ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
|
||||
<< request->method << " " << request->headers["User-Agent"] << " "
|
||||
ss << connectionState->getRemoteIp() << ":" << connectionState->getRemotePort()
|
||||
<< " " << request->method << " " << request->headers["User-Agent"] << " "
|
||||
<< request->uri << " " << content.size();
|
||||
logInfo(ss.str());
|
||||
|
||||
@ -209,16 +206,16 @@ namespace ix
|
||||
// See https://developer.mozilla.org/en-US/docs/Web/HTTP/Redirections
|
||||
//
|
||||
setOnConnectionCallback(
|
||||
[this, redirectUrl](HttpRequestPtr request,
|
||||
std::shared_ptr<ConnectionState> /*connectionState*/,
|
||||
std::unique_ptr<ConnectionInfo> connectionInfo) -> HttpResponsePtr {
|
||||
[this,
|
||||
redirectUrl](HttpRequestPtr request,
|
||||
std::shared_ptr<ConnectionState> connectionState) -> HttpResponsePtr {
|
||||
WebSocketHttpHeaders headers;
|
||||
headers["Server"] = userAgent();
|
||||
|
||||
// Log request
|
||||
std::stringstream ss;
|
||||
ss << connectionInfo->remoteIp << ":" << connectionInfo->remotePort << " "
|
||||
<< request->method << " " << request->headers["User-Agent"] << " "
|
||||
ss << connectionState->getRemoteIp() << ":" << connectionState->getRemotePort()
|
||||
<< " " << request->method << " " << request->headers["User-Agent"] << " "
|
||||
<< request->uri;
|
||||
logInfo(ss.str());
|
||||
|
||||
|
@ -23,9 +23,7 @@ namespace ix
|
||||
{
|
||||
public:
|
||||
using OnConnectionCallback =
|
||||
std::function<HttpResponsePtr(HttpRequestPtr,
|
||||
std::shared_ptr<ConnectionState>,
|
||||
std::unique_ptr<ConnectionInfo> connectionInfo)>;
|
||||
std::function<HttpResponsePtr(HttpRequestPtr, std::shared_ptr<ConnectionState>)>;
|
||||
|
||||
HttpServer(int port = SocketServer::kDefaultPort,
|
||||
const std::string& host = SocketServer::kDefaultHost,
|
||||
@ -46,8 +44,7 @@ namespace ix
|
||||
|
||||
// Methods
|
||||
virtual void handleConnection(std::unique_ptr<Socket>,
|
||||
std::shared_ptr<ConnectionState> connectionState,
|
||||
std::unique_ptr<ConnectionInfo> connectionInfo) final;
|
||||
std::shared_ptr<ConnectionState> connectionState) final;
|
||||
virtual size_t getConnectedClientsCount() final;
|
||||
|
||||
void setDefaultConnectionCallback();
|
||||
|
@ -332,12 +332,13 @@ namespace ix
|
||||
}
|
||||
|
||||
// Retrieve connection info, the ip address of the remote peer/client)
|
||||
std::unique_ptr<ConnectionInfo> connectionInfo;
|
||||
std::string remoteIp;
|
||||
int remotePort;
|
||||
|
||||
if (_addressFamily == AF_INET)
|
||||
{
|
||||
char remoteIp[INET_ADDRSTRLEN];
|
||||
if (inet_ntop(AF_INET, &client.sin_addr, remoteIp, INET_ADDRSTRLEN) == nullptr)
|
||||
char remoteIp4[INET_ADDRSTRLEN];
|
||||
if (inet_ntop(AF_INET, &client.sin_addr, remoteIp4, INET_ADDRSTRLEN) == nullptr)
|
||||
{
|
||||
int err = Socket::getErrno();
|
||||
std::stringstream ss;
|
||||
@ -350,12 +351,13 @@ namespace ix
|
||||
continue;
|
||||
}
|
||||
|
||||
connectionInfo = std::make_unique<ConnectionInfo>(remoteIp, client.sin_port);
|
||||
remotePort = client.sin_port;
|
||||
remoteIp = remoteIp4;
|
||||
}
|
||||
else // AF_INET6
|
||||
{
|
||||
char remoteIp[INET6_ADDRSTRLEN];
|
||||
if (inet_ntop(AF_INET6, &client.sin_addr, remoteIp, INET6_ADDRSTRLEN) == nullptr)
|
||||
char remoteIp6[INET6_ADDRSTRLEN];
|
||||
if (inet_ntop(AF_INET6, &client.sin_addr, remoteIp6, INET6_ADDRSTRLEN) == nullptr)
|
||||
{
|
||||
int err = Socket::getErrno();
|
||||
std::stringstream ss;
|
||||
@ -368,7 +370,8 @@ namespace ix
|
||||
continue;
|
||||
}
|
||||
|
||||
connectionInfo = std::make_unique<ConnectionInfo>(remoteIp, client.sin_port);
|
||||
remotePort = client.sin_port;
|
||||
remoteIp = remoteIp6;
|
||||
}
|
||||
|
||||
std::shared_ptr<ConnectionState> connectionState;
|
||||
@ -377,6 +380,8 @@ namespace ix
|
||||
connectionState = _connectionStateFactory();
|
||||
}
|
||||
connectionState->setOnSetTerminatedCallback([this] { onSetTerminatedCallback(); });
|
||||
connectionState->setRemoteIp(remoteIp);
|
||||
connectionState->setRemotePort(remotePort);
|
||||
|
||||
if (_stop) return;
|
||||
|
||||
@ -404,13 +409,10 @@ namespace ix
|
||||
|
||||
// Launch the handleConnection work asynchronously in its own thread.
|
||||
std::lock_guard<std::mutex> lock(_connectionsThreadsMutex);
|
||||
_connectionsThreads.push_back(
|
||||
std::make_pair(connectionState,
|
||||
std::thread(&SocketServer::handleConnection,
|
||||
this,
|
||||
std::move(socket),
|
||||
connectionState,
|
||||
std::move(connectionInfo))));
|
||||
_connectionsThreads.push_back(std::make_pair(
|
||||
connectionState,
|
||||
std::thread(
|
||||
&SocketServer::handleConnection, this, std::move(socket), connectionState)));
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -6,7 +6,6 @@
|
||||
|
||||
#pragma once
|
||||
|
||||
#include "IXConnectionInfo.h"
|
||||
#include "IXConnectionState.h"
|
||||
#include "IXSelectInterrupt.h"
|
||||
#include "IXSocketTLSOptions.h"
|
||||
@ -105,8 +104,7 @@ namespace ix
|
||||
ConnectionStateFactory _connectionStateFactory;
|
||||
|
||||
virtual void handleConnection(std::unique_ptr<Socket>,
|
||||
std::shared_ptr<ConnectionState> connectionState,
|
||||
std::unique_ptr<ConnectionInfo> connectionInfo) = 0;
|
||||
std::shared_ptr<ConnectionState> connectionState) = 0;
|
||||
virtual size_t getConnectedClientsCount() = 0;
|
||||
|
||||
// Returns true if all connection threads are joined
|
||||
|
@ -56,10 +56,9 @@ namespace ix
|
||||
|
||||
server.setOnConnectionCallback(
|
||||
[remoteUrl, remoteUrlsMapping](std::weak_ptr<ix::WebSocket> webSocket,
|
||||
std::shared_ptr<ConnectionState> connectionState,
|
||||
std::unique_ptr<ConnectionInfo> connectionInfo) {
|
||||
std::shared_ptr<ConnectionState> connectionState) {
|
||||
auto state = std::dynamic_pointer_cast<ProxyConnectionState>(connectionState);
|
||||
auto remoteIp = connectionInfo->remoteIp;
|
||||
auto remoteIp = connectionState->getRemoteIp();
|
||||
|
||||
// Server connection
|
||||
state->webSocket().setOnMessageCallback(
|
||||
|
@ -77,15 +77,14 @@ namespace ix
|
||||
}
|
||||
|
||||
void WebSocketServer::handleConnection(std::unique_ptr<Socket> socket,
|
||||
std::shared_ptr<ConnectionState> connectionState,
|
||||
std::unique_ptr<ConnectionInfo> connectionInfo)
|
||||
std::shared_ptr<ConnectionState> connectionState)
|
||||
{
|
||||
setThreadName("WebSocketServer::" + connectionState->getId());
|
||||
|
||||
auto webSocket = std::make_shared<WebSocket>();
|
||||
if (_onConnectionCallback)
|
||||
{
|
||||
_onConnectionCallback(webSocket, connectionState, std::move(connectionInfo));
|
||||
_onConnectionCallback(webSocket, connectionState);
|
||||
|
||||
if (!webSocket->isOnMessageCallbackRegistered())
|
||||
{
|
||||
@ -99,9 +98,8 @@ namespace ix
|
||||
else if (_onClientMessageCallback)
|
||||
{
|
||||
webSocket->setOnMessageCallback(
|
||||
[this, &ws = *webSocket.get(), connectionState, &ci = *connectionInfo.get()](
|
||||
const WebSocketMessagePtr& msg) {
|
||||
_onClientMessageCallback(connectionState, ci, ws, msg);
|
||||
[this, &ws = *webSocket.get(), connectionState](const WebSocketMessagePtr& msg) {
|
||||
_onClientMessageCallback(connectionState, ws, msg);
|
||||
});
|
||||
}
|
||||
else
|
||||
|
@ -23,14 +23,10 @@ namespace ix
|
||||
{
|
||||
public:
|
||||
using OnConnectionCallback =
|
||||
std::function<void(std::weak_ptr<WebSocket>,
|
||||
std::shared_ptr<ConnectionState>,
|
||||
std::unique_ptr<ConnectionInfo> connectionInfo)>;
|
||||
std::function<void(std::weak_ptr<WebSocket>, std::shared_ptr<ConnectionState>)>;
|
||||
|
||||
using OnClientMessageCallback = std::function<void(std::shared_ptr<ConnectionState>,
|
||||
ConnectionInfo&,
|
||||
WebSocket&,
|
||||
const WebSocketMessagePtr&)>;
|
||||
using OnClientMessageCallback = std::function<void(
|
||||
std::shared_ptr<ConnectionState>, WebSocket&, const WebSocketMessagePtr&)>;
|
||||
|
||||
WebSocketServer(int port = SocketServer::kDefaultPort,
|
||||
const std::string& host = SocketServer::kDefaultHost,
|
||||
@ -69,8 +65,7 @@ namespace ix
|
||||
|
||||
// Methods
|
||||
virtual void handleConnection(std::unique_ptr<Socket> socket,
|
||||
std::shared_ptr<ConnectionState> connectionState,
|
||||
std::unique_ptr<ConnectionInfo> connectionInfo);
|
||||
std::shared_ptr<ConnectionState> connectionState);
|
||||
virtual size_t getConnectedClientsCount() final;
|
||||
};
|
||||
} // namespace ix
|
||||
|
@ -6,4 +6,4 @@
|
||||
|
||||
#pragma once
|
||||
|
||||
#define IX_WEBSOCKET_VERSION "10.3.0"
|
||||
#define IX_WEBSOCKET_VERSION "10.3.1"
|
||||
|
Reference in New Issue
Block a user