HAClientImpl.hpp

//
// Copyright (c) 2010-2020 60East Technologies Inc., All Rights Reserved.
//
// This computer software is owned by 60East Technologies Inc. and is
// protected by U.S. copyright laws and other laws and by international
// treaties.  This computer software is furnished by 60East Technologies
// Inc. pursuant to a written license agreement and may be used, copied,
// transmitted, and stored only in accordance with the terms of such
// license agreement and with the inclusion of the above copyright notice.
// This computer software or any other copies thereof may not be provided
// or otherwise made available to any other person.
//
// U.S. Government Restricted Rights.  This computer software: (a) was
// developed at private expense and is in all respects the proprietary
// information of 60East Technologies Inc.; (b) was not developed with
// government funds; (c) is a trade secret of 60East Technologies Inc.
// for all purposes of the Freedom of Information Act; and (d) is a
// commercial item and thus, pursuant to Section 12.212 of the Federal
// Acquisition Regulations (FAR) and DFAR Supplement Section 227.7202,
// Government's use, duplication or disclosure of the computer software
// is subject to the restrictions set forth by 60East Technologies Inc..
//

#ifndef _HACLIENTIMPL_H_
#define _HACLIENTIMPL_H_

#include <typeinfo>
#include <ampsplusplus.hpp>
#include <ServerChooser.hpp>
#include <MemorySubscriptionManager.hpp>
#include <ReconnectDelayStrategy.hpp>

namespace AMPS
{

class HAClientImpl : public ClientImpl
{
public:
    HAClientImpl(const std::string& name_)
        : ClientImpl(name_), _timeout(AMPS_HACLIENT_TIMEOUT_DEFAULT)
        , _reconnectDelay(AMPS_HACLIENT_RECONNECT_DEFAULT)
        , _reconnectDelayStrategy(new ExponentialDelayStrategy(_reconnectDelay))
        , _disconnected(false)
    {
#ifdef AMPS_USE_FUNCTIONAL
        setDisconnectHandler(HADisconnectHandler());
#else
        setDisconnectHandler(AMPS::DisconnectHandler(&HADisconnectHandler::invoke, NULL));
#endif
        setSubscriptionManager(new MemorySubscriptionManager());
    }

    ~HAClientImpl()
    {
        _disconnected = true;
        _cleanup();
    }

    void setTimeout(int timeout_)
    {
        _timeout = timeout_;
    }

    int getTimeout() const
    {
        return _timeout;
    }

    unsigned int getReconnectDelay(void) const
    {
      return _reconnectDelay;
    }

    void setReconnectDelay(unsigned int reconnectDelay_)
    {
      _reconnectDelay = reconnectDelay_;
      setReconnectDelayStrategy(new FixedDelayStrategy(
            (unsigned int)reconnectDelay_));
    }

    void setReconnectDelayStrategy(const ReconnectDelayStrategy& strategy_)
    {
        _reconnectDelayStrategy = strategy_;
    }

    ReconnectDelayStrategy getReconnectDelayStrategy(void) const
    {
        return _reconnectDelayStrategy;
    }

    std::string getLogonOptions(void) const
    {
        return _logonOptions;
    }

    void setLogonOptions(const std::string& logonOptions_)
    {
        _logonOptions = logonOptions_;
    }

    void setLogonOptions(const char* logonOptions_)
    {
        _logonOptions = logonOptions_;
    }

    ServerChooser getServerChooser() const
    {
        return _serverChooser;
    }

    void setServerChooser(const ServerChooser& serverChooser_)
    {
        _serverChooser = serverChooser_;
    }

    class HADisconnectHandler
    {
    public:
        HADisconnectHandler() {}
        static void invoke(Client& client, void* );
#ifdef AMPS_USE_FUNCTIONAL
        void operator()(Client& client)
        {
            invoke(client, NULL);
        }
#endif
    };
    void connectAndLogon()
    {
        Lock<Mutex> l(_connectAndLogonLock);
        _disconnected = false;
        while(true)
        {
            connectAndLogonInternal();
            try
            {
                // Resubscribe
                if(_subscriptionManager)
                {
                    Client c(this);
                    _subscriptionManager->resubscribe(c);
                    broadcastConnectionStateChanged(
                                        ConnectionStateListener::Resubscribed);
                }
                return;
            }
            catch(const AMPSException& subEx)
            {
                _serverChooser.reportFailure(subEx, getConnectionInfo());
            }
        }
    }

    virtual void connect(const std::string& /*uri*/)
    {
        connectAndLogon();
    }

    virtual std::string logon(long /*timeout_*/, Authenticator& /*authenticator_*/,
                              const char* /*options_*/)
    {
        if (_disconnected) throw DisconnectedException("Attempt to call logon on a disconnected HAClient. Use connectAndLogon() instead.");
        throw AlreadyConnectedException("Attempt to call logon on an HAClient. Use connectAndLogon() instead.");
    }

    class HADoNothingDisconnectHandler
    {
    public:
        HADoNothingDisconnectHandler() {}
        static void invoke(Client& , void* ) {;}
#ifdef AMPS_USE_FUNCTIONAL
        void operator()(Client& client)
        {
            invoke(client, NULL);
        }
#endif
    };

    class DisconnectHandlerDisabler
    {
    public:
        DisconnectHandlerDisabler() : _client(NULL) { }
        DisconnectHandlerDisabler(HAClientImpl* client_)
        : _client(client_)
        {
            _handler = _client->getDisconnectHandler();
            setHandler();
        }
        ~DisconnectHandlerDisabler() { clear(); }
        void clear()
        {
            if(_client)
            {
                _client->setDisconnectHandler(_handler);
                _client = NULL;
            }
        }
        void setClient(HAClientImpl* client_)
        {
            if (!_client)
            {
                _client = client_;
                _handler = _client->getDisconnectHandler();
                setHandler();
            }
        }
        void setHandler()
        {
#ifdef AMPS_USE_FUNCTIONAL
            _client->setDisconnectHandler(HADoNothingDisconnectHandler());
#else
            _client->setDisconnectHandler(DisconnectHandler(
                                          &HADoNothingDisconnectHandler::invoke,
                                          NULL));
#endif
        }
    private:
        HAClientImpl* _client;
        DisconnectHandler _handler;
    };

    void connectAndLogonInternal()
    {
        if(!_serverChooser.isValid())
        {
            throw ConnectionException("No server chooser registered with HAClient");
        }
        {
            DisconnectHandlerDisabler disconnectDisabler;
            while(!_disconnected)
            {
                std::string uri = _serverChooser.getCurrentURI();
                if(uri.empty())
                {
                    throw ConnectionException("No AMPS instances available for connection. " + _serverChooser.getError());
                }
                Authenticator& auth = _serverChooser.getCurrentAuthenticator();
                _sleepBeforeConnecting(uri);
                try
                {
                    // Begin locked section
                    Lock<Mutex> l(_connectLock);
                    // Check if another thread disconnected or already connected
                    if(_disconnected || _connected)
                    {
                        return;
                    }
                    // Temporarily unset the disconnect handler since we will loop
                    disconnectDisabler.setClient((HAClientImpl*)this);
                    // Connect
                    ClientImpl::connect(uri);
                    if (_logonOptions.empty())
                    {
                        ClientImpl::logon(_timeout, auth);
                    }
                    else
                    {
                        ClientImpl::logon(_timeout, auth, _logonOptions.c_str());
                    }
                    disconnectDisabler.clear();
                    try
                    {
                        _serverChooser.reportSuccess(getConnectionInfo());
                        _reconnectDelayStrategy.reset();
                    }
                    catch(const AMPSException& e)
                    {
                        ClientImpl::disconnect();
                        throw AMPSException(e);
                    }
                    break;
                }
                catch(const AMPSException& ex)
                {
                    ConnectionInfo ci = getConnectionInfo();
                    // Substitute the URI on the connection info with the one we attempted
                    ci["client.uri"] = uri;
                    _serverChooser.reportFailure(ex,ci);
                    try
                    {
                        ClientImpl::setDisconnected();
                    }
                    catch (const AMPSException& e)
                    {
                        try
                        {
                            _exceptionListener->exceptionThrown(e);
                        } catch (...) { }
                    }
                }
            }
        }
        return;
    }

    ConnectionInfo gatherConnectionInfo() const
    {
        return getConnectionInfo();
    }

    ConnectionInfo getConnectionInfo() const
    {
        ConnectionInfo info = ClientImpl::getConnectionInfo();
        std::ostringstream writer;

        writer << getReconnectDelay();
        info["haClient.reconnectDelay"] = writer.str();
        writer.clear(); writer.str("");
        writer << _timeout;
        info["haClient.timeout"] = writer.str();

        return info;
    }

    bool disconnected() const
    {
        return _disconnected;
    }
    private:

    void disconnect()
    {
        {
            Lock<Mutex> l(_connectLock);
            _disconnected = true;
        }
        ClientImpl::disconnect();
    }
    void _millisleep(unsigned int millis_)
    {
#ifdef _WIN32
      Sleep(millis_);
#else
      struct timespec ts;
      ts.tv_sec  = millis_ / 1000;
      ts.tv_nsec = millis_ % 1000 * 1000000;
      nanosleep(&ts, NULL);
#endif
    }
    void _sleepBeforeConnecting(const std::string& uri_)
    {
      try
      {
        _millisleep(
            _reconnectDelayStrategy.getConnectWaitDuration(uri_));
      }
      catch(const ConnectionException&)
      {
        throw;
      }
      catch(const std::exception& ex_)
      {
        _exceptionListener->exceptionThrown(ex_);
        throw ConnectionException(ex_.what());
      }
      catch(...)
      {
        throw ConnectionException("Unknown exception thrown by "
            "the HAClient's delay strategy.");
      }
    }

    Mutex                  _connectLock;
    Mutex                  _connectAndLogonLock;
    int                    _timeout;
    unsigned int           _reconnectDelay;
    ReconnectDelayStrategy _reconnectDelayStrategy;
    ServerChooser          _serverChooser;
    volatile bool          _disconnected;
    std::string            _logonOptions;

}; // class HAClientImpl

}// namespace AMPS

#endif //_HACLIENTIMPL_H_