Add ability to reconnect from server side (#7226)

* Add ability to reconnect from server side

* Fix code smell

Co-authored-by: Serhii Hadzhilov <serhii@symless.com>
This commit is contained in:
Serhii Hadzhilov 2022-10-03 17:08:24 +03:00 committed by GitHub
parent 82b3696b1b
commit 0f3e128e17
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 104 additions and 59 deletions

View file

@ -51,7 +51,8 @@ void
InverseServerSocket::bind(const NetworkAddress& addr) InverseServerSocket::bind(const NetworkAddress& addr)
{ {
Lock lock(&m_mutex); Lock lock(&m_mutex);
m_socket.connectSocket(addr); m_address = addr;
m_socket.connectSocket(m_address);
setListeningJob(true); setListeningJob(true);
} }
@ -110,8 +111,8 @@ InverseServerSocket::serviceListening(ISocketMultiplexerJob* job,
bool, bool write, bool error) bool, bool write, bool error)
{ {
if (error) { if (error) {
close(); m_socket.connectSocket(m_address);
return nullptr; return job;
} }
if (write) { if (write) {
m_events->addEvent(Event(m_events->forIListenSocket().connecting(), this)); m_events->addEvent(Event(m_events->forIListenSocket().connecting(), this));

View file

@ -58,4 +58,7 @@ protected:
Mutex m_mutex; Mutex m_mutex;
IEventQueue* m_events; IEventQueue* m_events;
SocketMultiplexer* m_socketMultiplexer; SocketMultiplexer* m_socketMultiplexer;
private:
NetworkAddress m_address;
}; };

View file

@ -403,17 +403,20 @@ SecureSocket::freeSSL()
// could cause events to get called on a dead object. TCPSocket // could cause events to get called on a dead object. TCPSocket
// will do this, too, but the double-call is harmless // will do this, too, but the double-call is harmless
setJob(NULL); setJob(NULL);
if (m_ssl->m_ssl != NULL) { if (m_ssl) {
SSL_shutdown(m_ssl->m_ssl); if (m_ssl->m_ssl != NULL) {
SSL_shutdown(m_ssl->m_ssl);
SSL_free(m_ssl->m_ssl); SSL_free(m_ssl->m_ssl);
m_ssl->m_ssl = NULL; m_ssl->m_ssl = NULL;
}
if (m_ssl->m_context != NULL) {
SSL_CTX_free(m_ssl->m_context);
m_ssl->m_context = NULL;
}
delete m_ssl;
m_ssl = nullptr;
} }
if (m_ssl->m_context != NULL) {
SSL_CTX_free(m_ssl->m_context);
m_ssl->m_context = NULL;
}
delete m_ssl;
} }
int int

View file

@ -40,22 +40,13 @@ ClientListener::ClientListener(const NetworkAddress& address,
m_socketFactory(socketFactory), m_socketFactory(socketFactory),
m_server(NULL), m_server(NULL),
m_events(events), m_events(events),
m_useSecureNetwork(enableCrypto) m_useSecureNetwork(enableCrypto),
m_address(address)
{ {
assert(m_socketFactory != NULL); assert(m_socketFactory != NULL);
try { try {
m_listen = m_socketFactory->createListen(m_useSecureNetwork, ARCH->getAddrFamily(address.getAddress())); start();
// setup event handler
m_events->adoptHandler(m_events->forIListenSocket().connecting(),
m_listen,
new TMethodEventJob<ClientListener>(this,
&ClientListener::handleClientConnecting));
// bind listen address
LOG((CLOG_DEBUG1 "binding listen socket"));
m_listen->bind(address);
} }
catch (XSocketAddressInUse&) { catch (XSocketAddressInUse&) {
cleanupListenSocket(); cleanupListenSocket();
@ -72,31 +63,7 @@ ClientListener::ClientListener(const NetworkAddress& address,
ClientListener::~ClientListener() ClientListener::~ClientListener()
{ {
LOG((CLOG_DEBUG1 "stop listening for clients")); stop();
// discard already connected clients
for (NewClients::iterator index = m_newClients.begin();
index != m_newClients.end(); ++index) {
ClientProxyUnknown* client = *index;
m_events->removeHandler(
m_events->forClientProxyUnknown().success(), client);
m_events->removeHandler(
m_events->forClientProxyUnknown().failure(), client);
m_events->removeHandler(
m_events->forClientProxy().disconnected(), client);
delete client;
}
// discard waiting clients
ClientProxy* client = getNextClient();
while (client != NULL) {
delete client;
client = getNextClient();
}
m_events->removeHandler(m_events->forIListenSocket().connecting(), m_listen);
cleanupListenSocket();
cleanupClientSockets();
delete m_socketFactory; delete m_socketFactory;
} }
@ -119,6 +86,56 @@ ClientListener::getNextClient()
return client; return client;
} }
void ClientListener::start()
{
m_listen = m_socketFactory->createListen(m_useSecureNetwork, ARCH->getAddrFamily(m_address.getAddress()));
// setup event handler
m_events->adoptHandler(m_events->forIListenSocket().connecting(),
m_listen,
new TMethodEventJob<ClientListener>(this,
&ClientListener::handleClientConnecting));
// bind listen address
LOG((CLOG_DEBUG1 "binding listen socket"));
m_listen->bind(m_address);
}
void ClientListener::stop()
{
LOG((CLOG_DEBUG1 "stop listening for clients"));
// discard already connected clients
for (NewClients::iterator index = m_newClients.begin();
index != m_newClients.end(); ++index) {
ClientProxyUnknown* client = *index;
m_events->removeHandler(
m_events->forClientProxyUnknown().success(), client);
m_events->removeHandler(
m_events->forClientProxyUnknown().failure(), client);
m_events->removeHandler(
m_events->forClientProxy().disconnected(), client);
delete client;
}
// discard waiting clients
ClientProxy* client = getNextClient();
while (client != nullptr) {
delete client;
client = getNextClient();
}
m_events->removeHandler(m_events->forIListenSocket().connecting(), m_listen);
cleanupListenSocket();
cleanupClientSockets();
}
void ClientListener::restart()
{
stop();
start();
}
void void
ClientListener::handleClientConnecting(const Event&, void*) ClientListener::handleClientConnecting(const Event&, void*)
{ {

View file

@ -66,7 +66,10 @@ public:
ClientProxy* getNextClient(); ClientProxy* getNextClient();
//! Get server which owns this listener //! Get server which owns this listener
Server* getServer() { return m_server; } Server* getServer() { return m_server; }
//! This method restarts the listener
void restart();
//@} //@}
@ -79,6 +82,8 @@ private:
void cleanupListenSocket(); void cleanupListenSocket();
void cleanupClientSockets(); void cleanupClientSockets();
void start();
void stop();
private: private:
typedef std::set<ClientProxyUnknown*> NewClients; typedef std::set<ClientProxyUnknown*> NewClients;
@ -93,4 +98,5 @@ private:
IEventQueue* m_events; IEventQueue* m_events;
bool m_useSecureNetwork; bool m_useSecureNetwork;
ClientSockets m_clientSockets; ClientSockets m_clientSockets;
NetworkAddress m_address;
}; };

View file

@ -658,6 +658,11 @@ String Config::getClientAddress() const
return m_clientAddress; return m_clientAddress;
} }
bool Config::isClientMode() const
{
return (!m_clientAddress.empty());
}
void void
Config::readSection(ConfigReadContext& s) Config::readSection(ConfigReadContext& s)
{ {

View file

@ -455,6 +455,13 @@ public:
* Return client address a string. * Return client address a string.
*/ */
String getClientAddress() const; String getClientAddress() const;
//! Return true if server started in client mode
/*!
* In client mode the server initiates connection to client
*/
bool isClientMode() const;
//@} //@}
private: private:

View file

@ -1432,6 +1432,10 @@ Server::handleClientDisconnected(const Event&, void* vclient)
removeOldClient(client); removeOldClient(client);
delete client; delete client;
if (m_args.m_config->isClientMode()) {
m_clientListener->restart();
}
} }
void void

View file

@ -682,13 +682,12 @@ ServerApp::handleScreenSwitched(const Event& e, void*)
ISocketFactory* ServerApp::getSocketFactory() const ISocketFactory* ServerApp::getSocketFactory() const
{ {
ISocketFactory* socketFactory = nullptr; ISocketFactory* socketFactory = nullptr;
const auto clientAddress = args().m_config->getClientAddress();
if (clientAddress.empty()) { if (args().m_config->isClientMode()) {
socketFactory = new TCPSocketFactory(m_events, getSocketMultiplexer()); socketFactory = new InverseSocketFactory(m_events, getSocketMultiplexer());
} }
else { else {
socketFactory = new InverseSocketFactory(m_events, getSocketMultiplexer()); socketFactory = new TCPSocketFactory(m_events, getSocketMultiplexer());
} }
return socketFactory; return socketFactory;
@ -696,15 +695,15 @@ ISocketFactory* ServerApp::getSocketFactory() const
NetworkAddress ServerApp::getAddress(const NetworkAddress& address) const NetworkAddress ServerApp::getAddress(const NetworkAddress& address) const
{ {
const auto clientAddress = args().m_config->getClientAddress(); if (args().m_config->isClientMode()) {
if (clientAddress.empty()) { const auto clientAddress = args().m_config->getClientAddress();
return address;
}
else {
NetworkAddress addr(clientAddress.c_str(), kDefaultPort); NetworkAddress addr(clientAddress.c_str(), kDefaultPort);
addr.resolve(); addr.resolve();
return addr; return addr;
} }
else {
return address;
}
} }
int int