GPUdb.hpp

Go to the documentation of this file.

#ifndef __GPUDB_HPP__
#define __GPUDB_HPP__

#include "gpudb/protocol/EndpointKeywords.h"
#include "gpudb/utils/Utils.h"

#ifndef GPUDB_NO_HTTPS
#include <boost/asio/ssl.hpp>
#endif

#include <stdint.h>

#include <boost/thread/mutex.hpp>

namespace gpudb
{
    class GPUdb;
}

#include "gpudb/Avro.hpp"
#include "gpudb/GenericRecord.hpp"
#include "gpudb/GPUdbException.hpp"
#include "gpudb/Http.hpp"
#include "gpudb/Type.hpp"
#include "gpudb/protocol/GPUdbProtocol.h"


namespace gpudb
{
class GPUdb : private boost::noncopyable
{
public:

    class Options
    {
    public:
        Options();

#ifndef GPUDB_NO_HTTPS
        boost::asio::ssl::context* getSslContext() const;
#endif

        std::string getUsername() const;
        std::string getPassword() const;
        std::string getPrimaryUrl() const;
        bool getUseSnappy() const;
        size_t getThreadCount() const;
        avro::ExecutorPtr getExecutor() const;

        std::map<std::string, std::string>& getHttpHeaders();
        const std::map<std::string, std::string>& getHttpHeaders() const;

        size_t getTimeout() const;
        uint16_t getHostManagerPort() const;

#ifndef GPUDB_NO_HTTPS
        Options& setSslContext(boost::asio::ssl::context* value);
#endif

        Options& setUsername(const std::string& value);
        Options& setPassword(const std::string& value);
        Options& setPrimaryUrl(const std::string& value);
        Options& setUseSnappy(const bool value);
        Options& setThreadCount(const size_t value);
        Options& setExecutor(const avro::ExecutorPtr value);
        Options& setHttpHeaders(const std::map<std::string, std::string>& value);
        Options& addHttpHeader(const std::string& header, const std::string& value);
        Options& setTimeout(const size_t value);
        Options& setHostManagerPort(const uint16_t value);

    private:

#ifndef GPUDB_NO_HTTPS
        boost::asio::ssl::context* m_sslContext;
#endif

        std::string m_username;
        std::string m_password;
        std::string m_primaryUrl;
        bool m_useSnappy;
        size_t m_threadCount;
        avro::ExecutorPtr m_executor;
        std::map<std::string, std::string> m_httpHeaders;
        size_t   m_timeout;
        uint16_t m_hmPort;
    };


    enum HASynchronicityMode {
        // No override; defer to the HA process for synchronizing
        // endpoints (which has different logic for different endpoints)
        DEFAULT = 0,  // maps to 'none'
        // Synchronize all endpoint calls
        SYNCHRONOUS,  // maps to 'sync'
        // Do NOT synchronize any endpoint call
        ASYNCHRONOUS  // maps to 'async'
    };


    
    static const int64_t END_OF_SET = -9999;

    static const std::string DB_CONNECTION_RESET_ERROR_MESSAGE;
    static const std::string DB_CONNECTION_REFUSED_ERROR_MESSAGE;
    static const std::string DB_EXITING_ERROR_MESSAGE;
    static const std::string DB_OFFLINE_ERROR_MESSAGE;
    static const std::string DB_SYSTEM_LIMITED_ERROR_MESSAGE;
    static const std::string DB_HM_OFFLINE_ERROR_MESSAGE;
    
    static const std::string HEADER_AUTHORIZATION;
    static const std::string HEADER_CONTENT_TYPE;
    static const std::string HEADER_CONTENT_LENGTH;
    static const std::string HEADER_HA_SYNC_MODE;
    
    
    static inline std::string getApiVersion() { return GPUdb::API_VERSION; }

    GPUdb(const HttpUrl& url, const Options& options = Options());

    GPUdb(const std::string& url, const Options& options = Options());

    GPUdb(const std::vector<HttpUrl>& urls, const Options& options = Options());

    GPUdb(const std::vector<std::string>& urls, const Options& options = Options());

    ~GPUdb();
    
    const HttpUrl& getUrl() const;
    const std::vector<HttpUrl>& getUrls() const;
    const HttpUrl& getHmUrl() const;
    const std::vector<HttpUrl>& getHmUrls() const;

#ifndef GPUDB_NO_HTTPS
    boost::asio::ssl::context* getSslContext() const;
#endif

    const std::string& getUsername() const;
    const std::string& getPassword() const;

    const std::string& getPrimaryURL() const;

    bool getUseSnappy() const;
    size_t getThreadCount() const;
    avro::ExecutorPtr getExecutor() const;
    const std::map<std::string, std::string>& getHttpHeaders() const;
    HASynchronicityMode getHASyncMode() const;
    size_t getTimeout() const;


    void setHASyncMode( HASynchronicityMode mode );

    void addHttpHeader( const std::string& header,
                        const std::string& value );

    void removeHttpHeader( const std::string& header );

    
    void updateHostManagerPort();
    
    template<typename TRequest, typename TResponse>
    TResponse& submitRequest(const HttpUrl& url,
                             const TRequest& request,
                             TResponse& response,
                             const bool enableCompression = false) const
    {
        std::vector<uint8_t> requestBytes;
        try // encoding the request
        {
            avro::encode(requestBytes, request);
        }
        catch ( const std::exception& ex )
        {
            std::string message = GPUDB_STREAM_TO_STRING( "Error encountered while encoding request: "
                                                          << ex.what() );
            throw GPUdbException( message );
        }

        // Submit the request
        RawGpudbResponse gpudbResponse;
        submitRequestRaw(url, requestBytes, gpudbResponse, enableCompression);

        try // decoding the request
        {
            avro::decode(response, gpudbResponse.data);
        }
        catch ( const std::exception& ex )
        {
            std::string message = GPUDB_STREAM_TO_STRING( "Error encountered while decoding response: '"
                                                          << ex.what()
                                                          << "'; please ensure that the client API matches the server API version." );
            throw GPUdbException( message );
        }
        return response;
    }

    template<typename TRequest, typename TResponse>
    TResponse& submitRequest(const std::string& endpoint,
                             const TRequest& request,
                             TResponse& response,
                             const bool enableCompression = false) const
    {
        std::vector<uint8_t> requestBytes;
        try // encoding the request
        {
            avro::encode(requestBytes, request);
        }
        catch ( const std::exception& ex )
        {
            std::string message = GPUDB_STREAM_TO_STRING( "Error encountered while encoding request: "
                                                          << ex.what() );
            throw GPUdbException( message );
        }

        // Submit the request
        RawGpudbResponse gpudbResponse;
        submitRequestRaw(endpoint, requestBytes, gpudbResponse, enableCompression);

        try // decoding the request
        {
            avro::decode(response, gpudbResponse.data);
        }
        catch ( const std::exception& ex )
        {
            std::string message = GPUDB_STREAM_TO_STRING( "Error encountered while decoding response: '"
                                                          << ex.what()
                                                          << "'; please ensure that the client API matches the server API version." );
            throw GPUdbException( message );
        }

        return response;
    }

    template<typename TRequest, typename TResponse>
    TResponse& submitRequest(const char* endpoint,
                             const TRequest& request,
                             TResponse& response,
                             const bool enableCompression = false) const
    {
        submitRequest( (std::string) endpoint, request, response, enableCompression );
        return response;
    }


    template<typename TRequest, typename TResponse>
    TResponse& submitRequestToHostManager(const std::string& endpoint,
                                          const TRequest& request,
                                          TResponse& response,
                                          const bool enableCompression = false) const
    {
        // Handle host manager stuff here
        std::vector<uint8_t> requestBytes;
        try // encoding the request
        {
        avro::encode(requestBytes, request);
        }
        catch ( const std::exception& ex )
        {
            std::string message = GPUDB_STREAM_TO_STRING( "Error encountered while encoding request: "
                                                          << ex.what() );
            throw GPUdbException( message );
        }

        // Submit the request
        RawGpudbResponse gpudbResponse;
        submitRequestToHostManagerRaw(endpoint, requestBytes, gpudbResponse, enableCompression);

        try // decoding the request
        {
            avro::decode(response, gpudbResponse.data);
        }
        catch ( const std::exception& ex )
        {
            std::string message = GPUDB_STREAM_TO_STRING( "Error encountered while decoding response: '"
                                                          << ex.what()
                                                          << "'; please ensure that the client API matches the server API version." );
            throw GPUdbException( message );
        }
        return response;
    }   // end submitRequestToHostManager


    template<typename TRequest, typename TResponse>
    TResponse& submitRequestToHostManager(const char* endpoint,
                                          const TRequest& request,
                                          TResponse& response,
                                          const bool enableCompression = false) const
    {
        submitRequestToHostManager( (std::string) endpoint, request, response, enableCompression );
        return response;
    }   // end submitRequestToHostManager


#include "gpudb/GPUdbFunctions.hpp"

    void addKnownType(const std::string& typeId, const avro::DecoderPtr& decoder);

    template<typename T>
    void addKnownType(const std::string& typeId)
    {
        addKnownType(typeId, avro::createDecoder<T>());
    }

    template<typename T>
    void addKnownType(const std::string& typeId, const std::string& schemaString)
    {
        addKnownType(typeId, avro::createDecoder<T>(schemaString));
    }

    template<typename T>
    void addKnownType(const std::string& typeId, const ::avro::ValidSchema& schema)
    {
        addKnownType(typeId, avro::createDecoder<T>(schema));
    }

    void addKnownTypeFromTable(const std::string& tableName, const avro::DecoderPtr& decoder);

    template<typename T>
    void addKnownTypeFromTable(const std::string& tableName)
    {
        addKnownTypeFromTable(tableName, avro::createDecoder<T>());
    }

    template<typename T>
    void addKnownTypeFromTable(const std::string& tableName, const std::string& schemaString)
    {
        addKnownTypeFromTable(tableName, avro::createDecoder<T>(schemaString));
    }

    template<typename T>
    void addKnownTypeFromTable(const std::string& tableName, const ::avro::ValidSchema& schema)
    {
        addKnownTypeFromTable(tableName, avro::createDecoder<T>(schema));
    }

private:

    static const std::string API_VERSION;

    static const std::string FAILOVER_TRIGGER_MESSAGES[];
    static const std::string PROTECTED_HEADERS[];
    static const std::string HA_SYNCHRONICITY_MODE_VALUES[];

    static const size_t NUM_TRIGGER_MESSAGES;
    
    mutable std::vector<HttpUrl> m_urls;
    mutable std::vector<HttpUrl> m_hmUrls;
    std::string m_primaryUrlStr;
    HttpUrl* m_primaryUrlPtr;
    mutable std::vector<size_t>  m_urlIndices;
    mutable boost::mutex m_urlMutex;
    mutable size_t m_currentUrl;

#ifndef GPUDB_NO_HTTPS
    boost::asio::ssl::context* m_sslContext;
#endif

    std::string m_username;
    std::string m_password;
    std::string m_authorization;
    bool m_useSnappy;
    size_t m_threadCount;
    avro::ExecutorPtr m_executor;
    std::map<std::string, std::string> m_httpHeaders;
    size_t m_timeout;
    mutable HASynchronicityMode m_haSyncMode;
    Options m_options;

    mutable std::map<std::string, avro::DecoderPtr> m_knownTypes;
    mutable boost::mutex m_knownTypesMutex;

   
    void init();

    // Handle the primary host URL, if any is given via options
    void handlePrimaryURL();

    // Update the URLs with the available HA ring information
    void getHAringHeadNodeAdresses();

    void updateHostManagerUrls();
    void setHostManagerPort(uint16_t value);

    void randomizeURLs() const;

    const std::string& getHASynchronicityModeValue( HASynchronicityMode syncMode ) const;

    const HttpUrl* getUrlPointer() const;
    const HttpUrl* getHmUrlPointer() const;
    const HttpUrl* switchUrl(const HttpUrl* oldUrl) const;
    const HttpUrl* switchHmUrl(const HttpUrl* oldUrl) const;

    
    void initHttpRequest(HttpRequest& httpRequest) const;
    void submitRequestRaw(const std::string& endpoint,
                          const std::vector<uint8_t>& request,
                          RawGpudbResponse& response,
                          const bool enableCompression) const;
    void submitRequestToHostManagerRaw(const std::string& endpoint,
                                       const std::vector<uint8_t>& request,
                                       RawGpudbResponse& response,
                                       const bool enableCompression) const;
    void submitRequestRaw(const HttpUrl& url,
                          const std::vector<uint8_t>& request,
                          RawGpudbResponse& response,
                          const bool enableCompression,
                          const bool throwOnError = true) const;

    avro::DecoderPtr getDecoder(const std::string& typeId) const;
    void setDecoderIfMissing(const std::string& typeId,
                             const std::string& label,
                             const std::string& schemaString,
                             const std::map<std::string, std::vector<std::string> >& properties) const;

};

#include "gpudb/GPUdbTemplates.hpp"
    
}  // end namespace gpudb

#endif