//---------------------------------------------------------------------------- // EgalTech 2025-2025 //---------------------------------------------------------------------------- // File : EXE_Redis.cpp Data : 17.09.25 Versione : 2.7i3 // Contenuto : Funzioni per interfacciarsi con server Redis. // // // // Modifiche : 17.09.25 RE Creazione modulo. // // //---------------------------------------------------------------------------- //--------------------------- Include ---------------------------------------- #include #include #include "stdafx.h" #include "EXE.h" #include "EXE_Macro.h" #include "AuxTools.h" #include "/EgtDev/Include/EXeExecutor.h" #include "/EgtDev/Include/EGkStringUtils3d.h" #include "/EgtDev/Include/EGnStringKeyVal.h" #pragma warning( disable: 4244) // conversione da 'uint64_t' a 'size_t' #pragma warning( disable: 4200) // utilizzata estensione non standard: matrice di dimensioni zero in struct/union #include "/EgtDev/Extern/hiredis/Include/async.h" #include #include #include #include using namespace std ; // --------------------------------------------------------------------------- // Definizione variabili generali Redis // --------------------------------------------------------------------------- static int REDIS_MIN_DB = 0 ; static int REDIS_MAX_DB = 15 ; // Contesto Sincrono static redisContext* s_pRedisContext = nullptr ; // Contesto Asincrono static redisAsyncContext* s_pRedisAsyncSubContext = nullptr ; static redisAsyncContext* s_pRedisAsyncPubContext = nullptr ; static atomic s_nPendingDataBase = REDIS_MIN_DB ; // Default static atomic s_bSubConnected = false ; static atomic s_bPubConnected = false ; static atomic s_bPubLoopRunning = false ; static atomic s_bSubLoopRunning = false ; static atomic s_bMessage = false ; static string s_sMessage ; // Tipo di valore recuperabile mediante una chiave di tipo string static const string KEY_TYPE_STRING = "string" ; static const string KEY_TYPE_LIST = "list" ; // non usato static const string KEY_TYPE_SET = "set" ; // non usato static const string KEY_TYPE_HASH = "hash" ; // non usato static const string KEY_TYPE_ZSET = "zset" ; // non usato static const string KEY_TYPE_JSON = "ReJSON-RL" ; // non usato, disponibile se versione >= 8 ( noi usiamo la 6) // struttura per parametri della stringa di Connessione static const int DEFAULT_PORT = 6379 ; static const int DEFAULT_SENTINEL_MASTER_PORT = 26379 ; static const int DEFAULT_DATABASE = 0 ; static const int DEFAULT_TIMEOUT = 15000 ; // [ms] static const int DEFAULT_ALIVE_TIME = 180 ; // [s] struct RedisConnectionInfo { string sHost, sServiceName, sPassword ; int nPort, nDefaultDataBase, nKeepAlive, nConnectTimeout, nSyncTimeout, nAsyncTimeout ; bool bAbortConnect, bSsl, bAllowAdmin ; RedisConnectionInfo() { sHost = "" ; sServiceName = "" ; sPassword = "" ; nPort = DEFAULT_PORT ; nDefaultDataBase = DEFAULT_DATABASE ; nKeepAlive = DEFAULT_ALIVE_TIME ; nConnectTimeout = DEFAULT_TIMEOUT ; nSyncTimeout = DEFAULT_TIMEOUT ; nAsyncTimeout = DEFAULT_TIMEOUT ; bAbortConnect = false ; bSsl = false ; bAllowAdmin = false ; } ; RedisConnectionInfo( string sMyHost, string sMyServiceName, string sMyPassword, int nMyPort, int nMyDefaultDataBase, int dMyKeepAlive, int dMyConnectTimeout, int dMySyncTimeout, int dMyAsyncTimeout, bool bMyAbortConnect, bool bMySsl, bool bMyAllowAdmin) : sHost( sMyHost), sServiceName( sMyServiceName), sPassword( sMyPassword), nPort( nMyPort), nDefaultDataBase( nMyDefaultDataBase), nKeepAlive( dMyKeepAlive), nConnectTimeout( dMyConnectTimeout), nSyncTimeout( dMySyncTimeout), nAsyncTimeout( dMyAsyncTimeout), bAbortConnect( bMyAbortConnect), bSsl( bMySsl), bAllowAdmin( bMyAllowAdmin) {} } ; // --------------------------------------------------------------------------- // -------- Funzione di interpretazione della stringa di connessione --------- // --------------------------------------------------------------------------- static bool GetParamsFromConnectionString( const string& sConnection, RedisConnectionInfo& ConnectionInfo) { // Se la stringa di connessione è vuota, errore if ( sConnection.empty()) { LOG_INFO( GetCmdLogger(), "Error : Empty Connection String") ; return false ; } // Ricerco le informazioni Splittando la stringa di connessione per "," string sHostPort, sParams ; SplitFirst( sConnection, ",", sHostPort, sParams) ; if ( ! sHostPort.empty()) { // Host e Porta string sHost, sPort ; SplitFirst( sHostPort, ":", sHost, sPort) ; if ( sHost.empty()) { LOG_INFO( GetCmdLogger(), "Error : Invalid Host") ; return false ; } int nPort = DEFAULT_PORT ; if ( ! sPort.empty()) { if ( ! FromString( sPort, nPort)) { LOG_INFO( GetCmdLogger(), "Error : Invalid Port") ; return false ; } } ConnectionInfo.sHost = sHost ; ConnectionInfo.nPort = nPort ; } // Recupero i parametri ( rimuovo tutti gli spazi ) sParams.erase( remove( sParams.begin(), sParams.end(), ' '), sParams.end()) ; GetValInNotes( sParams, "serviceName", ",", ConnectionInfo.sServiceName) ; GetValInNotes( sParams, "password", ",", ConnectionInfo.sPassword) ; GetValInNotes( sParams, "DefaultDatabase", ",", ConnectionInfo.nDefaultDataBase) ; GetValInNotes( sParams, "keepAlive", ",", ConnectionInfo.nKeepAlive) ; GetValInNotes( sParams, "connectTimeout", ",", ConnectionInfo.nConnectTimeout) ; GetValInNotes( sParams, "syncTimeout", ",", ConnectionInfo.nSyncTimeout) ; GetValInNotes( sParams, "asyncTimeout", ",", ConnectionInfo.nAsyncTimeout) ; GetValInNotes( sParams, "abortConnect", ",", ConnectionInfo.bAbortConnect) ; GetValInNotes( sParams, "ssl", "," ,ConnectionInfo.bSsl) ; GetValInNotes( sParams, "allowAdmin", ",", ConnectionInfo.bAllowAdmin) ; return true ; } // --------------------------------------------------------------------------- // ------------------------------ Sync --------------------------------------- //---------------------------------------------------------------------------- //---------------------------------------------------------------------------- // Funzione di Connessione sincrona a Redis //---------------------------------------------------------------------------- bool ExeRedisConnect( const string& sConnection) { // Se connesione già presente, termino e creo la nuova if ( s_pRedisContext != nullptr) redisFree( s_pRedisContext) ; // Recupero i parametri dalla stringa di connessione RedisConnectionInfo ConnectionInfo ; if ( ! GetParamsFromConnectionString( sConnection, ConnectionInfo)) return false ; // Se connessione senza TimeOut if ( ConnectionInfo.nSyncTimeout <= 0) { // Imposto il contesto per connesione sincrona s_pRedisContext = redisConnect( ConnectionInfo.sHost.c_str(), ConnectionInfo.nPort) ; if ( s_pRedisContext == nullptr) { LOG_INFO( GetCmdLogger(), "Error : Can't allocate redis context") return false ; } else if ( s_pRedisContext->err != 0) { LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + s_pRedisContext->errstr).c_str()) redisFree( s_pRedisContext) ; return false ; } } // Se connessione con parametri else { // TimeOut per connessione struct timeval TimeOutConnection{} ; TimeOutConnection.tv_sec = static_cast( ConnectionInfo.nSyncTimeout) ; TimeOutConnection.tv_usec = static_cast( ( ConnectionInfo.nSyncTimeout - TimeOutConnection.tv_sec) * 1e6) ; // Imposto il contesto per connesione sincrona con TimeOut s_pRedisContext = redisConnectWithTimeout( ConnectionInfo.sHost.c_str(), ConnectionInfo.nPort, TimeOutConnection) ; if ( s_pRedisContext == nullptr) { LOG_INFO( GetCmdLogger(), "Error : Can't allocate redis context") return false ; } else if ( s_pRedisContext->err != 0) { LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + s_pRedisContext->errstr).c_str()) redisFree( s_pRedisContext) ; return false ; } // TimeOut per richiesta I/O struct timeval TimeOutIO{} ; TimeOutIO.tv_sec = static_cast( ConnectionInfo.nSyncTimeout) ; TimeOutIO.tv_usec = static_cast( ( ConnectionInfo.nSyncTimeout - TimeOutConnection.tv_sec) * 1e6) ; // Imposto TimeOut per I/O int nResult = redisSetTimeout( s_pRedisContext, TimeOutIO) ; if ( nResult != REDIS_OK) { LOG_INFO( GetCmdLogger(), "Error : Can't set timeout I/O") ; return false ; } } LOG_INFO( GetCmdLogger(), "Sync connection to redis !") // Verifico se la connessione è di tipo Sentinel redisReply* replySentinel = ( redisReply*)redisCommand( s_pRedisContext, "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 string sMasterHost = replySentinel->element[0]->str ; int nMasterPort = DEFAULT_SENTINEL_MASTER_PORT ; FromString( replySentinel->element[1]->str, nMasterPort) ; freeReplyObject( replySentinel) ; // Effettuo la connessione al master redis s_pRedisContext = redisConnect( sMasterHost.c_str(), nMasterPort) ; if ( s_pRedisContext == nullptr) { LOG_INFO( GetCmdLogger(), "Error : Can't allocate redis context") return false ; } else if ( s_pRedisContext->err != 0) { LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + s_pRedisContext->errstr).c_str()) redisFree( s_pRedisContext) ; return false ; } } // Verifico se richiesta autenticazione if ( ! ConnectionInfo.sPassword.empty()) { redisReply* reply = ( redisReply*)redisCommand( s_pRedisContext, "AUTH %s", ConnectionInfo.sPassword.c_str()) ; if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) { LOG_INFO(GetCmdLogger(), "Error : Authentication failed") ; if ( reply == nullptr) freeReplyObject( reply) ; redisFree( s_pRedisContext) ; return false ; } freeReplyObject( reply) ; } // Seleziono il DataBase redisReply* reply = ( redisReply*)redisCommand( s_pRedisContext, "SELECT %d", ConnectionInfo.nDefaultDataBase) ; if ( reply == nullptr) { LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + s_pRedisContext->errstr).c_str()) ; return false ; } else if ( reply->type == REDIS_REPLY_ERROR) { LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + reply->str).c_str()) freeReplyObject( reply) ; return false ; } LOG_INFO( GetCmdLogger(), ( string{ "Connected to DB #"} + ToString( ConnectionInfo.nDefaultDataBase) + " !").c_str()) freeReplyObject( reply) ; return true ; } //---------------------------------------------------------------------------- // Funzione di Disconnessione sincrona a Redis //---------------------------------------------------------------------------- bool ExeRedisDisconnect( void) { // Se connessione non presente if ( s_pRedisContext == nullptr) { LOG_INFO( GetCmdLogger(), "Error : Closing a Sync Connection never created") ; return false ; } // Se connessione in stato di errore if ( s_pRedisContext->err != 0) { LOG_INFO( GetCmdLogger(), "Warning : Closing Redis Sync Connection in error") ; redisFree( s_pRedisContext) ; s_pRedisContext = nullptr ; return false ; } // Effettuo disconnessione sincrona redisFree( s_pRedisContext) ; s_pRedisContext = nullptr ; LOG_INFO( GetCmdLogger(), "Sync Connection Closed !") ; return true ; } //---------------------------------------------------------------------------- // Funzione di scrittura sincrona chiave-valore //---------------------------------------------------------------------------- bool ExeRedisSetValFromKey( const string& sKey, const string& sVal) { // Se connesione non presente, errore if ( s_pRedisContext == nullptr) { LOG_INFO( GetCmdLogger(), "Error : No Connection") ; return false ; } // Se connessione in stato di errore, allora errore if ( s_pRedisContext->err != 0) { LOG_INFO( GetCmdLogger(), "Error : Invalid Connection") ; return false ; } // Il valore associato alla chiave può essere solo di tipo stringa redisReply* reply = ( redisReply*)redisCommand( s_pRedisContext, "SET %s %s", sKey.c_str(), sVal.c_str()) ; if ( reply == nullptr) { LOG_INFO( GetCmdLogger(), "Error : Null reply") ; return false ; } else if ( reply->type != REDIS_REPLY_STATUS || string( reply->str ) != "OK") { LOG_INFO( GetCmdLogger(), "Error SET : No answer") ; freeReplyObject( reply) ; return false ; } LOG_INFO( GetCmdLogger(), ( string{ "New Key Added : ["} + sKey + ":" + sVal + "]").c_str()) freeReplyObject( reply) ; return true ; } //---------------------------------------------------------------------------- // Funzione di lettura sincrona per valore di una chiave //---------------------------------------------------------------------------- bool ExeRedisGetValFromKey( const string& sKey, string& sVal) { sVal.clear() ; // Se connesione non presente, errore if ( s_pRedisContext == nullptr) { LOG_INFO( GetCmdLogger(), "Error : No Connection") ; return false ; } // Se connessione in stato di errore, allora errore if ( s_pRedisContext->err != 0) { LOG_INFO( GetCmdLogger(), "Error : Invalid Connection") ; return false ; } // Recupero il tipo associato alla chiave da Redis redisReply* typeReply = ( redisReply*)redisCommand( s_pRedisContext, "TYPE %s", sKey.c_str()) ; if ( typeReply == nullptr || typeReply->type != REDIS_REPLY_STATUS) { string sErr = ( typeReply != nullptr ? string( typeReply->str) : "null reply") ; if ( typeReply == nullptr) freeReplyObject( typeReply) ; LOG_INFO( GetCmdLogger(), ( "Error : Failed to get key type, " + sErr).c_str()) ; return false ; } string sType = typeReply->str ; freeReplyObject( typeReply) ; // Determino il formato per la chiamata al DataBase Redis // Per ora accettate solamente valori string string sFormat = ( sType == KEY_TYPE_STRING ? "Get %s" : "") ; if ( sFormat.empty()) { LOG_ERROR( GetCmdLogger(), "Error : Not valid Key Type") ; return false ; } // Effettuo la chiamata redisReply* reply = nullptr ; reply = ( redisReply*)redisCommand( s_pRedisContext, sFormat.c_str(), sKey.c_str()) ; if ( reply == nullptr) { LOG_INFO( GetCmdLogger(), "Error : Null reply") ; return false ; } // Recupero il Valore if ( sType == KEY_TYPE_STRING) { if ( reply->type != REDIS_REPLY_STRING) { LOG_INFO( GetCmdLogger(), "Error : Failed to read the value") ; freeReplyObject( reply) ; return false ; } sVal = string( reply->str) ; } freeReplyObject( reply) ; return true ; } // --------------------------------------------------------------------------- // ------------------------------ ASync -------------------------------------- //---------------------------------------------------------------------------- //---------------------------------------------------------------------------- // Funzione Ciclo degli eventi per chiamate CallBack //---------------------------------------------------------------------------- static void RedisEventLoop( redisAsyncContext* ctx, bool bIsPub) { // Verifica della validità del contesto if ( ctx == nullptr || ctx->err != 0) { LOG_INFO( GetCmdLogger(), ( string{ "Error: Invalid Redis context at start of event loop for " } + ( bIsPub ? "Pub" : "Sub")).c_str()) return ; } // Recupero del file descriptor del socket TCP usato da Redis. SOCKET sock = ctx->c.fd ; // Ciclo while ( bIsPub ? s_bPubLoopRunning : s_bSubLoopRunning) { // 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 ; 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} ; int nRet = select( 0, &rfds, &wfds, nullptr, &timeout) ; if ( nRet == SOCKET_ERROR) { // Se errore -> interruzione del ciclo LOG_INFO( GetCmdLogger(), ( string{ "Error : Socket select() failed -> "} + 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)) ; } 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)) ; } return ; } //---------------------------------------------------------------------------- // Funzione di controllo che la connessione sia attiva e il contesto valido // --------------------------------------------------------------------------- static bool CheckConnectionAndContext( redisAsyncContext* ctx, bool bIsPub) { // Se contesto nullo -> Errore if ( ctx == nullptr) { LOG_INFO( GetCmdLogger(), ( string{ "Error : Invalid Context for "} + ( bIsPub ? "Pub" : "Sub")).c_str()) return false ; } // Se connessione non effettuata -> Errore if ( bIsPub && ! s_bPubConnected) { LOG_INFO( GetCmdLogger(), "Error : aSync Pub connection not connected") ; redisAsyncFree( ctx) ; ctx = nullptr ; return false ; } if ( ! bIsPub && ! s_bSubConnected) { LOG_INFO( GetCmdLogger(), "Error : aSync Sub connection not closed") ; redisAsyncFree( ctx) ; ctx = nullptr ; return false ; } // Se connessione effettuata ma in stato di errore -> Errore if ( ctx->err != 0) { LOG_INFO( GetCmdLogger(), ( string{ "Warning : Closing aSync Connection in error for "} + ( bIsPub ? "Pub" : "Sub")).c_str()) ; redisAsyncFree( ctx) ; ctx = nullptr ; return false ; } return true ; } //---------------------------------------------------------------------------- // Funzione di CallBack per PUBLISH, SUBSCRIBE e UNSUBSCRIBE //---------------------------------------------------------------------------- static void MessageCallback( redisAsyncContext* ctx, void* r, void*) { // Verifico che il contesto sia definito if ( ctx == nullptr) { LOG_INFO( GetCmdLogger(), "Error : Redis context null in CallBack") return ; } LOG_INFO( GetCmdLogger(), "RedisMessageCallback invoked") // Recupero della risposta dal server redis redisReply* reply = static_cast( r) ; if ( reply == nullptr) { LOG_INFO( GetCmdLogger(), "Error : Redis reply null in CallBack") return ; } else if ( reply->type == REDIS_REPLY_INTEGER) { // Recupero il numero di clients iscritti al canale dove ho effettuato una publish LOG_INFO( GetCmdLogger(), ( string{ "Message published to "} + ToString( reply->integer) + " clients !").c_str()) ; return ; } else if ( reply->type == REDIS_REPLY_ARRAY && reply->elements >= 3) { // Recupero la tipologia di chiamata effettuata const char* msgType = ( reply->element[0] != nullptr && reply->element[0]->str != nullptr ? reply->element[0]->str : "") ; // --- Se Messaggio if ( strcmp( msgType, "message") == 0 && reply->elements == 3) { // Messaggio ricevut da un canale if ( reply->element[1] != nullptr && reply->element[1]->str != nullptr && reply->element[2] != nullptr && reply->element[2]->str != nullptr) { string sChannel = reply->element[1]->str ; string sMessage = reply->element[2]->str ; LOG_INFO( GetCmdLogger(), ( string{ "Message Received on ["} + sChannel + "] : " + sMessage).c_str()) } else { LOG_INFO( GetCmdLogger(), "Errror : Invalid Message received") return ; } } // --- Se Subscribe else if ( strcmp( msgType, "subscribe") == 0 && reply->elements >= 3) { if ( reply->element[1] != nullptr && reply->element[1] != nullptr && reply->element[2] != nullptr) { string sChannel = reply->element[1]->str ; int nCount = reply->element[2]->integer ; LOG_INFO( GetCmdLogger(), ( string{ "Subscribed to ["} + sChannel + "]," + " total subscribtions : " + ToString( nCount)).c_str()) ; } else { LOG_INFO( GetCmdLogger(), "Errror : Invalid Subscription reply") return ; } } // --- se Unsubscribe else if ( strcmp( msgType, "unsubscribe") == 0 && reply->elements >= 3) { if ( reply->element[1] != nullptr && reply->element[1] != nullptr && reply->element[2] != nullptr) { string sChannel = reply->element[1]->str ; int nCount = reply->element[2]->integer ; LOG_INFO( GetCmdLogger(), ( string{ "Unsubscribed to ["} + sChannel + "]," + " subscriptions left : " + ToString( nCount)).c_str()) } else { LOG_INFO( GetCmdLogger(), "Errror : Invalid Unbubscription reply") return ; } } // --- Undefined else { LOG_INFO( GetCmdLogger(), "Undefined reply in CallBack from Redis") return ; } } return ; } //---------------------------------------------------------------------------- // Funzione di CallBack per WaitForReadisMessage //---------------------------------------------------------------------------- static void WaitMessageCallback( redisAsyncContext* ctx, void* r, void*) { // Verifico che il contesto sia definito if ( ctx == nullptr) { LOG_INFO( GetCmdLogger(), "Error : Redis context null in CallBack") return ; } LOG_INFO( GetCmdLogger(), "RedisMessageCallback invoked") // Recupero della risposta dal server redis redisReply* reply = static_cast( r) ; if ( reply == nullptr) { LOG_INFO( GetCmdLogger(), "Error : Redis reply null in CallBack") return ; } else if ( reply->type != REDIS_REPLY_ARRAY || reply->elements < 3) LOG_INFO( GetCmdLogger(), "Error : Invalid Message CallBack reply") else { // Recupero il Messaggio ( se callBack riferita al tipo "message") if ( reply->element[0] == nullptr || reply->element[0]->str == nullptr) LOG_INFO( GetCmdLogger(), "Error : Null message type received") else { const char* msgType = reply->element[0]->str ; if ( strcmp( msgType, "message") == 0) { if ( reply->element[2] == nullptr || reply->element[2]->str == nullptr) LOG_INFO( GetCmdLogger(), "Error : Null message received") else { s_sMessage = reply->element[2]->str ; s_bMessage = true ; } } } } } //---------------------------------------------------------------------------- // Funzione per Memorizzare il DataBase Redis da selezionare //---------------------------------------------------------------------------- static bool SetPendingDataBase( int nDB) { // I DataBase su Redis sono al più 15 if ( REDIS_MIN_DB <= nDB && nDB <= REDIS_MAX_DB) { s_nPendingDataBase = nDB ; return true ; } return false ; } //---------------------------------------------------------------------------- // Funzione per Ricavare il numero di DataBase Redis da selezionare //---------------------------------------------------------------------------- static int GetPendingDataBase() { return s_nPendingDataBase ; } //---------------------------------------------------------------------------- static bool GetConnectionContext( redisAsyncContext*& ctx, const string& sHost, int nPort) { ctx = redisAsyncConnect( sHost.c_str(), nPort) ; if ( ctx == nullptr) { LOG_INFO( GetCmdLogger(), "Error : Can't allocate Redis Async context") return false ; } if ( ctx->err != 0) { LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + ctx->errstr).c_str()) redisAsyncFree( ctx) ; ctx = nullptr ; WSACleanup() ; return false ; } return true ; } //---------------------------------------------------------------------------- static void CleanupWinsock( void) { // entrambi i contesti devono essere nulli if ( s_pRedisAsyncPubContext == nullptr && s_pRedisAsyncSubContext == nullptr) { // entrambi i loop devono essere terminati if ( ! s_bPubLoopRunning && ! s_bSubLoopRunning) { WSACleanup() ; LOG_INFO( GetCmdLogger(), "Winsock cleaned up") ; } } } //---------------------------------------------------------------------------- // Funzione di Connessione asincrona a Redis //---------------------------------------------------------------------------- bool ExeRedisAsyncConnect( const string& sConnection) { // Flag di connessione s_bPubConnected = false ; s_bSubConnected = false ; // Recupero i parametri dalla stringa di connessione RedisConnectionInfo ConnectionInfo ; if ( ! GetParamsFromConnectionString( sConnection, ConnectionInfo)) return false ; // Se connesione già presente, termino e creo la nuova if ( s_pRedisAsyncPubContext != nullptr || s_pRedisAsyncSubContext != nullptr) ExeRedisAsyncDisconnect() ; // Memorizzo il valore del DataBase, la selezione del DataBase deve avvenire una volta // che la connessione sia effettivamente stabilita e corretta ( -> CallBack di connessione) if ( ! SetPendingDataBase( ConnectionInfo.nDefaultDataBase)) { LOG_INFO( GetCmdLogger(), "Error : DataBase number must be in range [0,15]") ; return false ; } // Inizializzazione libreria Winsock, necessaria per usare le funzionalità di rete su Windows, // come socket TCP/IP. Versione utilizzata 2.2 WSADATA wsaData ; int nStatus = WSAStartup( MAKEWORD( 2, 2), &wsaData) ; if ( nStatus != 0) { LOG_INFO( GetCmdLogger(), ( string{ "Error : WSAStartup -> "} + ToString( nStatus)).c_str()) ; return false ; } // 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)) return false ; LOG_INFO( GetCmdLogger(), "Connected to Redis ! ( Pub/Sub)") // Imposto e Definisco la funzione di CallBack per contesto PUB per Connessione asincrona redisAsyncSetConnectCallback( 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) { LOG_INFO( GetCmdLogger(), "Error : Async Connect CallBack Pub failed ") ; return ; } // --- Se Connessione Ok else { // --- Selezione del DataBase 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 redisReply* reply = static_cast( r) ; if ( reply == nullptr) { // --- Nulla LOG_INFO( GetCmdLogger(), "Error : Null Database selection reply") ; 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 ; } }, nullptr, "SELECT %d", GetPendingDataBase() ) ; } } ) ; // Imposto e Definisco la funzione di CallBack per contesto SUB per Connessione asincrona redisAsyncSetConnectCallback( 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") ; 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 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") ; return ; } // --- Richiesta di Selezione DB a Redis redisReply* reply = static_cast( r) ; if ( reply == nullptr) { // --- Nulla LOG_INFO( GetCmdLogger(), "Error : Null Database selection 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 ; return ; } else { // --- Errore string sErrMsg = ( reply->str != nullptr) ? reply->str : "Null asnwer" ; LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str()) return ; } }, nullptr, "SELECT %d", GetPendingDataBase() ) ; } } ) ; // Imposto e Definisco la funzione di CallBack per la Disconnessione Pub asincrona redisAsyncSetDisconnectCallback( s_pRedisAsyncPubContext, []( const redisAsyncContext* ctx, int status) { LOG_INFO( GetCmdLogger(), "Disconnected from Redis ( Pub )") ; s_bPubLoopRunning = false ; s_pRedisAsyncPubContext = nullptr ; this_thread::sleep_for( chrono::milliseconds( 50)) ; // per sicurezza CleanupWinsock() ; } ) ; // Imposto e Definisco la funzione di CallBack per la Disconnessione Sub asincrona redisAsyncSetDisconnectCallback( s_pRedisAsyncSubContext, []( const redisAsyncContext* ctx, int status) { LOG_INFO( GetCmdLogger(), "Disconnected from Redis ( Sub )") ; s_pRedisAsyncSubContext = nullptr ; s_bSubLoopRunning = false ; this_thread::sleep_for( chrono::milliseconds( 50)) ; // per sicurezza CleanupWinsock() ; } ) ; // Definizione e partenza del ciclo degli eventi in un thread separato ( in BackGround) s_bPubLoopRunning = true ; s_bSubLoopRunning = true ; thread tPubEventLoop( RedisEventLoop, s_pRedisAsyncPubContext, true) ; thread tSubEventLoop( RedisEventLoop, s_pRedisAsyncSubContext, false) ; tPubEventLoop.detach() ; tSubEventLoop.detach() ; // 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( ConnectionInfo.nAsyncTimeout)) { this_thread::sleep_for( chrono::milliseconds( nStepMillSec)) ; nWaitedMillSec += nStepMillSec ; } 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()) } return true ; } //---------------------------------------------------------------------------- // Funzione di Disconnessione asincrona a Redis //---------------------------------------------------------------------------- bool ExeRedisAsyncDisconnect( void) { // Controllo che la connessione sia valida e il contesto ben definito bool bOkPub = CheckConnectionAndContext( s_pRedisAsyncPubContext, true) ; bool bOkSub = CheckConnectionAndContext( s_pRedisAsyncSubContext, false) ; if ( ! bOkSub || ! bOkPub) return false ; // Effettuo la Disconnesione asincrona redisAsyncDisconnect( s_pRedisAsyncPubContext) ; // libera la memoria in automatico redisAsyncDisconnect( s_pRedisAsyncSubContext) ; // libera la memoria in automatico LOG_INFO( GetCmdLogger(), "Async Connection Closed !") ; return true ; } //---------------------------------------------------------------------------- // Funzione di Publish asincrona a Redis //---------------------------------------------------------------------------- bool ExeRedisAsyncPublish( const string& sChannel, const string& sMessage) { // Controllo che la connessione Pub sia valida e il contesto Pub ben definito if ( ! CheckConnectionAndContext( s_pRedisAsyncPubContext, true)) return false ; // Se non ho alcun canale -> errore if ( sChannel.empty()) { LOG_INFO( GetCmdLogger(), "Error : Not a valid Channel") return false ; } // Se non ho alcun messaggio -> warning if ( sMessage.empty()) LOG_INFO( GetCmdLogger(), "Warning : Empty Message") ; // Eseguo il comando PUBLISH if ( redisAsyncCommand( s_pRedisAsyncPubContext, MessageCallback, nullptr, "PUBLISH %s %s", sChannel.c_str(), sMessage.c_str()) != REDIS_OK) { LOG_INFO( GetCmdLogger(), "Error: redisAsyncCommand PUBLISH failed") return false ; } return true ; } //---------------------------------------------------------------------------- // Funzione di Subscribe asincrona a Redis //---------------------------------------------------------------------------- bool ExeRedisAsyncSubscribe( const string& sChannel) { // Controllo che la connessione sia valida e il contesto ben definito if ( ! CheckConnectionAndContext( s_pRedisAsyncSubContext, false)) return false ; // Se non ho alcun canale, errore if ( sChannel.empty()) { LOG_INFO( GetCmdLogger(), "Error : Not a valid Channel") ; return false ; } // Eseguo il comando Subscribe if ( redisAsyncCommand( s_pRedisAsyncSubContext, MessageCallback, nullptr, ( "SUBSCRIBE " + sChannel).c_str()) != REDIS_OK) { LOG_INFO( GetCmdLogger(), "Error : redisAsyncCommand SUBSCRIBE failed") return false ; } return true ; } //---------------------------------------------------------------------------- // Funzione di Unsubscribe asincrona a Redis //---------------------------------------------------------------------------- bool ExeRedisAsyncUnsubscribe( const string& sChannel) { // Controllo che la connessione sia valida e il contesto ben definito if ( ! CheckConnectionAndContext( s_pRedisAsyncSubContext, false)) return false ; // Se non ho alcun canale, errore if ( sChannel.empty()) { LOG_INFO( GetCmdLogger(), "Error : Not a valid Channel") return false ; } // Eseguo il comando Unsubscribe if ( redisAsyncCommand( s_pRedisAsyncSubContext, MessageCallback, nullptr, ( "UNSUBSCRIBE " + sChannel).c_str()) != REDIS_OK) { LOG_INFO( GetCmdLogger(), "Error : redisAsyncCommand UNSUBSCRIBE failed") return false ; } return true ; } //---------------------------------------------------------------------------- // Funzione di che aspetta un messaggio asincro su un canale Redis //---------------------------------------------------------------------------- // La funzione esegue una Subscribe e una Unsubscribe // Se entro il tempo stabilito riceve un messaggio, lo comunica, altrimenti errore bool ExeRedisAsyncSubscribeOneMessage( const string& sChannel, double dMaxTimeOut, string& sMessage) { // Controllo che la connessione sia valida e il contesto ben definito if ( ! CheckConnectionAndContext( s_pRedisAsyncPubContext, true)) return false ; // Se non ho alcun canale, errore if ( sChannel.empty()) { LOG_INFO( GetCmdLogger(), "Error : Not a valid Channel") return false ; } // Controllo validità tempo di TimeOut if ( dMaxTimeOut <= 0.) { LOG_INFO( GetCmdLogger(), "Error : Invalid timeout value") ; return false ; } // flag per ricezione del messaggio s_bMessage = false ; s_sMessage.clear() ; // Invio del comando di SubScribe if ( redisAsyncCommand( s_pRedisAsyncSubContext, WaitMessageCallback, nullptr, ( "SUBSCRIBE " + sChannel).c_str()) != REDIS_OK) { LOG_INFO( GetCmdLogger(), ( string{ "Error: Failed to subscribe to"} + sChannel).c_str()) return false ; } // Attivo il TimeOut per l'attesa del messaggio const int nStepMillSeconds = 2 ; int nWaitedMillSeconds = 0 ; while ( ! s_bMessage && nWaitedMillSeconds < static_cast( dMaxTimeOut)) { this_thread::sleep_for( chrono::milliseconds( nStepMillSeconds)) ; nWaitedMillSeconds += nStepMillSeconds ; } // Invio del comando Unsubscribe if ( redisAsyncCommand( s_pRedisAsyncSubContext, WaitMessageCallback, nullptr, ( "UNSUBSCRIBE " + sChannel).c_str()) != REDIS_OK) { LOG_INFO( GetCmdLogger(), ( string{ "Error: Failed to Unsubscribe to "} + sChannel).c_str()) return false ; } // Controllo se ho ricevuto un messaggio if ( s_bMessage) { sMessage = s_sMessage ; LOG_INFO( GetCmdLogger(), ( string{ "Message received on ["} + sChannel + "]: " + sMessage).c_str()) return true ; } LOG_INFO( GetCmdLogger(), ( string{ "Timeout: No message received on ["} + sChannel + "]").c_str()) ; return false ; }