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

SocketServer::handleConnection takes an std::shared_ptr<Socket> instead of a file descriptor

Browse Source
This commit is contained in:
Benjamin Sergeant
2019-09-30 21:48:55 -07:00
parent 562d7484e4
commit 1ed39677ce
14 changed files with 47 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
ixsnake/ixsnake/IXRedisServer.cpp
View File

@@ -8,7 +8,6 @@
#include <ixwebsocket/IXNetSystem.h>
#include <ixwebsocket/IXSocketConnect.h>
#include <ixwebsocket/IXSocketFactory.h>
#include <ixwebsocket/IXSocket.h>
#include <ixwebsocket/IXCancellationRequest.h>
#include <fstream>
@@ -45,16 +44,11 @@ namespace ix
SocketServer::stop();
}
void RedisServer::handleConnection(int fd, std::shared_ptr<ConnectionState> connectionState)
void RedisServer::handleConnection(std::shared_ptr<Socket> socket,
std::shared_ptr<ConnectionState> connectionState)
{
_connectedClientsCount++;
std::string errorMsg;
auto socket = createSocket(fd, errorMsg);
// Set the socket to non blocking mode + other tweaks
SocketConnect::configure(fd);
while (!_stopHandlingConnections)
{
std::vector<std::string> tokens;
@@ -105,7 +99,6 @@ namespace ix
logInfo("Connection closed for connection id " + connectionState->getId());
connectionState->setTerminated();
Socket::closeSocket(fd);
_connectedClientsCount--;
}

2
ixsnake/ixsnake/IXRedisServer.h
View File

@@ -42,7 +42,7 @@ namespace ix
std::atomic<bool> _stopHandlingConnections;
// Methods
virtual void handleConnection(int fd,
virtual void handleConnection(std::shared_ptr<Socket>,
std::shared_ptr<ConnectionState> connectionState) final;
virtual size_t getConnectedClientsCount() final;

11
ixwebsocket/IXHttpServer.cpp
View File

@@ -8,7 +8,6 @@
#include "IXNetSystem.h"
#include "IXSocketConnect.h"
#include "IXSocketFactory.h"
#include "IXUserAgent.h"
#include <fstream>
#include <iostream>
@@ -70,16 +69,11 @@ namespace ix
_onConnectionCallback = callback;
}
void HttpServer::handleConnection(int fd, std::shared_ptr<ConnectionState> connectionState)
void HttpServer::handleConnection(std::shared_ptr<Socket> socket,
std::shared_ptr<ConnectionState> connectionState)
{
_connectedClientsCount++;
std::string errorMsg;
auto socket = createSocket(fd, errorMsg);
// Set the socket to non blocking mode + other tweaks
SocketConnect::configure(fd);
auto ret = Http::parseRequest(socket);
// FIXME: handle errors in parseRequest
@@ -92,7 +86,6 @@ namespace ix
}
}
connectionState->setTerminated();
Socket::closeSocket(fd);
_connectedClientsCount--;
}

2
ixwebsocket/IXHttpServer.h
View File

@@ -42,7 +42,7 @@ namespace ix
std::atomic<int> _connectedClientsCount;
// Methods
virtual void handleConnection(int fd,
virtual void handleConnection(std::shared_ptr<Socket>,
std::shared_ptr<ConnectionState> connectionState) final;
virtual size_t getConnectedClientsCount() final;

17
ixwebsocket/IXSocketServer.cpp
View File

@@ -9,6 +9,7 @@
#include "IXNetSystem.h"
#include "IXSocket.h"
#include "IXSocketConnect.h"
#include "IXSocketFactory.h"
#include <assert.h>
#include <iostream>
#include <sstream>
@@ -267,11 +268,25 @@ namespace ix
if (_stop) return;
// create socket
std::string errorMsg;
auto socket = createSocket(clientFd, errorMsg);
if (socket == nullptr)
{
logError("SocketServer::run() cannot create socket: " + errorMsg);
Socket::closeSocket(clientFd);
continue;
}
// Set the socket to non blocking mode + other tweaks
SocketConnect::configure(clientFd);
// 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, clientFd, connectionState)));
std::thread(&SocketServer::handleConnection, this, socket, connectionState)));
}
}

5
ixwebsocket/IXSocketServer.h
View File

@@ -21,6 +21,8 @@
namespace ix
{
class Socket;
class SocketServer
{
public:
@@ -96,7 +98,8 @@ namespace ix
// the factory to create ConnectionState objects
ConnectionStateFactory _connectionStateFactory;
virtual void handleConnection(int fd, std::shared_ptr<ConnectionState> connectionState) = 0;
virtual void handleConnection(std::shared_ptr<Socket>,
std::shared_ptr<ConnectionState> connectionState) = 0;
virtual size_t getConnectedClientsCount() = 0;
// Returns true if all connection threads are joined

5
ixwebsocket/IXWebSocket.cpp
View File

@@ -201,7 +201,8 @@ namespace ix
return status;
}
WebSocketInitResult WebSocket::connectToSocket(int fd, int timeoutSecs)
WebSocketInitResult WebSocket::connectToSocket(std::shared_ptr<Socket> socket,
int timeoutSecs)
{
{
std::lock_guard<std::mutex> lock(_configMutex);
@@ -212,7 +213,7 @@ namespace ix
_pingTimeoutSecs);
}
WebSocketInitResult status = _ws.connectToSocket(fd, timeoutSecs);
WebSocketInitResult status = _ws.connectToSocket(socket, timeoutSecs);
if (!status.success)
{
return status;

3
ixwebsocket/IXWebSocket.h
View File

@@ -113,7 +113,8 @@ namespace ix
static void invokeTrafficTrackerCallback(size_t size, bool incoming);
// Server
WebSocketInitResult connectToSocket(int fd, int timeoutSecs);
WebSocketInitResult connectToSocket(std::shared_ptr<Socket>,
int timeoutSecs);
WebSocketTransport _ws;

9
ixwebsocket/IXWebSocketHandshake.cpp
View File

@@ -239,18 +239,13 @@ namespace ix
return WebSocketInitResult(true, status, "", headers, path);
}
WebSocketInitResult WebSocketHandshake::serverHandshake(int fd, int timeoutSecs)
WebSocketInitResult WebSocketHandshake::serverHandshake(int timeoutSecs)
{
_requestInitCancellation = false;
// Set the socket to non blocking mode + other tweaks
SocketConnect::configure(fd);
auto isCancellationRequested =
makeCancellationRequestWithTimeout(timeoutSecs, _requestInitCancellation);
std::string remote = std::string("remote fd ") + std::to_string(fd);
// Read first line
auto lineResult = _socket->readLine(isCancellationRequested);
auto lineValid = lineResult.first;
@@ -358,7 +353,7 @@ namespace ix
if (!_socket->writeBytes(ss.str(), isCancellationRequested))
{
return WebSocketInitResult(
false, 0, std::string("Failed sending response to ") + remote);
false, 0, std::string("Failed sending response to remote end"));
}
return WebSocketInitResult(true, 200, "", headers, uri);

2
ixwebsocket/IXWebSocketHandshake.h
View File

@@ -56,7 +56,7 @@ namespace ix
int port,
int timeoutSecs);
WebSocketInitResult serverHandshake(int fd, int timeoutSecs);
WebSocketInitResult serverHandshake(int timeoutSecs);
private:
std::string genRandomString(const int len);

7
ixwebsocket/IXWebSocketServer.cpp
View File

@@ -63,7 +63,8 @@ namespace ix
_onConnectionCallback = callback;
}
void WebSocketServer::handleConnection(int fd, std::shared_ptr<ConnectionState> connectionState)
void WebSocketServer::handleConnection(std::shared_ptr<Socket> socket,
std::shared_ptr<ConnectionState> connectionState)
{
auto webSocket = std::make_shared<WebSocket>();
_onConnectionCallback(webSocket, connectionState);
@@ -81,7 +82,7 @@ namespace ix
_clients.insert(webSocket);
}
auto status = webSocket->connectToSocket(fd, _handshakeTimeoutSecs);
auto status = webSocket->connectToSocket(socket, _handshakeTimeoutSecs);
if (status.success)
{
// Process incoming messages and execute callbacks
@@ -107,8 +108,6 @@ namespace ix
logInfo("WebSocketServer::handleConnection() done");
connectionState->setTerminated();
Socket::closeSocket(fd);
}
std::set<std::shared_ptr<WebSocket>> WebSocketServer::getClients()

2
ixwebsocket/IXWebSocketServer.h
View File

@@ -55,7 +55,7 @@ namespace ix
const static bool kDefaultEnablePong;
// Methods
virtual void handleConnection(int fd,
virtual void handleConnection(std::shared_ptr<Socket> socket,
std::shared_ptr<ConnectionState> connectionState) final;
virtual size_t getConnectedClientsCount() final;
};

13
ixwebsocket/IXWebSocketTransport.cpp
View File

@@ -171,20 +171,15 @@ namespace ix
}
// Server
WebSocketInitResult WebSocketTransport::connectToSocket(int fd, int timeoutSecs)
WebSocketInitResult WebSocketTransport::connectToSocket(std::shared_ptr<Socket> socket,
int timeoutSecs)
{
std::lock_guard<std::mutex> lock(_socketMutex);
// Server should not mask the data it sends to the client
_useMask = false;
std::string errorMsg;
_socket = createSocket(fd, errorMsg);
if (!_socket)
{
return WebSocketInitResult(false, 0, errorMsg);
}
_socket = socket;
WebSocketHandshake webSocketHandshake(_requestInitCancellation,
_socket,
@@ -192,7 +187,7 @@ namespace ix
_perMessageDeflateOptions,
_enablePerMessageDeflate);
auto result = webSocketHandshake.serverHandshake(fd, timeoutSecs);
auto result = webSocketHandshake.serverHandshake(timeoutSecs);
if (result.success)
{
setReadyState(ReadyState::OPEN);

7
ixwebsocket/IXWebSocketTransport.h
View File

@@ -77,11 +77,14 @@ namespace ix
int pingIntervalSecs,
int pingTimeoutSecs);
WebSocketInitResult connectToUrl( // Client
// Client
WebSocketInitResult connectToUrl(
const std::string& url,
const WebSocketHttpHeaders& headers,
int timeoutSecs);
WebSocketInitResult connectToSocket(int fd, // Server
// Server
WebSocketInitResult connectToSocket(std::shared_ptr<Socket> socket,
int timeoutSecs);
PollResult poll();