//----------------------------------------------------------------------------
//  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. ( ver. 2.7i3)
// Modifiche : 31.10.25 RE Aggiunta gestione connessioni multiple. ( ver. 2.7j4)
// Modifiche : 04.12.25 RE Aggiunta Memorizzazione messaggio. ( ver 2.7l1)
//
//
//----------------------------------------------------------------------------

//--------------------------- Include ----------------------------------------
#include <winsock2.h>
#include <WS2tcpip.h>
#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 <thread>
#include <atomic>
#include <map>
#include <sstream>
#include <future>

using namespace std ;

// ---------------------------------------------------------------------------
// Ambiente Debug
// ---------------------------------------------------------------------------
#define DEBUG 0

// ---------------------------------------------------------------------------
// Definizione variabili generali Redis
// ---------------------------------------------------------------------------
static const int REDIS_MIN_DB = 0 ;
static const int REDIS_MAX_DB = 15 ;
static const int DEFAULT_PORT = 6379 ;
static const int DEFAULT_SENTINEL_MASTER_PORT = 26379 ;
static const string DEFAULT_USER = "default" ;
static const int DEFAULT_DATABASE = 0 ;
static const int DEFAULT_TIMEOUT = 15000 ;   // [ms]
static const int DEFAULT_ALIVE_TIME = 180 ;  // [s]
static const int MAX_CONNECTION_NBR = 10 ;
static const int NO_CONNECTION = - 1 ;
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
struct RedisConnectionInfo {

   string sHost, sServiceName, sUser, sPassword ;
   int nPort, nDefaultDataBase, nKeepAlive, nConnectTimeout, nSyncTimeout, nAsyncTimeout ;
   bool bAbortConnect, bSsl, bAllowAdmin ;

   RedisConnectionInfo() {
      sHost = "" ; sServiceName = "" ; sUser = DEFAULT_USER ; 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 sMyUser, string sMyPassword,
                        int nMyConnection, int nMyPort, int nMyDefaultDataBase, int dMyKeepAlive,
                        int dMyConnectTimeout, int dMySyncTimeout, int dMyAsyncTimeout,
                        bool bMyAbortConnect, bool bMySsl, bool bMyAllowAdmin)
      : sHost( sMyHost), sServiceName( sMyServiceName), sUser( sMyUser), sPassword( sMyPassword),
        nPort( nMyPort), nDefaultDataBase( nMyDefaultDataBase), nKeepAlive( dMyKeepAlive),
        nConnectTimeout( dMyConnectTimeout), nSyncTimeout( dMySyncTimeout),
        nAsyncTimeout( dMyAsyncTimeout), bAbortConnect( bMyAbortConnect), bSsl( bMySsl),
        bAllowAdmin( bMyAllowAdmin) {}

   inline void Clear() {
      sHost = "" ; sServiceName = "" ; sUser = DEFAULT_USER ; 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 ;
   }

} ;

// Struttura per Connessiona Sincrona
struct RedisSync {

   atomic<bool> bFreeConnection ;
   redisContext* pRedisContext ;
   RedisConnectionInfo ConnectionInfo ;

   RedisSync()
      : bFreeConnection( true), pRedisContext( nullptr), ConnectionInfo() {}

   inline void Clear() {
      bFreeConnection = true ;
      redisFree( pRedisContext) ;
      pRedisContext = nullptr ;
      ConnectionInfo.Clear() ;
   }

} ;
static array<RedisSync, MAX_CONNECTION_NBR> s_vRedisClients ;

// Struttura per stato di Connessione Asincrona
struct RedisConnectionStatus {

   atomic<int>  nIdConnection ;
   atomic<bool> bConnected ;
   atomic<bool> bAlive ;
   atomic<bool> bAuthenticated ;
   atomic<bool> bSelectedDB ;
   atomic<bool> bDisconnected ;
   atomic<bool> bPublished ;
   atomic<bool> bSubscribed ;
   atomic<bool> bUnsubscribed ;
   atomic<bool> bMessage ;
   string sMessage ;                // univoco grazie a bMessage

   RedisConnectionStatus()
      : nIdConnection( 0),
        bConnected( false), bAlive( false), bAuthenticated( false), bSelectedDB( false), bDisconnected( false),
        bPublished( false), bSubscribed( false), bUnsubscribed( false), bMessage( false),
        sMessage( "") {}

   inline void Clear() {
      nIdConnection = 0 ;
      bConnected = false ;
      bAlive = false ;
      bAuthenticated = false ;
      bSelectedDB = false ;
      bDisconnected = false ;
      bPublished = false ;
      bSubscribed = false ;
      bUnsubscribed = false ;
      bMessage = false ;
      sMessage = "" ;
   }

} ;

// Classe per Connessione Asincrona
class RedisAsync {

   public :
      atomic<bool> m_bFreeConnection ;
      redisAsyncContext* m_pRedisAsyncPubContext ;
      redisAsyncContext* m_pRedisAsyncSubContext ;
      string m_sConnectionString ;
      unordered_set<string> m_set_SubChannels ;
      int m_nDataBase ;
      string m_sPassword ;
      string m_sUser ;
      atomic<bool> m_bPubLoopRunning ;
      atomic<bool> m_bSubLoopRunning ;
      RedisConnectionInfo m_ConnectionInfo ;
      RedisConnectionStatus m_ConnectionSubStatus ;
      RedisConnectionStatus m_ConnectionPubStatus ;
      thread m_PubThread ;
      thread m_SubThread ;
      WSADATA m_wsaData ;

   RedisAsync()
      :  m_bFreeConnection( true), m_pRedisAsyncPubContext( nullptr), m_pRedisAsyncSubContext( nullptr),
         m_sConnectionString( ""), m_set_SubChannels(), m_nDataBase( REDIS_MIN_DB), m_sPassword( ""), m_sUser( DEFAULT_USER),
         m_bPubLoopRunning( false), m_bSubLoopRunning( false),
         m_ConnectionInfo(), m_ConnectionSubStatus(), m_ConnectionPubStatus() {} ;
   // thread è già inizializzato con il suo costruttore di Default valido

   inline void Clear() {

     // Loop da terminare
      m_bPubLoopRunning.store( false, memory_order_release) ;
      m_bSubLoopRunning.store( false, memory_order_release) ;

     // Sveglio i Thread bloccanti
      if ( m_pRedisAsyncPubContext)
         redisAsyncDisconnect( m_pRedisAsyncPubContext) ;
      if ( m_pRedisAsyncSubContext) 
         redisAsyncDisconnect( m_pRedisAsyncSubContext) ;

     // Join dei Thread
      if ( m_PubThread.joinable())
         m_PubThread.join() ;
      if ( m_SubThread.joinable())
         m_SubThread.join() ;

     // Rilascio dei contesti Redis
      if ( m_pRedisAsyncPubContext) {
         redisAsyncFree( m_pRedisAsyncPubContext) ;
         m_pRedisAsyncPubContext = nullptr ;
      }
      if ( m_pRedisAsyncSubContext) {
         redisAsyncFree( m_pRedisAsyncSubContext) ;
         m_pRedisAsyncSubContext = nullptr ;
      }

     // Reset dello Stato
      m_bFreeConnection.store( true, memory_order_release) ;
      m_sConnectionString.clear() ;
      m_nDataBase = REDIS_MIN_DB ;
      m_sPassword.clear() ;
      m_sUser = DEFAULT_USER ;

      m_ConnectionInfo.Clear() ;
      m_ConnectionSubStatus.Clear() ;
      m_ConnectionPubStatus.Clear() ;
      m_set_SubChannels.clear() ;
   }

   ~RedisAsync() {
      if ( m_PubThread.joinable())
         m_PubThread.join() ;
      if ( m_SubThread.joinable())
         m_SubThread.join();
   }

   public :
      void RedisEventLoop( bool bIsSub) const ;

} ;
static array<RedisAsync, MAX_CONNECTION_NBR> s_vAsyncRedisClients ;

// Mappa lettura Messaggi per Modalità Asicnrona
struct RedisAsyncMessage {

   atomic_flag Lock = ATOMIC_FLAG_INIT ;
   int nCount = 0 ;
   string sMessage ;

} ;
static unordered_map<string, RedisAsyncMessage> s_vAsyncRedisMessages ;

// ---------------------------------------------------------------------------
// [Redis] Funzione Ciclo degli eventi per Read/Write su Socket
// ---------------------------------------------------------------------------
void
RedisAsync::RedisEventLoop( bool bIsPub) const
{
  // Verifica della validità del contesto
   redisAsyncContext* ctx = ( bIsPub ? m_pRedisAsyncPubContext : m_pRedisAsyncSubContext) ;
   if ( ctx == nullptr  ||  ctx->err != 0)
      return ; // se attivazione loop prima della inizializzazione callBack

  // Recupero del file descriptor del socket TCP usato da Redis.
   SOCKET sock = ctx->c.fd ;
  // 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 ? m_bPubLoopRunning.load( memory_order_acquire)
                    : m_bSubLoopRunning.load( memory_order_acquire))) {
     // Se contesto non valido, interrompo il ciclo
      if ( ctx == nullptr  ||  ( ctx->c.flags & REDIS_DISCONNECTING))
         break ;
     // Recupero file descriptor di lettura e scrittura
      FD_ZERO( &rfds) ;
      FD_ZERO( &wfds) ;
      FD_SET( sock, &rfds) ;
      FD_SET( sock, &wfds) ;
     // Controllo il risultato ottenuto
      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) {
        // 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) ;
         }
        // Evito saturazione CPU
         this_thread::sleep_for( chrono::milliseconds( 1)) ;
      }

     // Reset del timeOut ad ogni ciclo
      timeout.tv_sec = 0 ;
      timeout.tv_usec = 100000 ; // 100ms
   }

   return ;
}

//----------------------------------------------------------------------------
// [Utility] Funzione di Controllo per Id Connessione
// ---------------------------------------------------------------------------
static bool
CheckIdConnection( int nIdConnection)
{
   if ( nIdConnection < 0  ||  nIdConnection >= MAX_CONNECTION_NBR) {
      LOG_INFO( GetCmdLogger(), ( string{ "Error : Not a Valid id Connestion : [1, "} + ToString( MAX_CONNECTION_NBR) + "]").c_str())
      return false ;
   }
   return true ;
}

// ---------------------------------------------------------------------------
// [Utility] Funzione di Split parametri per 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, "user", ",", ConnectionInfo.sUser) ;
   GetValInNotes( sParams, "password", ",", ConnectionInfo.sPassword) ;
   GetValInNotes( sParams, "DefaultDatabase", ",", ConnectionInfo.nDefaultDataBase) ;
   if ( ConnectionInfo.nDefaultDataBase < 0  ||  ConnectionInfo.nDefaultDataBase > REDIS_MAX_DB) {
      ConnectionInfo.Clear() ;
      LOG_INFO( GetCmdLogger(), "Error : DB Out of range [0, 15]")
      return false ;
   }
   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 ;
}

// ---------------------------------------------------------------------------
// [Utility] Funzione di Creazione Chiave di accesso alla Mappa dei Messaggi
// ---------------------------------------------------------------------------
static string
GetMessageMapKey( int nIdConnection, const string& sChannel)
{
   return ( ToString( nIdConnection) + "_" + sChannel) ;
}

//----------------------------------------------------------------------------
// Funzione per comando CONNECT Sincrono
// ---------------------------------------------------------------------------
bool
ExeRedisConnect( const string& sConnection, int& nIdConnection)
{
  // Inizializzo una nuova connessione, recuperando l'Id
   int nMyIdConnection = NO_CONNECTION ;
   for ( int i = 0 ; i < int( s_vRedisClients.size()) ; ++ i) {
      if ( s_vRedisClients[i].bFreeConnection) {
         nMyIdConnection = i ;
         break ;
      }
   }
   if ( nMyIdConnection == NO_CONNECTION) {
      LOG_INFO( GetCmdLogger(), ( string{ "Error : Limit connection Exceeded ( "} +
                                  ToString( MAX_CONNECTION_NBR) + " )").c_str())
      return false ;
   }

  // Recupero i riferimenti necessari e blocco la connessione corrente
   RedisSync& SyncRedisClient = s_vRedisClients[nMyIdConnection] ;
   SyncRedisClient.bFreeConnection = false ;
   RedisConnectionInfo& ConnectionInfo = SyncRedisClient.ConnectionInfo ;

  // Recupero i parametri dalla stringa di connessione
   if ( ! GetParamsFromConnectionString( sConnection, ConnectionInfo)) {
      SyncRedisClient.Clear() ;
      return false ;
   }

  // Se connessione senza TimeOut
   if ( ConnectionInfo.nSyncTimeout <= 0) {
     // Imposto il contesto per connesione sincrona
      SyncRedisClient.pRedisContext = redisConnect( ConnectionInfo.sHost.c_str(), ConnectionInfo.nPort) ;
      if ( SyncRedisClient.pRedisContext == nullptr) {
         LOG_INFO( GetCmdLogger(), "Error : Can't allocate redis context")
         return false ;
      }
      else if ( SyncRedisClient.pRedisContext->err != 0) {
         LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + SyncRedisClient.pRedisContext->errstr).c_str())
         SyncRedisClient.Clear() ;
         return false ;
      }
   }
  // Se connessione con parametri
   else {
     // TimeOut per connessione
      struct timeval TimeOutConnection{} ;
      TimeOutConnection.tv_sec = static_cast<long>( ConnectionInfo.nSyncTimeout) ;
      TimeOutConnection.tv_usec = static_cast<long>( ( ConnectionInfo.nSyncTimeout - TimeOutConnection.tv_sec) * 1e6) ;

     // Imposto il contesto per connesione sincrona con TimeOut
      SyncRedisClient.pRedisContext = redisConnectWithTimeout( ConnectionInfo.sHost.c_str(), ConnectionInfo.nPort, TimeOutConnection) ;
      if ( SyncRedisClient.pRedisContext == nullptr) {
         LOG_INFO( GetCmdLogger(), "Error : Can't allocate redis context")
         return false ;
      }
      else if ( SyncRedisClient.pRedisContext->err != 0) {
         LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + SyncRedisClient.pRedisContext->errstr).c_str())
         SyncRedisClient.Clear() ;
         return false ;
      }

     // TimeOut per richiesta I/O
      struct timeval TimeOutIO{} ;
      TimeOutIO.tv_sec = static_cast<long>( ConnectionInfo.nSyncTimeout) ;
      TimeOutIO.tv_usec = static_cast<long>( ( ConnectionInfo.nSyncTimeout - TimeOutConnection.tv_sec) * 1e6) ;

     // Imposto TimeOut per I/O
      int nResult = redisSetTimeout( SyncRedisClient.pRedisContext, TimeOutIO) ;
      if ( nResult != REDIS_OK) {
         LOG_INFO( GetCmdLogger(), "Error : Can't set timeout I/O") ;
         return false ;
      }
   }

  // Verifico se la Connessione è di tipo Sentinel
   redisReply* replySentinel = ( redisReply*)redisCommand( SyncRedisClient.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
      freeReplyObject( SyncRedisClient.pRedisContext) ;
      SyncRedisClient.pRedisContext = redisConnect( sMasterHost.c_str(), nMasterPort) ;
      if ( SyncRedisClient.pRedisContext == nullptr) {
         LOG_INFO( GetCmdLogger(), "Error : Can't allocate redis context")
         return false ;
      }
      else if ( SyncRedisClient.pRedisContext->err != 0) {
         LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + SyncRedisClient.pRedisContext->errstr).c_str())
         SyncRedisClient.Clear() ;
         return false ;
      }
   }
   LOG_INFO( GetCmdLogger(), "Connected to Redis !")

  // Verifico se richiesta autenticazione
   if ( ! ConnectionInfo.sPassword.empty()) {
      redisReply* reply = ( redisReply*)redisCommand( SyncRedisClient.pRedisContext, "AUTH %s %s",
                                                      ConnectionInfo.sUser.c_str(), ConnectionInfo.sPassword.c_str()) ;
      if ( reply == nullptr  ||  reply->type == REDIS_REPLY_ERROR) {
         LOG_INFO( GetCmdLogger(), "Error : Authentication failed") ;
         if ( reply == nullptr)
            freeReplyObject( reply) ;
         SyncRedisClient.Clear() ;
         return false ;
      }
      freeReplyObject( reply) ;
   }
   LOG_INFO( GetCmdLogger(), "Authenticated to Redis !")

  // Seleziono il DataBase
   redisReply* reply = ( redisReply*)redisCommand( SyncRedisClient.pRedisContext, "SELECT %d", ConnectionInfo.nDefaultDataBase) ;
   if ( reply == nullptr) {
      LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + SyncRedisClient.pRedisContext->errstr).c_str()) ;
      return false ;
   }
   else if ( reply->type == REDIS_REPLY_ERROR) {
      LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + reply->str).c_str())
      freeReplyObject( reply) ;
      SyncRedisClient.Clear() ;
      return false ;
   }
   freeReplyObject( reply) ;
   LOG_INFO( GetCmdLogger(), "DB Selected to Redis !") ;

  // Restituisco l'Id della connessione [1, MAX_CONNECTION_NBR] ( per Lua)
   nIdConnection = nMyIdConnection + 1 ;
   return true ;
}

//----------------------------------------------------------------------------
// Funzione per comando DISCONNECT Sincrono
// ---------------------------------------------------------------------------
bool
ExeRedisDisconnect( int nIdConnection)
{
  // Verifico che l'Id di connesione sia valido
   int nMyIdConnection = nIdConnection - 1 ;
   if ( ! CheckIdConnection( nMyIdConnection))
      return false ;
  // Verifico che la connessione sia attiva
   if ( s_vRedisClients[nMyIdConnection].bFreeConnection) {
      LOG_INFO( GetCmdLogger(), "Error : Id Connection not found")
      return false ;
   }

  // Pulisco la connessione corrente
   s_vRedisClients[nMyIdConnection].Clear() ;
   LOG_INFO( GetCmdLogger(), "Disconnected to Redis !") ;

   return true ;
}

//----------------------------------------------------------------------------
// Funzione per comando SET Sincrono ( sola scrittura {key:string, val:string})
//----------------------------------------------------------------------------
bool
ExeRedisSetValFromKey( int nIdConnection, const string& sKey, const string& sVal)
{
  // Verifico che l'Id di connesione sia valido
   int nMyIdConnection = nIdConnection - 1 ;
   if ( ! CheckIdConnection( nMyIdConnection))
      return false ;
  // Verifico che la connessione sia attiva
   if ( s_vRedisClients[nMyIdConnection].bFreeConnection) {
      LOG_INFO( GetCmdLogger(), "Error : Id Connection not found")
      return false ;
   }

  // Recupero la connessione corrente
   RedisSync& RedisClient = s_vRedisClients[nMyIdConnection] ;

  // Il valore associato alla chiave può essere solo di tipo stringa
   redisReply* reply = ( redisReply*)redisCommand( RedisClient.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 ;
   }
   freeReplyObject( reply) ;

   LOG_INFO( GetCmdLogger(), ( string{ "New Key Set : ["} + sKey + ":" + sVal + "]").c_str())
   return true ;
}

//----------------------------------------------------------------------------
// Funzione per comando Get Sincrono ( sola lettura {key:string, val:string})
//----------------------------------------------------------------------------
bool
ExeRedisGetValFromKey( int nIdConnection, const string& sKey, string& sVal)
{
   sVal.clear() ;
  // Verifico che l'Id di connesione sia valido
   int nMyIdConnection = nIdConnection - 1 ;
   if ( ! CheckIdConnection( nMyIdConnection))
      return false ;
  // Verifico che la connessione sia attiva
   if ( s_vRedisClients[nMyIdConnection].bFreeConnection) {
      LOG_INFO( GetCmdLogger(), "Error : Id Connection not found")
      return false ;
   }

  // Recupero la connessione corrente
   RedisSync& RedisClient = s_vRedisClients[nMyIdConnection] ;

  // Recupero il tipo associato alla chiave da Redis
   redisReply* typeReply = ( redisReply*)redisCommand( RedisClient.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( RedisClient.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) ;
   }
   LOG_INFO( GetCmdLogger(), ( string{ "Key Read : ["} + sKey + ":" + sVal + "]").c_str())

   freeReplyObject( reply) ;
   return true ;
}

// ---------------------------------------------------------------------------
// [CB] Funzione di CallBack per comando SELECT Asincrono ( Selezione DB)
// ---------------------------------------------------------------------------
static void
RedisDbSElectionCallBack( redisAsyncContext* ctx, void* r, void*)
{
  // 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 answer" ;
       LOG_INFO( GetCmdLogger(), ( string{ "Error : Db Selection -> "} + sErrMsg).c_str()) ;
       return ;
   }
  // Imposto flag di Autenticazione
   RedisConnectionStatus* MyStatus = static_cast<RedisConnectionStatus*>( ctx->data) ;
   if ( MyStatus == nullptr) {
      LOG_INFO( GetCmdLogger(), "Error : No Status detected") ;
      return ;
   }
  // Selezione riuscita
   MyStatus->bSelectedDB.store( true, memory_order_release) ;
   MyStatus->bSelectedDB.notify_one() ;
}

//----------------------------------------------------------------------------
// [CB] Funzione di CallBack per comando AUTH Asincrono ( Autenticazione)
// ---------------------------------------------------------------------------
static void
RedisAuthenticationCallBack( redisAsyncContext* ctx, void* r, void*)
{
  // 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 answer" ;
       LOG_INFO( GetCmdLogger(), ( string{ "Error : Authentication -> "} + sErrMsg).c_str()) ;
       return ;
   }
  // Imposto flag di Autenticazione
   RedisConnectionStatus* MyStatus = static_cast<RedisConnectionStatus*>( ctx->data) ;
   if ( MyStatus == nullptr) {
      LOG_INFO( GetCmdLogger(), "Error : No Status detected") ;
      return ;
   }
  // Autenticazione riuscita
   MyStatus->bAuthenticated.store( true, memory_order_release) ;
   MyStatus->bAuthenticated.notify_one() ;
}

// ---------------------------------------------------------------------------
// [CB] Funzione di CallBack per comandi PUBLISH, SUBSCRIBE, MESSAGE, UNSUBSCRIBE
// Asincroni
// ---------------------------------------------------------------------------
static void
RedisPubSubUnsubMsgCallback( redisAsyncContext* ctx, void* r, void*)
{
  // Verifico che il contesto sia definito
   if ( ctx == nullptr) {
      LOG_INFO( GetCmdLogger(), "Error : Redis context null in CallBack")
      return ;
   }

  // Recupero della risposta dal server redis
   redisReply* reply = static_cast<redisReply*>( r) ;
   if ( reply == nullptr)
      return ;

   LOG_INFO( GetCmdLogger(), "   --- RedisMessageCallback invoked")

  // Caso PUBLISH: hiredis risponde con REDIS_REPLY_INTEGER (numero di subscribers)
   if ( reply->type == REDIS_REPLY_INTEGER) {
     // Recupero lo Stato
      RedisConnectionStatus* MyStatus = static_cast<RedisConnectionStatus*>( ctx->data) ;
      if ( MyStatus == nullptr) {
         LOG_INFO( GetCmdLogger(), "Error : No Status detected") ;
         return ;
      }
     // Imposto stato di pubblicazione
      MyStatus->bPublished.store( true, memory_order_release) ;
      MyStatus->bPublished.notify_one() ;
     // Recupero il numero di clients iscritti al canale
      LOG_INFO( GetCmdLogger(), ( string{ "Message published to "} + ToString( reply->integer) + " clients !").c_str()) ;
      return ;
   }
  // Caso ARRAY: Subscribe / Unsubscribe / Message
   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 ricevuto 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, reply->element[1]->len} ;
            string sMessage{ reply->element[2]->str, reply->element[2]->len} ;
            #if DEBUG
               LOG_INFO( GetCmdLogger(), ( string{ "Message Received on ["} + sChannel + "] : " + sMessage).c_str())
            #endif
           // Recupero lo Stato
            RedisConnectionStatus* MyStatus = static_cast<RedisConnectionStatus*>( ctx->data) ;
            if ( MyStatus == nullptr) {
               LOG_INFO( GetCmdLogger(), "Error : No Status detected") ;
               return ;
            }
           // Aggiorno la Mappa dei Messaggi ( viene sempre sovrascritto all'ultimo)
            string sKey = GetMessageMapKey( MyStatus->nIdConnection, sChannel) ;
            auto Iter = s_vAsyncRedisMessages.find( sKey) ;
            if ( Iter != s_vAsyncRedisMessages.end()) {
               RedisAsyncMessage& LockedMessage = Iter->second ;
               while ( LockedMessage.Lock.test_and_set( memory_order_acquire))
                  LockedMessage.Lock.wait( true, memory_order_relaxed) ;
               ++ LockedMessage.nCount ;
               LockedMessage.sMessage = sMessage ;
               LockedMessage.Lock.clear( memory_order_release) ;
               LockedMessage.Lock.notify_one() ;
            }
         }
         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) {
           // Recupero lo Stato
            RedisConnectionStatus* MyStatus = static_cast<RedisConnectionStatus*>( ctx->data) ;
            if ( MyStatus == nullptr) {
               LOG_INFO( GetCmdLogger(), "Error : No Status detected") ;
               return ;
            }
           // Imposto stato di Subscribe
            MyStatus->bSubscribed.store( true, memory_order_release) ;
            MyStatus->bSubscribed.notify_one() ;
           // Comunico il risultato
            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) {
            // Recupero lo Stato
            RedisConnectionStatus* MyStatus = static_cast<RedisConnectionStatus*>( ctx->data) ;
            if ( MyStatus == nullptr) {
               LOG_INFO( GetCmdLogger(), "Error : No Status detected") ;
               return ;
            }
           // Imposto stato Unsubscribe
            MyStatus->bUnsubscribed.store( true, memory_order_release) ;
            MyStatus->bUnsubscribed.notify_one() ;
           // Comunico il risultato
            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 Unsubscription reply")
            return ;
         }
      }
     // --- Undefined
      else {
         LOG_INFO( GetCmdLogger(), "Error : Undefined reply in CallBack from Redis")
         return ;
      }
   }

   return ;
}

// ---------------------------------------------------------------------------
// [CB] Funzione di CallBack per ricezione MESSAGE Asincrona
// ---------------------------------------------------------------------------
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<redisReply*>( r) ;
   if ( reply == nullptr)
      return ;

   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 {
              // Recupero lo Stato
               RedisConnectionStatus* MyStatus = static_cast<RedisConnectionStatus*>( ctx->data) ;
               if ( MyStatus == nullptr) {
                  LOG_INFO( GetCmdLogger(), "Error : No Status detected") ;
                  return ;
               }
              // Imposto stato di pubblicazione
               MyStatus->bMessage = true ;
               MyStatus->sMessage = reply->element[2]->str ;
            }
         }
      }
   }
}

// ---------------------------------------------------------------------------
// [CB] Funzione di CallBack per comando CONNECT Asincrono
// ---------------------------------------------------------------------------
static void
RedisConnectCallBack( const redisAsyncContext* ctx, int nStatus)
{
  // Verifico che il Contesto sia valido
   if ( ctx == nullptr) {
      LOG_INFO( GetCmdLogger(), "Error : Async Connect CallBack failed ") ;
      return ;
   }

  // Recupero dal Contesto lo stato della connessione
   RedisConnectionStatus* MyStatus = static_cast<RedisConnectionStatus*>( ctx->data) ;
   if ( MyStatus == nullptr) {
      LOG_INFO( GetCmdLogger(), "Error : No Status detected") ;
      return ;
   }

  // Verifico che lo stato internno al contesto Redis sia OK
   if ( nStatus != REDIS_OK) {
      LOG_INFO( GetCmdLogger(), "Error : Redis Async connection failed") ;
      MyStatus->bAlive.store( false, memory_order_release) ;
      MyStatus->bConnected.store( false, memory_order_release) ;
      MyStatus->bConnected.notify_one() ;
      return ;
   }

  // Connessione riuscita
   MyStatus->bAlive.store( true, memory_order_release) ;
   MyStatus->bConnected.store( true, memory_order_release) ;
   MyStatus->bConnected.notify_one() ;
}

// ---------------------------------------------------------------------------
// [CB] Funzione di CallBack per comando DISCONNECT Asincrono
// ---------------------------------------------------------------------------
static void
RedisDisconnectCallBack( const redisAsyncContext* ctx, int nStatus)
{
  // Verifico che il Contesto sia valido
   if ( ctx == nullptr) {
      LOG_INFO( GetCmdLogger(), "Error : Null Context for Disconnetion CallBack ") ;
      return ;
   }

  // Recupero lo stato della connessione
   RedisConnectionStatus* MyStatus = static_cast<RedisConnectionStatus*>( ctx->data) ;
   if ( MyStatus == nullptr) {
      LOG_INFO( GetCmdLogger(), "Error : No Status detected") ;
      return ;
   }

  // nStatus non è affidabile in questo contesto, può essere != REDIS_OK anche in disconnessione normale
  // NB. nStatus è solo informativo, ignora tutte le disconnessioni reali
   LOG_INFO( GetCmdLogger(), "Redis Async Disconnected ( x2, called for PUB and SUB)") ;

  // Aggiorno gli stati ed eventuali thread in attesa
   MyStatus->bAlive.store( false, memory_order_release) ;
   MyStatus->bConnected.store( false, memory_order_release) ;
   MyStatus->bConnected.notify_one() ;
   MyStatus->bDisconnected.store( true, memory_order_release) ;
   MyStatus->bDisconnected.notify_one() ;
}

//----------------------------------------------------------------------------
// Funzione per comando CONNECT Asincrono
// ---------------------------------------------------------------------------
bool
ExeRedisAsyncConnect( const string& sConnection, int& nIdConnection)
{
  // Inizializzo una nuova connessione, recuperando l'Id
   int nMyIdConnection = NO_CONNECTION ;
   for ( int i = 0 ; i < int( s_vAsyncRedisClients.size()) ; ++ i) {
      if ( s_vAsyncRedisClients[i].m_bFreeConnection.load( memory_order_acquire)) {
         nMyIdConnection = i ;
         break ;
      }
   }
   if ( nMyIdConnection == NO_CONNECTION) {
      LOG_INFO( GetCmdLogger(), ( string{ "Error : Limit connection Exceeded ( "} +
                                  ToString( MAX_CONNECTION_NBR) + " )").c_str())
      return false ;
   }
  // Recupero lo Slot
   RedisAsync& AsyncRedisClient = s_vAsyncRedisClients[nMyIdConnection] ;
   AsyncRedisClient.Clear() ; // per possibile riconnessione
   AsyncRedisClient.m_bFreeConnection.store( false, memory_order_release) ;

  // Recupero i riferimenti necessari e blocco la connessione corrente
   RedisConnectionInfo& ConnectionInfo = AsyncRedisClient.m_ConnectionInfo ;

  // Recupero i parametri dalla stringa di connessione
   AsyncRedisClient.m_sConnectionString = sConnection ; // per riconnessione
   if ( ! GetParamsFromConnectionString( sConnection, ConnectionInfo)) {
      AsyncRedisClient.Clear() ;
      return false ;
   }

  // Inizializzazione libreria Winsock, necessaria per usare le funzionalità di rete su Windows,
  // come socket TCP/IP. Versione utilizzata 2.2
   int nStatus = WSAStartup( MAKEWORD( 2, 2), &AsyncRedisClient.m_wsaData) ;
   if ( nStatus != 0) {
      LOG_INFO( GetCmdLogger(), ( string{ "Error : WSAStartup -> "} + ToString( nStatus)).c_str()) ;
      AsyncRedisClient.Clear() ;
      return false ;
   }

  // Verifico se si tratta di un nodo Sentinella
   redisContext* sentinelCtx = redisConnect( AsyncRedisClient.m_ConnectionInfo.sHost.c_str(), ConnectionInfo.nPort) ;
   if ( sentinelCtx == nullptr) {
      LOG_INFO( GetCmdLogger(), "Error : Can't allocate redis context")
      AsyncRedisClient.Clear() ;
      return false ;
   }
   else if ( sentinelCtx->err != 0) {
      LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + sentinelCtx->errstr).c_str())
      redisFree( sentinelCtx) ;
      AsyncRedisClient.Clear() ;
      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)
   AsyncRedisClient.m_pRedisAsyncPubContext = redisAsyncConnect( ConnectionInfo.sHost.c_str(), ConnectionInfo.nPort) ;
   AsyncRedisClient.m_pRedisAsyncSubContext = redisAsyncConnect( ConnectionInfo.sHost.c_str(), ConnectionInfo.nPort) ;
   if ( AsyncRedisClient.m_pRedisAsyncPubContext == nullptr  ||  AsyncRedisClient.m_pRedisAsyncSubContext == nullptr) {
      LOG_INFO( GetCmdLogger(), "Error : Can't allocate Redis Async context")
      AsyncRedisClient.Clear() ;
      return false ;
   }
   else if ( AsyncRedisClient.m_pRedisAsyncPubContext->err != 0) {
      LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + AsyncRedisClient.m_pRedisAsyncPubContext->errstr).c_str())
      AsyncRedisClient.Clear() ;
      return false ;
   }
   else if ( AsyncRedisClient.m_pRedisAsyncSubContext->err != 0) {
      LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + AsyncRedisClient.m_pRedisAsyncSubContext->errstr).c_str())
      AsyncRedisClient.Clear() ;
      return false ;
   }

  // Definizione e partenza del ciclo degli eventi in un thread separato ( in BackGround)
   AsyncRedisClient.m_ConnectionPubStatus.Clear() ;
   AsyncRedisClient.m_ConnectionSubStatus.Clear() ;
   AsyncRedisClient.m_bPubLoopRunning.store( true, memory_order_release) ;
   AsyncRedisClient.m_bSubLoopRunning.store( true, memory_order_release) ;
   AsyncRedisClient.m_PubThread = thread( &RedisAsync::RedisEventLoop, &AsyncRedisClient, true) ;
   AsyncRedisClient.m_SubThread = thread( &RedisAsync::RedisEventLoop, &AsyncRedisClient, false) ;

  // Imposto e Definisco la funzione di CallBack per contesto PUB/SUB per Connessione/Disconnessione asincrona
   AsyncRedisClient.m_pRedisAsyncPubContext->data = &AsyncRedisClient.m_ConnectionPubStatus ;
   AsyncRedisClient.m_pRedisAsyncSubContext->data = &AsyncRedisClient.m_ConnectionSubStatus ;
   AsyncRedisClient.m_ConnectionPubStatus.bConnected.store( false, memory_order_release) ;
   AsyncRedisClient.m_ConnectionSubStatus.bConnected.store( false, memory_order_release) ;
   redisAsyncSetConnectCallback( AsyncRedisClient.m_pRedisAsyncPubContext, RedisConnectCallBack) ;
   redisAsyncSetConnectCallback( AsyncRedisClient.m_pRedisAsyncSubContext, RedisConnectCallBack) ;
   redisAsyncSetDisconnectCallback( AsyncRedisClient.m_pRedisAsyncPubContext, RedisDisconnectCallBack) ;
   redisAsyncSetDisconnectCallback( AsyncRedisClient.m_pRedisAsyncSubContext, RedisDisconnectCallBack) ;
  // Attendo la risposta di CallBack
   auto tStart = chrono::steady_clock::now() ;
   auto tTimeOut = chrono::milliseconds( static_cast<int>( ConnectionInfo.nAsyncTimeout)) ;
   while ( ! AsyncRedisClient.m_ConnectionPubStatus.bConnected.load( memory_order_acquire)) {
      if ( chrono::steady_clock::now() - tStart > tTimeOut) {
         LOG_INFO( GetCmdLogger(), "Timeout Connection exceeded") ;
         AsyncRedisClient.Clear() ;
         return false ;
      }
     // NB. non usare la wait( false) !!! se in rete perdo qualche informazione
     //     rimango bloccato all'infinito, meglio usare un timer, meno efficiente ma sicuro
      this_thread::sleep_for( chrono::milliseconds( 10)) ;
   }
   while ( ! AsyncRedisClient.m_ConnectionSubStatus.bConnected.load( memory_order_acquire)) {
      if ( chrono::steady_clock::now() - tStart > tTimeOut) {
         LOG_INFO( GetCmdLogger(), "Timeout Connection exceeded") ;
         AsyncRedisClient.Clear() ;
         return false ;
      }
     // NB. non usare la wait( false) !!! se in rete perdo qualche informazione
     //     rimango bloccato all'infinito, meglio usare un timer, meno efficiente ma sicuro
      this_thread::sleep_for( chrono::milliseconds( 10)) ;
   }
   LOG_INFO( GetCmdLogger(), "Connected to Redis !")
   #if DEBUG
      double dMsConnectionTime = chrono::duration_cast<chrono::milliseconds>( chrono::steady_clock::now() - tStart).count() ;
      LOG_INFO( GetCmdLogger(), ( string{ "  + "} + ToString( dMsConnectionTime) + " ms").c_str())
   #endif

  // Se presente una Password, eseguo l'autenticazione
   if ( ! AsyncRedisClient.m_sPassword.empty()) {
      AsyncRedisClient.m_ConnectionPubStatus.bAuthenticated.store( false, memory_order_relaxed) ;
      AsyncRedisClient.m_ConnectionSubStatus.bAuthenticated.store( false, memory_order_relaxed) ;
      redisAsyncCommand( AsyncRedisClient.m_pRedisAsyncPubContext, RedisAuthenticationCallBack,
                         nullptr, "AUTH %s %s", AsyncRedisClient.m_sUser.c_str(), AsyncRedisClient.m_sPassword.c_str()) ;
      redisAsyncCommand( AsyncRedisClient.m_pRedisAsyncSubContext, RedisAuthenticationCallBack,
                         nullptr, "AUTH %s %s", AsyncRedisClient.m_sUser.c_str(), AsyncRedisClient.m_sPassword.c_str()) ;
      tStart = chrono::steady_clock::now() ;
      tTimeOut = chrono::milliseconds( static_cast<int>( ConnectionInfo.nAsyncTimeout)) ;
      while ( ! AsyncRedisClient.m_ConnectionPubStatus.bAuthenticated.load( memory_order_acquire)) {
         if ( chrono::steady_clock::now() - tStart > tTimeOut) {
            LOG_INFO( GetCmdLogger(), "Timeout Autentication exceeded") ;
            AsyncRedisClient.Clear() ;
            return false ;
         }
        // NB. non usare la wait( false) !!! se in rete perdo qualche informazione
        //     rimango bloccato all'infinito, meglio usare un timer, meno efficiente ma sicuro
         this_thread::sleep_for( chrono::milliseconds( 10)) ;
      }
      while ( ! AsyncRedisClient.m_ConnectionSubStatus.bAuthenticated.load( memory_order_acquire)) {
         if ( chrono::steady_clock::now() - tStart > tTimeOut) {
            LOG_INFO( GetCmdLogger(), "Timeout Autentication exceeded") ;
            AsyncRedisClient.Clear() ;
            return false ;
         }
        // NB. non usare la wait( false) !!! se in rete perdo qualche informazione
        //     rimango bloccato all'infinito, meglio usare un timer, meno efficiente ma sicuro
         this_thread::sleep_for( chrono::milliseconds( 10)) ;
      }
      LOG_INFO( GetCmdLogger(), "Authenticated to Redis !")
      #if DEBUG
         double dMsAuthenticationTime = chrono::duration_cast<chrono::milliseconds>( chrono::steady_clock::now() - tStart).count() ;
         LOG_INFO( GetCmdLogger(), ( string{ "  + "} + ToString( dMsAuthenticationTime) + " ms").c_str())
      #endif
   }
   else {
      AsyncRedisClient.m_ConnectionPubStatus.bAuthenticated.store( true, memory_order_release) ;
      AsyncRedisClient.m_ConnectionSubStatus.bAuthenticated.store( true, memory_order_release) ;
   }

  // Se Presente una Selezione del Database, eseguo la selezione
   if ( AsyncRedisClient.m_nDataBase != DEFAULT_DATABASE) {
      AsyncRedisClient.m_ConnectionPubStatus.bSelectedDB.store( false, memory_order_relaxed) ;
      AsyncRedisClient.m_ConnectionSubStatus.bSelectedDB.store( false, memory_order_relaxed) ;
      redisAsyncCommand( AsyncRedisClient.m_pRedisAsyncPubContext, RedisDbSElectionCallBack,
                         nullptr, "SELECT %d", AsyncRedisClient.m_nDataBase) ;
      redisAsyncCommand( AsyncRedisClient.m_pRedisAsyncSubContext, RedisDbSElectionCallBack,
                         nullptr, "SELECT %d", AsyncRedisClient.m_nDataBase) ;
      tStart = chrono::steady_clock::now() ;
      tTimeOut = chrono::milliseconds( static_cast<int>( ConnectionInfo.nAsyncTimeout)) ;
      while ( ! AsyncRedisClient.m_ConnectionPubStatus.bSelectedDB.load( memory_order_acquire)) {
         if ( chrono::steady_clock::now() - tStart > tTimeOut) {
            LOG_INFO( GetCmdLogger(), "Timeout DB Selection exceeded") ;
            AsyncRedisClient.Clear() ;
            return false ;
         }
        // NB. non usare la wait( false) !!! se in rete perdo qualche informazione
        //     rimango bloccato all'infinito, meglio usare un timer, meno efficiente ma sicuro
         this_thread::sleep_for( chrono::milliseconds( 10)) ;
      }
      while ( ! AsyncRedisClient.m_ConnectionSubStatus.bSelectedDB.load( memory_order_acquire)) {
         if ( chrono::steady_clock::now() - tStart > tTimeOut) {
            LOG_INFO( GetCmdLogger(), "Timeout DB Selection exceeded") ;
            AsyncRedisClient.Clear() ;
            return false ;
         }
        // NB. non usare la wait( false) !!! se in rete perdo qualche informazione
        //     rimango bloccato all'infinito, meglio usare un timer, meno efficiente ma sicuro
         this_thread::sleep_for( chrono::milliseconds( 10)) ;
      }
      LOG_INFO( GetCmdLogger(), "DB Selected to Redis !") ;
      #if DEBUG
         double dMsDbSelectionTime = chrono::duration_cast<chrono::milliseconds>( chrono::steady_clock::now() - tStart).count() ;
         LOG_INFO( GetCmdLogger(), ( string{ "  + "} + ToString( dMsDbSelectionTime) + " ms").c_str())
      #endif
   }
   else {
      AsyncRedisClient.m_ConnectionPubStatus.bSelectedDB.store( true, memory_order_release) ;
      AsyncRedisClient.m_ConnectionSubStatus.bSelectedDB.store( true, memory_order_release) ;
   }

  // Restituisco e salvo l'Id della connessione [1, MAX_CONNECTION_NBR] ( per Lua)
   nIdConnection = nMyIdConnection + 1 ;
   AsyncRedisClient.m_ConnectionPubStatus.nIdConnection = nIdConnection ;
   AsyncRedisClient.m_ConnectionSubStatus.nIdConnection = nIdConnection ;
   return true ;
}

//----------------------------------------------------------------------------
// Funzione per comando DISCONNECT Asincrono
// ---------------------------------------------------------------------------
bool
ExeRedisAsyncDisconnect( int nIdConnection)
{
  // Verifico che l'Id di connesione sia valido
   int nMyIdConnection = nIdConnection - 1 ;
   if ( ! CheckIdConnection( nMyIdConnection))
      return false ;

  // Recupero la connessione
   RedisAsync& AsyncRedisClient = s_vAsyncRedisClients[nMyIdConnection] ;

  // Verifico che la connessione sia utilizzata, e non libera
   if ( s_vAsyncRedisClients[nMyIdConnection].m_bFreeConnection.load( memory_order_acquire)) {
      LOG_INFO( GetCmdLogger(), "Error : Id Connection not found")
      return false ;
   }

  // Reset dei Flag prima della connessione
   AsyncRedisClient.m_ConnectionPubStatus.bDisconnected.store( false, memory_order_relaxed) ;
   AsyncRedisClient.m_ConnectionSubStatus.bDisconnected.store( false, memory_order_relaxed) ;
   AsyncRedisClient.m_ConnectionPubStatus.bAlive.store( false, memory_order_relaxed) ;
   AsyncRedisClient.m_ConnectionSubStatus.bAlive.store( false, memory_order_relaxed) ;

  // Effettuo la Disconnesione asincrona ( libera la memoria in automatico)
   redisAsyncDisconnect( AsyncRedisClient.m_pRedisAsyncPubContext) ;
   redisAsyncDisconnect( AsyncRedisClient.m_pRedisAsyncSubContext) ;

  // Attendo la CallBack di disconnessione
   auto tStart = chrono::steady_clock::now() ;
   auto tTimeOut = chrono::milliseconds( static_cast<int>( AsyncRedisClient.m_ConnectionInfo.nAsyncTimeout)) ;
   while ( ! AsyncRedisClient.m_ConnectionPubStatus.bDisconnected.load( memory_order_acquire)) {
      AsyncRedisClient.m_ConnectionPubStatus.bDisconnected.wait( false) ;
      if ( chrono::steady_clock::now() - tStart > tTimeOut) {
         LOG_INFO( GetCmdLogger(), "Timeout Connection exceeded") ;
         return false ;
      }
   }
   while ( ! AsyncRedisClient.m_ConnectionSubStatus.bDisconnected.load( memory_order_acquire)) {
      AsyncRedisClient.m_ConnectionSubStatus.bDisconnected.wait( false) ;
      if ( chrono::steady_clock::now() - tStart > tTimeOut) {
         LOG_INFO( GetCmdLogger(), "Timeout Connection exceeded") ;
         return false ;
      }
   }

  // Rendo libera la connessione
   AsyncRedisClient.m_bFreeConnection.store( true) ;
   AsyncRedisClient.Clear() ;

  // Elimino i Messaggi dalla mappa
   for ( auto Iter = s_vAsyncRedisMessages.begin() ; Iter != s_vAsyncRedisMessages.end() ; ) {
      if ( Iter->first.starts_with( GetMessageMapKey( nIdConnection, "")))
         Iter = s_vAsyncRedisMessages.erase( Iter) ;
      else
         ++ Iter ;
   }

   LOG_INFO( GetCmdLogger(), "Disconnected to Redis !") ;
   #if DEBUG
      double dMsDisconnectionTime = chrono::duration_cast<chrono::milliseconds>( chrono::steady_clock::now() - tStart).count() ;
      LOG_INFO( GetCmdLogger(), ( string{ "  + "} + ToString( dMsDisconnectionTime) + " ms").c_str())
   #endif
   return true ;
}

// ---------------------------------------------------------------------------
// Funzione per comando PUBLISH Asincrono
// ---------------------------------------------------------------------------
bool
ExeRedisAsyncPublish( int nIdConnection, const string& sChannel, const string& sMessage)
{
  // Verifico che l'Id di connesione sia valido
   int nMyIdConnection = nIdConnection - 1 ;
   if ( ! CheckIdConnection( nMyIdConnection))
      return false ;

  // Verifico che la connessione sia attiva
   RedisAsync& AsyncRedisClient = s_vAsyncRedisClients[nMyIdConnection] ;
   if ( AsyncRedisClient.m_bFreeConnection.load( memory_order_acquire)) {
      LOG_INFO( GetCmdLogger(), "Error : Id Connection not found")
      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") ;

  // Numero massimo di Tentativi per effettuare PUBLISH
   const int MAX_RETRY = 3 ;
   for ( int nAttempt = 0 ; nAttempt < MAX_RETRY ; ++ nAttempt) {

     // Rileggo il Client, dato che potrebbe essere cambiato dopo la riconnessione
      RedisAsync& currClient = s_vAsyncRedisClients[nMyIdConnection] ;
      RedisConnectionStatus& curStatus = currClient.m_ConnectionPubStatus ;

     // Se la connessione non è attiva, provo a riconnetermi
      if ( ! currClient.m_ConnectionPubStatus.bAlive.load( memory_order_acquire)) {
         LOG_INFO( GetCmdLogger(), ( string{ "Publish: connection not alive, reconnect attempt "} + ToString( nAttempt)).c_str()) ;

        // Recupero la stringa di connessione corrente
         string sConnStr = s_vAsyncRedisClients[nMyIdConnection].m_sConnectionString ;
        // Recupero i Canali su cui ero iscritto
         unordered_set<string> setSubChannels = currClient.m_set_SubChannels ;

        // Pulisco il contesto corrente
         AsyncRedisClient.Clear() ;

         int nNewId = 0 ;
         bool bOkReconnect = false ;
         if ( ExeRedisAsyncConnect( sConnStr, nNewId)) {
           // Se ho scelto lo stesso slot, aggiorno nMyIdConnection
            int nNewIndex = nNewId - 1 ;
            if ( nNewIndex >= 0 && nNewIndex < ssize( s_vAsyncRedisClients)) {
              // Sostituisco il riferimento Client con il nuovo
               if ( nNewIndex != nMyIdConnection)
                  nMyIdConnection = nNewIndex ;
            }
            bOkReconnect = true ;
           // Mi iscrivo a tutti i canali a cui ero precedentemente iscritto
            for ( auto Iter = setSubChannels.begin() ; Iter != setSubChannels.end() ; ++ Iter) {
               if ( ! ExeRedisAsyncSubscribe( nMyIdConnection + 1, *Iter))
                  return false ;
            }
         }
         else {
            LOG_INFO( GetCmdLogger(), "Publish: reconnect attempt failed") ;
           // riprovo al ciclo successivo...
         }
        // Se non mi sono riconesso metto un piccolo Delay per estrema sicureza
         if ( ! bOkReconnect) {
            this_thread::sleep_for( chrono::milliseconds( 100)) ;
            continue ;
         }
      }

     // Rileggo il Clinet e lo Stato ( in caso di eventuale Riconnessione)
      RedisAsync& nextClient = s_vAsyncRedisClients[nMyIdConnection] ;
      RedisConnectionStatus& nextStatus = nextClient.m_ConnectionPubStatus ;

     // Ricalcolo timeout dal client aggiornato
      auto tTimeOut = chrono::milliseconds( static_cast<int>( nextClient.m_ConnectionInfo.nAsyncTimeout)) ;

     // Reset dello stato di pubblicazione
      nextStatus.bPublished.store( false, memory_order_relaxed) ;

     // Invio del comando PUBLISH
      int nResult = redisAsyncCommand( nextClient.m_pRedisAsyncPubContext, RedisPubSubUnsubMsgCallback, nullptr,
                                       "PUBLISH %s %s", sChannel.c_str(), sMessage.c_str()) ;
      if ( nResult != REDIS_OK) {
         LOG_INFO( GetCmdLogger(), "Publish: redisAsyncCommand failed") ;
         nextStatus.bAlive.store( false, memory_order_release) ;
         this_thread::sleep_for( chrono::milliseconds( 100)) ;
         continue ;
      }
      auto tStart = chrono::steady_clock::now() ;
      while ( ! nextStatus.bPublished.load( memory_order_acquire)) {
         if ( chrono::steady_clock::now() - tStart > tTimeOut) {
            LOG_INFO( GetCmdLogger(), "Publish: timeout waiting for publish reply") ;
            break ;
         }
        // NB. non usare la wait( false) !!! se in rete perdo qualche informazione
        //     rimango bloccato all'infinito, meglio usare un timer, meno efficiente ma sicuro
         this_thread::sleep_for( chrono::milliseconds( 10)) ;
      }
      if ( nextStatus.bPublished.load( memory_order_acquire)) {
         #if DEBUG
            double dMsPublishTime = chrono::duration_cast<chrono::milliseconds>( chrono::steady_clock::now() - tStart).count() ;
            LOG_INFO( GetCmdLogger(), ( string{" + "} + ToString( dMsPublishTime) + " ms").c_str()) ;
         #endif
         return true ;
      }
      LOG_INFO( GetCmdLogger(), ( string{ "Publish failed, attempt "} + ToString( nAttempt)).c_str()) ;

     // Breve pausa prima del Retry
      this_thread::sleep_for( chrono::milliseconds( 100)) ;
   }

  // Tutti i tentativi sono falliti
   return false ;
}

//----------------------------------------------------------------------------
// Funzione per comando SUBSCRIBE Asincrono
// ---------------------------------------------------------------------------
bool
ExeRedisAsyncSubscribe( int nIdConnection, const string& sChannel)
{
  // Verifico che l'Id di connesione sia valido
   int nMyIdConnection = nIdConnection - 1 ;
   if ( ! CheckIdConnection( nMyIdConnection))
      return false ;
  // Verifico che la connessione sia attiva
   RedisAsync& AsyncRedisClient = s_vAsyncRedisClients[nMyIdConnection] ;
   if ( AsyncRedisClient.m_bFreeConnection.load( memory_order_acquire)) {
      LOG_INFO( GetCmdLogger(), "Error : Id Connection not found")
      return false ;
   }
  // Se non ho alcun canale, errore
   if ( sChannel.empty()) {
      LOG_INFO( GetCmdLogger(), "Error : Not a valid Channel") ;
      return false ;
   }

  // Numero massimo di Tentativi per effettuare SUBSCRIBE
   const int MAX_RETRY = 3 ;
   for ( int nAttempt = 0 ; nAttempt < MAX_RETRY ; ++ nAttempt) {

     // Rileggo il client e lo stato (potrebbero cambiare dopo reconnect)
      RedisAsync& currClient = s_vAsyncRedisClients[nMyIdConnection] ;
      RedisConnectionStatus& currStatus = currClient.m_ConnectionSubStatus ;

     // Se la connessione non è viva, provo a riconnettere
      if ( ! currStatus.bAlive.load( memory_order_acquire)) {
         LOG_INFO( GetCmdLogger(), ( string{ "Subscribe: connection not alive, reconnect attempt "} + ToString( nAttempt)).c_str()) ;

        // Recupero la stringa di connessione corrente
         string sConnStr = currClient.m_sConnectionString ;
        // Recupero i Canali su cui ero iscritto
         unordered_set<string> setSubChannels = currClient.m_set_SubChannels ;

        // Pulisco il contesto corrente
         AsyncRedisClient.Clear() ;

        int nNewId = 0 ;
         bool bOkReconnect = false ;
         if ( ExeRedisAsyncConnect( sConnStr, nNewId)) {
           // Se ho scelto lo stesso slot, aggiorno nMyIdConnection
            int nNewIndex = nNewId - 1 ;
            if ( nNewIndex >= 0 && nNewIndex < ssize( s_vAsyncRedisClients)) {
              // Sostituisco il riferimento Client con il nuovo
               if ( nNewIndex != nMyIdConnection)
                  nMyIdConnection = nNewIndex ;
            }
            bOkReconnect = true ;
           // Mi iscrivo a tutti i canali a cui ero precedentemente iscritto
            for ( auto Iter = setSubChannels.begin() ; Iter != setSubChannels.end() ; ++ Iter) {
               if ( ! ExeRedisAsyncSubscribe( nMyIdConnection + 1, *Iter))
                  return false ;
            }
         }
         else {
            LOG_INFO( GetCmdLogger(), "Publish: reconnect attempt failed") ;
           // riprovo al ciclo successivo...
         }
        // Se non mi sono riconesso metto un piccolo Delay per estrema sicureza
         if ( ! bOkReconnect) {
            this_thread::sleep_for( chrono::milliseconds( 100)) ;
            continue ;
         }
      }

     // Rileggo il Clinet e lo Stato ( in caso di eventuale Riconnessione)
      RedisAsync& nextClient = s_vAsyncRedisClients[nMyIdConnection] ;
      RedisConnectionStatus& nextStatus = nextClient.m_ConnectionSubStatus ;

     // Ricalcolo timeout dal client aggiornato
      auto tTimeOut = chrono::milliseconds( static_cast<int>( nextClient.m_ConnectionInfo.nAsyncTimeout)) ;

     // Reset dello stato di pubblicazione
      nextStatus.bSubscribed.store( false, memory_order_relaxed) ;

      int nResult = redisAsyncCommand( nextClient.m_pRedisAsyncSubContext, RedisPubSubUnsubMsgCallback, nullptr,
                                       "SUBSCRIBE %s", sChannel.c_str()) ;
      if ( nResult != REDIS_OK) {
         LOG_INFO( GetCmdLogger(), "Subscribe: redisAsyncCommand failed") ;
         nextStatus.bAlive.store( false, memory_order_release) ;
         this_thread::sleep_for( chrono::milliseconds( 100)) ;
         continue ;
      }
      auto tStart = chrono::steady_clock::now() ;
      while ( ! nextStatus.bSubscribed.load( memory_order_acquire)) {
         if ( chrono::steady_clock::now() - tStart > tTimeOut) {
            LOG_INFO( GetCmdLogger(), "Publish: timeout waiting for publish reply") ;
            break ;
         }
        // NB. non usare la wait( false) !!! se in rete perdo qualche informazione
        //     rimango bloccato all'infinito, meglio usare un timer, meno efficiente ma sicuro
         this_thread::sleep_for( chrono::milliseconds( 10)) ;
      }

      if ( nextStatus.bSubscribed.load( memory_order_acquire)) {
        // Uso l'ID corrente ( aggiornato) per la chiave messaggi
         int nEffectiveId = nMyIdConnection + 1 ;
         string sKey = GetMessageMapKey( nEffectiveId, sChannel) ;
         s_vAsyncRedisMessages.try_emplace( sKey) ;
         s_vAsyncRedisMessages[sKey].Lock.clear() ;
         #if DEBUG
            double dMsPublishTime = chrono::duration_cast<chrono::milliseconds>( chrono::steady_clock::now() - tStart).count() ;
            LOG_INFO( GetCmdLogger(), ( string{" + "} + ToString( dMsPublishTime) + " ms").c_str()) ;
         #endif
        // Memorizzo il canale nella set dei canali
         nextClient.m_set_SubChannels.insert( sChannel) ;
         return true ;
      }
      LOG_INFO( GetCmdLogger(), ( string{ "Subscribe failed, attempt "} + ToString( nAttempt)).c_str()) ;

     // Breve pausa prima del Retry
      this_thread::sleep_for( chrono::milliseconds( 100)) ;
   }

  // Tutti i tentativi sono falliti
   return false ;
}

// ---------------------------------------------------------------------------
// Funzione per comando UNSUBSCRIBE Asincrono
// ---------------------------------------------------------------------------
bool
ExeRedisAsyncUnsubscribe( int nIdConnection, const string& sChannel)
{
  // Verifico che l'Id di connesione sia valido
   int nMyIdConnection = nIdConnection - 1 ;
   if ( ! CheckIdConnection( nMyIdConnection))
      return false ;
  // Verifico che la connessione sia attiva
   RedisAsync& AsyncRedisClient = s_vAsyncRedisClients[nMyIdConnection] ;
   if ( s_vAsyncRedisClients[nMyIdConnection].m_bFreeConnection.load( memory_order_acquire)) {
      LOG_INFO( GetCmdLogger(), "Error : Id Connection not found")
      return false ;
   }
  // Se non ho alcun canale, errore
   if ( sChannel.empty()) {
      LOG_INFO( GetCmdLogger(), "Error : Not a valid Channel")
      return false ;
   }

   const int MAX_RETRY = 3 ;
   for ( int nAttempt = 0 ; nAttempt < MAX_RETRY ; ++ nAttempt) {

     // Rileggo il client e lo stato ( potrebbero cambiare dopo reconnect)
      RedisAsync& currClient = s_vAsyncRedisClients[nMyIdConnection] ;
      RedisConnectionStatus& currStatus = currClient.m_ConnectionSubStatus ;

     // Se la connessione non è viva, provo a riconnettere
      if ( ! currStatus.bAlive.load( memory_order_acquire)) {
         LOG_INFO( GetCmdLogger(), ( string{ "Unubscribe: connection not alive, reconnect attempt "} + ToString( nAttempt)).c_str()) ;

        // Recupero la stringa di connessione corrente
         string sConnStr = currClient.m_sConnectionString ;
        // Recupero i Canali su cui ero iscritto
         unordered_set<string> setSubChannels = currClient.m_set_SubChannels ;

        // Pulisco il contesto corrente
         AsyncRedisClient.Clear() ;

         int nNewId = 0 ;
         bool bOkReconnect = false ;
         if ( ExeRedisAsyncConnect( sConnStr, nNewId)) {
           // Se ho scelto lo stesso slot, aggiorno nMyIdConnection
            int nNewIndex = nNewId - 1 ;
            if ( nNewIndex >= 0 && nNewIndex < ssize( s_vAsyncRedisClients)) {
              // Sostituisco il riferimento Client con il nuovo
               if ( nNewIndex != nMyIdConnection)
                  nMyIdConnection = nNewIndex ;
            }
            bOkReconnect = true ;
           // Mi iscrivo a tutti i canali a cui ero precedentemente iscritto
            for ( auto Iter = setSubChannels.begin() ; Iter != setSubChannels.end() ; ++ Iter) {
               if ( ! ExeRedisAsyncSubscribe( nMyIdConnection + 1, *Iter))
                  return false ;
            }
         }
         else {
            LOG_INFO( GetCmdLogger(), "Unsubscribe: reconnect attempt failed") ;
           // riprovo al ciclo successivo...
         }
        // Se non mi sono riconesso metto un piccolo Delay per estrema sicureza
         if ( ! bOkReconnect) {
            this_thread::sleep_for( chrono::milliseconds( 100)) ;
            continue ;
         }
      }

     // Rileggo il Clinet e lo Stato ( in caso di eventuale Riconnessione)
      RedisAsync& nextClient = s_vAsyncRedisClients[nMyIdConnection] ;
      RedisConnectionStatus& nextStatus = nextClient.m_ConnectionSubStatus ;

     // Ricalcolo timeout dal client aggiornato
      auto tTimeOut = chrono::milliseconds( static_cast<int>( nextClient.m_ConnectionInfo.nAsyncTimeout)) ;

     // Reset dello stato di pubblicazione
      nextStatus.bUnsubscribed.store( false, memory_order_relaxed) ;

      int nResult = redisAsyncCommand( nextClient.m_pRedisAsyncSubContext, RedisPubSubUnsubMsgCallback, nullptr,
                                       "UNSUBSCRIBE %s", sChannel.c_str()) ;
      if ( nResult != REDIS_OK) {
         LOG_INFO( GetCmdLogger(), "Error : redisAsyncCommand UNSUBSCRIBE failed") ;
         nextStatus.bAlive.store( false, memory_order_release) ;
         this_thread::sleep_for( chrono::milliseconds( 100)) ;
         continue ;
      }
      auto tStart = chrono::steady_clock::now() ;
      while ( ! nextStatus.bUnsubscribed.load( memory_order_acquire)) {
         if ( chrono::steady_clock::now() - tStart > tTimeOut) {
            LOG_INFO( GetCmdLogger(), "Unsubscribe: timeout waiting for publish reply") ;
            break ;
         }
        // NB. non usare la wait( false) !!! se in rete perdo qualche informazione
        //     rimango bloccato all'infinito, meglio usare un timer, meno efficiente ma sicuro
         this_thread::sleep_for( chrono::milliseconds( 10)) ;
      }

      if ( nextStatus.bUnsubscribed.load( memory_order_acquire)) {
        // Rimuovo la chiave dalla mappa messaggi usando l'ID effettivo (aggiornato)
         int nEffectiveId = nMyIdConnection + 1 ;
         string keyToRemove = GetMessageMapKey( nEffectiveId, sChannel) ;
         for ( auto it = s_vAsyncRedisMessages.begin() ; it != s_vAsyncRedisMessages.end() ; ) {
            if ( it->first == keyToRemove)
               it = s_vAsyncRedisMessages.erase( it) ;
            else
               ++ it ;
         }
         #if DEBUG
            double dMsUnsubscribeTime = chrono::duration_cast<chrono::milliseconds>( chrono::steady_clock::now() - tStart).count() ;
            LOG_INFO( GetCmdLogger(), ( string{" + "} + ToString( dMsUnsubscribeTime) + " ms").c_str()) ;
         #endif
        // Tolgo il canale dalla lista
         nextClient.m_set_SubChannels.erase( sChannel) ;
         return true ;
      }
      LOG_INFO( GetCmdLogger(), ( string{ "Unsubscribe failed, attempt "} + ToString( nAttempt)).c_str()) ;

     // Breve pausa prima del Retry
      this_thread::sleep_for( chrono::milliseconds( 100)) ;
   }

  // Tutti i tentativi sono falliti
   return false ;
}

// ---------------------------------------------------------------------------
// La funzione esegue una Subscribe e una Unsubscribe
// Se entro il tempo stabilito riceve un messaggio, lo comunica, altrimenti errore
// ---------------------------------------------------------------------------
bool
ExeRedisAsyncSubscribeOneMessage( int nIdConnection, const string& sChannel, double dMaxTimeOut,
                                  string& sMessage)
{
  // Verifico che l'Id di connesione sia valido
   int nMyIdConnection = nIdConnection - 1 ;
   if ( ! CheckIdConnection( nMyIdConnection))
      return false ;
  // Verifico che la connessione sia attiva
   if ( s_vAsyncRedisClients[nMyIdConnection].m_bFreeConnection) {
      LOG_INFO( GetCmdLogger(), "Error : Id Connection not found")
      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 ;
   }

  // Recupero la connessione
   RedisAsync& AsyncRedisClient = s_vAsyncRedisClients[nMyIdConnection] ;

  // Invio del comando di SUBSCRIBE
   AsyncRedisClient.m_ConnectionSubStatus.bMessage = false ;
   AsyncRedisClient.m_ConnectionSubStatus.sMessage = "" ;
   if ( redisAsyncCommand( AsyncRedisClient.m_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
   auto tStart = chrono::steady_clock::now() ;
   auto tTimeOut = chrono::milliseconds( static_cast<int>( dMaxTimeOut)) ;
   while ( ! AsyncRedisClient.m_ConnectionSubStatus.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
   if ( redisAsyncCommand( AsyncRedisClient.m_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 ( AsyncRedisClient.m_ConnectionSubStatus.bMessage) {
      sMessage = AsyncRedisClient.m_ConnectionSubStatus.sMessage ;
      LOG_INFO( GetCmdLogger(), ( string{ "Message received on ["} + sChannel + "]: " + sMessage).c_str())
      return true ;
   }

   #if DEBUG
      LOG_INFO( GetCmdLogger(), ( string{ "Timeout: No message received on ["} + sChannel + "]").c_str()) ;
   #endif
   return false ;
}

// ---------------------------------------------------------------------------
// Funzione per lettura Asicnrona dell'ultimo messaggio su un canale
// ---------------------------------------------------------------------------
bool
ExeRedisAsyncGetMessage( int nIdConnection, const string& sChannel, int& nCount, string& sMessage)
{
  // Verifico che l'Id di connesione sia valido
   int nMyIdConnection = nIdConnection - 1 ;
   if ( ! CheckIdConnection( nMyIdConnection))
      return false ;
  // Verifico che la connessione sia attiva
   if ( s_vAsyncRedisClients[nMyIdConnection].m_bFreeConnection) {
      LOG_INFO( GetCmdLogger(), "Error : Id Connection not found")
      return false ;
   }
  // Se non ho alcun canale, errore
   if ( sChannel.empty()) {
      LOG_INFO( GetCmdLogger(), "Error : Not a valid Channel")
      return false ;
   }

  // Recupero la Chiave di accesso alla Mappa dei Messaggi
   string sKey = GetMessageMapKey( nIdConnection, sChannel) ;
   auto Iter = s_vAsyncRedisMessages.find( sKey) ;
   if ( Iter == s_vAsyncRedisMessages.end()) {
      LOG_INFO( GetCmdLogger(), "Error : Connection not Subscribed to this Channel") ;
      return false ;
   }
   else {
      RedisAsyncMessage& LockedMessage = Iter->second ;
      while ( LockedMessage.Lock.test_and_set( memory_order_acquire))
         LockedMessage.Lock.wait( true, memory_order_relaxed) ;
      nCount = LockedMessage.nCount ;
      sMessage = LockedMessage.sMessage ;
      LockedMessage.Lock.clear( memory_order_release) ;
      LockedMessage.Lock.notify_one() ;
   }

   return true ;
}