#include "DDBConnection.h"
#include "Swordfish.h"

using namespace ddb;

DDBConnection::DDBConnection(const string &host, const int &port, const string &username, const string &passwd): host_(host), port_(port), username_(username), passwd_(passwd) {
    session_ = DolphinDBLib::createSession();
    vector<ConstantSP> arguments = { new String(host), new Int(port), new String(username), new String(passwd) };
    conn_ = DolphinDBLib::execute(session_, "xdb", arguments);
}

DDBConnection::DDBConnection(SessionSP session, const string &host, const int &port, const string &username, const string &passwd): session_(session), host_(host), port_(port), username_(username), passwd_(passwd) {
    vector<ConstantSP> arguments = { new String(host), new Int(port), new String(username), new String(passwd) };
    conn_ = DolphinDBLib::execute(session_, "xdb", arguments);
}

DDBConnection::~DDBConnection() {
    if (!session_.isNull() and !session_->isCancelled() and !conn_->isNull()) {
        try {
            close();
        } catch (...) {
            // ignore
        }
        conn_ = nullptr;
    }
}

ConstantSP DDBConnection::run(const string &script) {
    vector<ConstantSP> arguments = { conn_, new String(script) };
    return DolphinDBLib::execute(session_, "remoteRun", arguments);
}

ConstantSP DDBConnection::run(const string &functionName, const vector<ConstantSP> &arguments) {
    VectorSP vec = Util::createVector(DT_ANY, arguments.size());
    for (std::size_t i = 0; i < arguments.size(); ++i) {
        vec->set(i, arguments[i]);
    }
    vector<ConstantSP> args = { conn_, new String(functionName), vec };
    return DolphinDBLib::execute(session_, "remoteRun", args);
}

ConstantSP DDBConnection::upload(const string &name, const ConstantSP &value) {
    vector<ConstantSP> args1 =  { value };
    ConstantSP jsonValue = DolphinDBLib::execute(session_, "toJson", args1);
    const string script = name + "=fromJson('" + jsonValue->getString() + "')";
    vector<ConstantSP> args2 = { conn_, new String(script) };
    return DolphinDBLib::execute(session_, "remoteRun", args2);
}

ConstantSP DDBConnection::subscribe(const string &tableName, const string &actionName, int offset, ConstantSP handler, int batchSize, int throttle) {
    vector<ConstantSP> arguments = { conn_, new String(tableName), new String(actionName), new Int(offset), handler, new Int(batchSize), new Int(throttle) };
    return DolphinDBLib::execute(session_, "subscribeTable", arguments);
}

ConstantSP DDBConnection::unsubscribe(const string &tableName, const string &actionName) {
    vector<ConstantSP> arguments = { conn_, new String(tableName), new String(actionName) };
    return DolphinDBLib::execute(session_, "unsubscribeTable", arguments);
}

void DDBConnection::close() {
    vector<ConstantSP> arguments = { conn_ };
    DolphinDBLib::execute(session_, "close", arguments);
    conn_ = nullptr;
}