Egalware/EgtExecutor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

EgtExecutor :

Browse Source 
- in redis migliorato ciclo degli eventi e aggiunte funzioni per collegamento asincrono per nodo sentinalla e autenticazione (test).
This commit is contained in:
Riccardo Elitropi
2025-09-30 11:29:39 +02:00
parent 55aca40285
commit b1f606e821
1 changed files with 193 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
EXE_Redis.cpp
+193 -98
View File
@@ -28,6 +28,7 @@
#include <atomic>
#include <map>
#include <sstream>
#include <future>
using namespace std ;
@@ -42,7 +43,8 @@ static redisContext* s_pRedisContext = nullptr ;
// Contesto Asincrono
static redisAsyncContext* s_pRedisAsyncSubContext = nullptr ;
static redisAsyncContext* s_pRedisAsyncPubContext = nullptr ;
static atomic<int> s_nPendingDataBase = REDIS_MIN_DB ; // Default
static atomic<int> s_nPendingDataBase = REDIS_MIN_DB ; // Default
static string s_sPassword = "" ; // Default
static atomic<bool> s_bSubConnected = false ;
static atomic<bool> s_bPubConnected = false ;
static atomic<bool> s_bPubLoopRunning = false ;
@@ -399,30 +401,33 @@ RedisEventLoop( redisAsyncContext* ctx, bool bIsPub)
// Recupero del file descriptor del socket TCP usato da Redis.
SOCKET sock = ctx->c.fd ;
// Imposto condizione di ascolto degli eventi
bool bLoopRunning = bIsPub ? s_bPubLoopRunning : s_bSubLoopRunning ;
// Creazione dei due insiemi per il file descriptor ( lettura e scrittura )
// r = read, w = write
fd_set rfds, wfds ;
// Timeout per funzione select { secondi, millisecondi}
// La funzione select attende che il socket sia pronto per lettura e scrittura
// Resituisce :
// SOCKET_ERROR in caso di errore
// 0 se scade il timeout
// > 0 se ci sono eventi da gestire
timeval timeout = {0, 100000 } ; // 100ms
// Ciclo
while ( bIsPub ? s_bPubLoopRunning : s_bSubLoopRunning) {
while ( bLoopRunning) {
// Se contesto non valido, interrompo il ciclo
if ( ctx == nullptr || ( ctx->c.flags & REDIS_DISCONNECTING)) {
LOG_INFO ( GetCmdLogger(), ( string{ "Redis event loop terminated for Async "} +
( bIsPub ? "Pub" : "Sub") + " Events").c_str())
break ;
}
// Creazione dei due insiemi per il file descriptor ( lettura e scrittura )
// r = read, w = write
fd_set rfds, wfds ;
// Recupero file descriptor di lettura e scrittura
FD_ZERO( &rfds) ;
FD_ZERO( &wfds) ;
FD_SET( sock, &rfds) ;
FD_SET( sock, &wfds) ;
// TimeOut per funzione select { secondi, millisecondi}
// La funzione select attende che il socket sia pronto per lettura e scrittura
// Resituisce :
// SOCKET_ERROR in caso di errore
// 0 se scade il timeout
// > 0 se ci sono eventi da gestire
timeval timeout = {0, 5000} ;
// Controllo il risultato ottenuto
int nRet = select( 0, &rfds, &wfds, nullptr, &timeout) ;
if ( nRet == SOCKET_ERROR) {
// Se errore -> interruzione del ciclo
@@ -430,19 +435,19 @@ RedisEventLoop( redisAsyncContext* ctx, bool bIsPub)
ToString( WSAGetLastError())).c_str()) ;
break ;
}
else if ( nRet == 0) {
// Inserisco una breve pausa per non saturare le CPU
this_thread::sleep_for( chrono::milliseconds( 10)) ;
else if ( nRet > 0) {
// Se Socket pronto per lettura/ scrittua
if ( ctx != nullptr && ! ( ctx->c.flags & REDIS_DISCONNECTING) && ctx->c.fd != -1) {
if ( FD_ISSET( sock, &rfds))
redisAsyncHandleRead( ctx) ;
if ( FD_ISSET( sock, &wfds))
redisAsyncHandleWrite( ctx) ;
}
}
if ( nRet > 0) {
// Se Socket pronto per lettura, leggo il messaggio dal contesto di Redis
if ( FD_ISSET( sock, &rfds) && ctx != nullptr)
redisAsyncHandleRead( ctx) ;
// Se Socket pronto per scrittura, scrivo il messaggio per il contesto di Redis
if ( FD_ISSET( sock, &wfds) && ctx != nullptr)
redisAsyncHandleWrite( ctx) ;
}
this_thread::sleep_for( chrono::milliseconds( 10)) ;
// Reset del timeOut ad ogni ciclo
timeout.tv_sec = 0 ;
timeout.tv_usec = 100000 ; // 100ms
}
return ;
@@ -609,7 +614,7 @@ WaitMessageCallback( redisAsyncContext* ctx, void* r, void*)
}
//----------------------------------------------------------------------------
// Funzione per Memorizzare il DataBase Redis da selezionare
// Funzione per Memorizzare il DataBase Redis da selezionare [Callback]
//----------------------------------------------------------------------------
static bool
SetPendingDataBase( int nDB)
@@ -623,7 +628,7 @@ SetPendingDataBase( int nDB)
}
//----------------------------------------------------------------------------
// Funzione per Ricavare il numero di DataBase Redis da selezionare
// Funzione per Ricavare il numero di DataBase Redis da selezionare [Callback]
//----------------------------------------------------------------------------
static int
GetPendingDataBase()
@@ -631,6 +636,24 @@ GetPendingDataBase()
return s_nPendingDataBase ;
}
// ---------------------------------------------------------------------------
// Funzione per Memorizzare la Password di autenticazione [Callback]
// ---------------------------------------------------------------------------
static bool
SetPendingPassword( string sPsw)
{
s_sPassword = sPsw ;
}
// --------------------------------------------------------------------------
// Funzione per ottenere la Password di autenticazione [Callback]
// --------------------------------------------------------------------------
static string
GetPendingPassword()
{
return s_sPassword ;
}
//----------------------------------------------------------------------------
static bool
GetConnectionContext( redisAsyncContext*& ctx, const string& sHost, int nPort)
@@ -700,6 +723,33 @@ ExeRedisAsyncConnect( const string& sConnection)
return false ;
}
// Verifico se si tratta di un nodo Sentinella
redisContext* sentinelCtx = redisConnect( ConnectionInfo.sHost.c_str(), ConnectionInfo.nPort) ;
if ( sentinelCtx == nullptr) {
LOG_INFO( GetCmdLogger(), "Error : Can't allocate redis context")
return false ;
}
else if ( sentinelCtx->err != 0) {
LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + sentinelCtx->errstr).c_str())
redisFree( sentinelCtx) ;
return false ;
}
redisReply* replySentinel = ( redisReply*)redisCommand( sentinelCtx, "SENTINEL get-master-addr-by-name %s",
ConnectionInfo.sServiceName.c_str()) ;
if ( replySentinel == nullptr || replySentinel->type != REDIS_REPLY_ARRAY || replySentinel->elements != 2) {
if ( replySentinel != nullptr)
freeReplyObject( replySentinel) ;
}
else if ( replySentinel->element[0] != nullptr && replySentinel->element[0]->str != nullptr &&
replySentinel->element[1] != nullptr && replySentinel->element[1]->str != nullptr) {
// --- Nodo sentinella
ConnectionInfo.sHost = replySentinel->element[0]->str ;
ConnectionInfo.nPort = DEFAULT_SENTINEL_MASTER_PORT ;
FromString( replySentinel->element[1]->str, ConnectionInfo.nPort) ;
freeReplyObject( replySentinel) ;
redisFree( sentinelCtx) ;
}
// Definisco le 2 connessioni asincroni con Host e Porta ( per PUBLISH e SUBSCRIBE/UNSUBSCRIBE)
if ( ! GetConnectionContext( s_pRedisAsyncPubContext, ConnectionInfo.sHost, ConnectionInfo.nPort) ||
! GetConnectionContext( s_pRedisAsyncSubContext, ConnectionInfo.sHost, ConnectionInfo.nPort))
@@ -711,49 +761,57 @@ ExeRedisAsyncConnect( const string& sConnection)
s_pRedisAsyncPubContext,
[]( const redisAsyncContext* ctx, int nStatus) {
// --- Se contesto nullo, errore
if ( ctx == nullptr) {
LOG_INFO( GetCmdLogger(), "Error : Redis null Context in CallBack Async Pub Connection") ;
return ;
}
// --- Se Connessione non valida, errore
if ( nStatus != REDIS_OK) {
if ( ctx == nullptr || nStatus != REDIS_OK) {
LOG_INFO( GetCmdLogger(), "Error : Async Connect CallBack Pub failed ") ;
return ;
}
// --- Se Connessione Ok
else {
// --- Selezione del DataBase
// --- Se Autenticazione richiesta -> AUTH e SELECT
string sPassword = GetPendingPassword() ;
if ( ! sPassword.empty()) {
redisAsyncCommand(
s_pRedisAsyncPubContext,
[]( redisAsyncContext* ctx, void* r, void*) {
// --- Se contesto nullo, errore
if ( ctx == nullptr) {
LOG_INFO( GetCmdLogger(), "Error : Redis null Context in CallBack Async Pub Connection") ;
return ;
}
// --- Richiesta di Selezione DB a Redis
// Invio la richiesta
redisReply* reply = static_cast<redisReply*>( r) ;
if ( reply == nullptr) {
// --- Nulla
LOG_INFO( GetCmdLogger(), "Error : Null Database selection reply") ;
if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) {
string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null asnwer" ;
LOG_INFO( GetCmdLogger(), ( string{ " Error : Authentication -> "} + sErrMsg).c_str())
return ;
}
if ( reply->type == REDIS_REPLY_STATUS && strcmp( reply->str, "OK") == 0) {
// --- Valida
LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Pub commands !").c_str())
s_bPubConnected = true ;
return ;
}
else {
// --- Errore
string sErrMsg = ( reply->str != nullptr ? reply->str : "Null asnwer") ;
LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str())
return ;
}
LOG_INFO( GetCmdLogger(), "Valid Authentication for Pub connection")
redisAsyncCommand(
ctx,
[]( redisAsyncContext* ctx, void* r, void*) {
redisReply* reply = static_cast<redisReply*>( r) ;
if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) {
string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null answer" ;
LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str()) ;
return ;
}
LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Pub commands !").c_str()) ;
s_bPubConnected = true ;
},
nullptr, "SELECT %d", GetPendingDataBase()
) ;
},
nullptr,
"SELECT %d",
GetPendingDataBase()
nullptr, "AUTH %s", GetPendingPassword().c_str()
) ;
}
// Se nessuna autenticazione -> SELECT
else {
redisAsyncCommand(
s_pRedisAsyncPubContext,
[]( redisAsyncContext* ctx, void* r, void*) {
redisReply* reply = static_cast<redisReply*>( r) ;
if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) {
string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null answer" ;
LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str()) ;
return ;
}
LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Pub commands !").c_str())
s_bPubConnected = true ;
},
nullptr, "SELECT %d", GetPendingDataBase()
) ;
}
}
@@ -764,49 +822,82 @@ ExeRedisAsyncConnect( const string& sConnection)
s_pRedisAsyncSubContext,
[]( const redisAsyncContext* ctx, int nStatus) {
// --- Se contesto nullo, errore
if ( ctx == nullptr) {
LOG_INFO( GetCmdLogger(), "Error : Redis null Context in CallBack Async Sub Connection") ;
if ( ctx == nullptr || nStatus != REDIS_OK) {
LOG_INFO( GetCmdLogger(), "Error : Async Connect CallBack Pub failed ") ;
return ;
}
// --- Se Connessione non valida, errore
if ( nStatus != REDIS_OK) {
LOG_INFO( GetCmdLogger(), "Error : Async Connect CallBack Sub failed ") ;
return ;
}
// --- Se Connessione Ok
else {
// --- Selezione del DataBase
// --- Se Autenticazione richiesta -> AUTH e SELECT
string sPassword = GetPendingPassword() ;
if ( ! sPassword.empty()) {
redisAsyncCommand(
s_pRedisAsyncSubContext,
[]( redisAsyncContext* ctx, void* r, void*) {
// --- Se contesto nullo, errore
if ( ctx == nullptr) {
LOG_INFO( GetCmdLogger(), "Error : Redis null Context in CallBack Async Sub Connection") ;
// Invio la richiesta
redisReply* reply = static_cast<redisReply*>( r) ;
if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) {
string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null asnwer" ;
LOG_INFO( GetCmdLogger(), ( string{ " Error : Authentication -> "} + sErrMsg).c_str())
return ;
}
// --- Richiesta di Selezione DB a Redis
LOG_INFO( GetCmdLogger(), "Valid Authentication for Pub connection")
redisAsyncCommand(
ctx,
[]( redisAsyncContext* ctx, void* r, void*) {
redisReply* reply = static_cast<redisReply*>( r) ;
if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) {
string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null answer" ;
LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str()) ;
return ;
}
LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Pub commands !").c_str()) ;
s_bSubConnected = true ;
},
nullptr, "SELECT %d", GetPendingDataBase()
) ;
},
nullptr, "SELECT %d", GetPendingDataBase()
) ;
// --- Autenticazione richiesta
redisAsyncCommand(
s_pRedisAsyncSubContext,
[]( redisAsyncContext* ctx, void* r, void*) {
redisReply* reply = static_cast<redisReply*>( r) ;
if ( reply == nullptr) {
// --- Nulla
LOG_INFO( GetCmdLogger(), "Error : Null Database selection reply") ;
LOG_INFO( GetCmdLogger(), "Error : Null Authentication reply") ;
return ;
}
if ( reply->type == REDIS_REPLY_STATUS && strcmp( reply->str, "OK") == 0) {
// --- Valida
LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Sub commands !").c_str())
s_bSubConnected = true ;
LOG_INFO( GetCmdLogger(), "Valid Authentication for Pub connection")
s_bPubConnected = true ;
return ;
}
else {
// --- Errore
string sErrMsg = ( reply->str != nullptr) ? reply->str : "Null asnwer" ;
LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str())
string sErrMsg = ( reply->str != nullptr ? reply->str : "Null asnwer") ;
LOG_INFO( GetCmdLogger(), ( string{ "Error : Authentication -> "} + sErrMsg).c_str())
return ;
}
},
nullptr,
"SELECT %d",
GetPendingDataBase()
nullptr, "AUTH %s", GetPendingPassword().c_str()
) ;
}
// Se nessuna autenticazione -> SELECT
else {
redisAsyncCommand(
s_pRedisAsyncSubContext,
[]( redisAsyncContext* ctx, void* r, void*) {
redisReply* reply = static_cast<redisReply*>( r) ;
if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) {
string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null answer" ;
LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str()) ;
return ;
}
LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Pub commands !").c_str())
s_bSubConnected = true ;
},
nullptr, "SELECT %d", GetPendingDataBase()
) ;
}
}
@@ -845,18 +936,19 @@ ExeRedisAsyncConnect( const string& sConnection)
// Se richiesto massimo tempo di connessione, aspetto
if ( ConnectionInfo.nAsyncTimeout > 0.) {
const int nStepMillSec = 2 ;
int nWaitedMillSec = 0 ;
while ( ! s_bSubConnected && ! s_bPubConnected && nWaitedMillSec < static_cast<int>( ConnectionInfo.nAsyncTimeout)) {
this_thread::sleep_for( chrono::milliseconds( nStepMillSec)) ;
nWaitedMillSec += nStepMillSec ;
auto tStart = chrono::steady_clock::now() ;
auto tTimeOut = chrono::milliseconds( static_cast<int>( ConnectionInfo.nAsyncTimeout)) ;
while ( ! s_bSubConnected && ! s_bPubConnected) {
this_thread::sleep_for( chrono::milliseconds( 10)) ;
if ( chrono::steady_clock::now() - tStart > tTimeOut) {
LOG_INFO( GetCmdLogger(), "Error in Connection : TimeOut exceeded")
return false ;
}
}
if ( ! s_bSubConnected || ! s_bPubConnected) {
LOG_INFO( GetCmdLogger(), "Error in Connection : TimeOut exceeded")
return false ;
}
else
LOG_INFO( GetCmdLogger(), ( string{ "Connected in "} + ToString( nWaitedMillSec) + " ms").c_str())
auto tElapsed = chrono::duration_cast<chrono::milliseconds>( chrono::steady_clock::now() - tStart).count() ;
LOG_INFO( GetCmdLogger(), ( string{ "Connected in "} + ToString( tElapsed) + " ms").c_str())
}
return true ;
@@ -993,11 +1085,14 @@ ExeRedisAsyncSubscribeOneMessage( const string& sChannel, double dMaxTimeOut, st
}
// Attivo il TimeOut per l'attesa del messaggio
const int nStepMillSeconds = 2 ;
int nWaitedMillSeconds = 0 ;
while ( ! s_bMessage && nWaitedMillSeconds < static_cast<int>( dMaxTimeOut)) {
this_thread::sleep_for( chrono::milliseconds( nStepMillSeconds)) ;
nWaitedMillSeconds += nStepMillSeconds ;
auto tStart = chrono::steady_clock::now() ;
auto tTimeOut = chrono::milliseconds( static_cast<int>( dMaxTimeOut)) ;
while ( ! s_bMessage) {
this_thread::sleep_for( chrono::milliseconds( 10)) ;
if ( chrono::steady_clock::now() - tStart > tTimeOut) {
LOG_INFO( GetCmdLogger(), ( string{ "Timeout : No Message received on ["} + sChannel + "]").c_str())
return false ;
}
}
// Invio del comando Unsubscribe