connection.h

#pragma once
#include <QObject>
#include <QList>
#include <QMap>
#include <QByteArray>
#include <functional>
#include <QVariantList>
#include <QSharedPointer>
#include <QEventLoop>
#include <QTimer>
#include <QMutex>
#include "connectionconfig.h"
#include "exception.h"
#include "command.h"
#include "response.h"
#include "scancommand.h"

namespace RedisClient {

class AbstractTransporter;
class CommandExecutor;

typedef QMap<int, int> DatabaseList;

struct ServerInfo
{
    ServerInfo();

    double version;
    bool clusterMode;
    bool sentinelMode;
    DatabaseList databases;    

    class ParsedServerInfo : public QHash<QString, QHash<QString, QString>>
    {
    public:
        QVariantMap toVariantMap();
    };

    ParsedServerInfo parsed;

    static ServerInfo fromString(const QString& info);
};

class Connection : public QObject
{
    Q_OBJECT
    ADD_EXCEPTION

    friend class AbstractTransporter;

public:
    enum class Mode { Normal, PubSub, Cluster, Sentinel };
    class InvalidModeException : public Connection::Exception {};
public:
    Connection(const ConnectionConfig &c, bool autoConnect=true);

    virtual ~Connection();

    virtual bool connect(bool wait=true);

    virtual bool isConnected();

    virtual void disconnect();

    ConnectionConfig getConfig() const;

    void setConnectionConfig(const ConnectionConfig &);

    Mode mode() const;

    virtual double getServerVersion();

    virtual DatabaseList getKeyspaceInfo();

    /*
     * Hi-Level Operations API
     */
    typedef QList<QByteArray> RawKeysList;
    typedef std::function<void(const RawKeysList&, const QString&)> RawKeysListCallback;

    virtual void getDatabaseKeys(RawKeysListCallback callback,
                                 const QString& pattern=QString("*"),
                                 uint dbIndex = 0);

    virtual void getClusterKeys(RawKeysListCallback callback,
                                const QString &pattern);


    typedef QPair<QString, int> Host;
    typedef QList<Host> HostList;

    HostList getMasterNodes();

    /*
     * Command execution API
     */
    void command(const Command& cmd);

    void command(QList<QByteArray> rawCmd, int db = -1);

    void command(QList<QByteArray> rawCmd, QObject *owner,
                 RedisClient::Command::Callback callback, int db = -1);

    Response commandSync(const Command& cmd);

    Response commandSync(QList<QByteArray> rawCmd, int db = -1);    

    /*
     * Aliases for ^ function
     */
    Response commandSync(QString cmd, int db = -1);
    Response commandSync(QString cmd, QString arg1, int db = -1);
    Response commandSync(QString cmd, QString arg1, QString arg2, int db = -1);
    Response commandSync(QString cmd, QString arg1, QString arg2, QString arg3, int db = -1);

    typedef std::function<void(QVariant, QString err)>  CollectionCallback;

    typedef std::function<void(QVariant, QString err, bool final)>  IncrementalCollectionCallback;

    virtual void retrieveCollection(QSharedPointer<ScanCommand> cmd,
                                    CollectionCallback callback);

    virtual void retrieveCollectionIncrementally(QSharedPointer<ScanCommand> cmd,
                                                 IncrementalCollectionCallback callback);

    virtual void runCommand(const Command &cmd);


    bool waitForIdle(uint timeout);

    /*
     * Low level functions for modification
     * commands execution.
     */
    void setTransporter(QSharedPointer<AbstractTransporter>);
    QSharedPointer<AbstractTransporter> getTransporter() const;

signals:    
    void addCommandToWorker(const Command&);
    void error(const QString&);
    void log(const QString&);
    void connected();
    void disconnected();
    void authOk();
    void authError(const QString&);

    // Cluster & Sentinel
    void reconnectTo(const QString& host, int port);

protected:
    void createTransporter();
    bool isTransporterRunning();

    Response internalCommandSync(QList<QByteArray> rawCmd);

    void processScanCommand(QSharedPointer<ScanCommand> cmd,
                            CollectionCallback callback,
                            QSharedPointer<QVariantList> result=QSharedPointer<QVariantList>(),
                            bool incrementalProcessing=false);

    void changeCurrentDbNumber(int db);    

protected slots:
    void auth();

protected:
    ConnectionConfig m_config;   
    QSharedPointer<QThread> m_transporterThread;
    QSharedPointer<AbstractTransporter> m_transporter;

    int m_dbNumber;
    ServerInfo m_serverInfo;
    Mode m_currentMode;
    QMutex m_dbNumberMutex;
    bool m_autoConnect;    
    bool m_stoppingTransporter;
    RawKeysListCallback m_wrapper;
};
}