QCassandraPrivate.cpp

Go to the documentation of this file.

/*
 * Text:
 *      QCassandraPrivate.cpp
 *
 * Description:
 *      Handling of the cassandra::CassandraClient and corresponding transports,
 *      protocols, sockets, etc.
 *
 * Documentation:
 *      See each function below.
 *
 * License:
 *      Copyright (c) 2011 Made to Order Software Corp.
 * 
 *      http://snapwebsites.org/
 *      contact@m2osw.com
 * 
 *      Permission is hereby granted, free of charge, to any person obtaining a
 *      copy of this software and associated documentation files (the
 *      "Software"), to deal in the Software without restriction, including
 *      without limitation the rights to use, copy, modify, merge, publish,
 *      distribute, sublicense, and/or sell copies of the Software, and to
 *      permit persons to whom the Software is furnished to do so, subject to
 *      the following conditions:
 *
 *      The above copyright notice and this permission notice shall be included
 *      in all copies or substantial portions of the Software.
 *
 *      THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
 *      OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 *      MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
 *      IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
 *      CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
 *      TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
 *      SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

#include "QCassandraPrivate.h"
#include <protocol/TBinaryProtocol.h>
#include <transport/TSocket.h>
#include <transport/TTransportUtils.h>

#include <QtCore/QDebug>

namespace QtCassandra
{


QCassandraPrivate::QCassandraPrivate(QCassandra *parent)
    : f_parent(parent)
      //f_socket(NULL) -- auto-initialized
      //f_transport(NULL) -- auto-initialized
      //f_protocol(NULL) -- auto-initialized
      //f_client(NULL) -- auto-initialized
{
}

QCassandraPrivate::~QCassandraPrivate()
{
    disconnect();
}

bool QCassandraPrivate::connect(const QString& host, int port)
{
    bool    worked = true;

    // disconnect any existing connection
    disconnect();

    try {
        // create a socket, transportation, protocol, and client
        // the client is what we use to communicate with the Cassandra server
        f_socket.reset(new apache::thrift::transport::TSocket(host.toUtf8().data(), port));
        f_transport.reset(new apache::thrift::transport::TFramedTransport(f_socket));
        f_protocol.reset(new apache::thrift::protocol::TBinaryProtocol(f_transport));
        f_client.reset(new org::apache::cassandra::CassandraClient(f_protocol));

        // once everything is connected as it should, open the transport link
        f_transport->open();
    }
    catch(...) {
        // TBD: should we check for some specific error here?
        worked = false;
    }

    // if it failed, make sure to clear all the pointers
    if(!worked) {
        disconnect();
    }

    return worked;
}

void QCassandraPrivate::disconnect()
{
    if(f_client.get()) {
        // this is probably not necessary (it should anyway happen as required)
        f_transport->close();
    }
    f_client.reset();
    f_protocol.reset();
    f_transport.reset();
    f_socket.reset();
}

bool QCassandraPrivate::isConnected() const
{
    return f_client.get() != NULL;
}

void QCassandraPrivate::mustBeConnected() const throw(std::runtime_error)
{
    if(!isConnected()) {
        throw std::runtime_error("not connected to the Cassandra server.");
    }
}

QString QCassandraPrivate::clusterName() const
{
    mustBeConnected();
    std::string cluster_name;
    f_client->describe_cluster_name(cluster_name);
    return cluster_name.c_str();
}

QString QCassandraPrivate::protocolVersion() const
{
    mustBeConnected();
    std::string protocol_version;
    f_client->describe_version(protocol_version);
    return protocol_version.c_str();
}

void QCassandraPrivate::setContext(const QString& context_name)
{
    mustBeConnected();
    f_client->set_keyspace(context_name.toUtf8().data());
}

void QCassandraPrivate::contexts() const
{
    mustBeConnected();

    // retrieve the key spaces from Cassandra
    std::vector<org::apache::cassandra::KsDef> keyspaces;
    f_client->describe_keyspaces(keyspaces);

    for(std::vector<org::apache::cassandra::KsDef>::const_iterator
                    ks = keyspaces.begin(); ks != keyspaces.end(); ++ks) {
        QSharedPointer<QCassandraContext> c(f_parent->context(ks->name.c_str()));
        const org::apache::cassandra::KsDef& ks_def = *ks;
        c->parseContextDefinition(&ks_def);
    }
}

void QCassandraPrivate::createContext(const QCassandraContext& context)
{
    mustBeConnected();
    org::apache::cassandra::KsDef ks;
    context.prepareContextDefinition(&ks);
    std::string id_ignore;
    f_client->system_add_keyspace(id_ignore, ks);
}

void QCassandraPrivate::updateContext(const QCassandraContext& context)
{
    mustBeConnected();
    org::apache::cassandra::KsDef ks;
    context.prepareContextDefinition(&ks);
    std::string id_ignore;
    f_client->system_update_keyspace(id_ignore, ks);
}

void QCassandraPrivate::dropContext(const QCassandraContext& context)
{
    mustBeConnected();
    f_parent->clearCurrentContextIf(context);
    std::string id_ignore;
    f_client->system_drop_keyspace(id_ignore, context.contextName().toUtf8().data());
}

void QCassandraPrivate::createTable(const QCassandraTable *table)
{
    mustBeConnected();
    org::apache::cassandra::CfDef cf;
    table->prepareTableDefinition(&cf);
    std::string id_ignore;
    f_client->system_add_column_family(id_ignore, cf);
}

void QCassandraPrivate::updateTable(const QCassandraTable *table)
{
    mustBeConnected();
    org::apache::cassandra::CfDef cf;
    table->prepareTableDefinition(&cf);
    std::string id_ignore;
    f_client->system_update_column_family(id_ignore, cf);
}

void QCassandraPrivate::dropTable(const QString& table_name)
{
    mustBeConnected();
    std::string id_ignore;
    f_client->system_drop_column_family(id_ignore, table_name.toUtf8().data());
}

void QCassandraPrivate::truncateTable(const QCassandraTable *table)
{
    mustBeConnected();
    f_client->truncate(table->tableName().toUtf8().data());
}

void QCassandraPrivate::insertValue(const QString& table_name, const QByteArray& row_key, const QByteArray& column_key, const QCassandraValue& value)
{
    mustBeConnected();

    std::string key(row_key.data(), row_key.size());

    org::apache::cassandra::ColumnParent column_parent;
    column_parent.__set_column_family(table_name.toUtf8().data());
    // no super column support here

    org::apache::cassandra::Column column;
    std::string ck(column_key.data(), column_key.size());
    column.__set_name(ck);

    const QByteArray& data(value.binaryValue());
    column.__set_value(std::string(data.data(), data.size())); // "unavoidable" copy of the data

    QCassandraValue::timestamp_mode_t mode = value.timestampMode();
    switch(mode) {
    case QCassandraValue::TIMESTAMP_MODE_AUTO:
        column.__set_timestamp(QCassandra::timeofday());
        break;

    case QCassandraValue::TIMESTAMP_MODE_DEFINED:
        // user defined
        column.__set_timestamp(value.timestamp());
        break;

    }

    if(value.ttl() != QCassandraValue::TTL_PERMANENT) {
        column.__set_ttl(value.ttl());
    }

    // our consistency level is 100% based on the Thrift consistency level
    // a cast is enough to get the value we want to get
    // (see the QCassandraValue.cpp file)

    consistency_level_t consistency_level = value.consistencyLevel();
    if(consistency_level == CONSISTENCY_LEVEL_DEFAULT) {
        consistency_level = f_parent->defaultConsistencyLevel();
    }

    f_client->insert(key, column_parent, column, static_cast<org::apache::cassandra::ConsistencyLevel::type>(static_cast<cassandra_consistency_level_t>(consistency_level)));
}

void QCassandraPrivate::getValue(const QString& table_name, const QByteArray& row_key, const QByteArray& column_key, QCassandraValue& value)
{
    mustBeConnected();

    std::string key(row_key.data(), row_key.size());

    org::apache::cassandra::ColumnPath column_path;
    column_path.__set_column_family(table_name.toUtf8().data());
    // no super column support here
    column_path.__set_column(column_key.data());

    org::apache::cassandra::ColumnOrSuperColumn column_result;

    // our consistency level is 100% based on the Thrift consistency level
    // a cast is enough to get the value we want to get
    // (see the QCassandraValue.cpp file)

    consistency_level_t consistency_level = value.consistencyLevel();
    if(consistency_level == CONSISTENCY_LEVEL_DEFAULT) {
        consistency_level = f_parent->defaultConsistencyLevel();
    }

    f_client->get(column_result, key, column_path, static_cast<org::apache::cassandra::ConsistencyLevel::type>(static_cast<cassandra_consistency_level_t>(consistency_level)));

    if(!column_result.__isset.column) {
        throw std::runtime_error("attempt to retrieve a cell failed");
    }

    // we got a column, copy the data to the value parameter
    if(column_result.column.__isset.value) {
        value.setBinaryValue(QByteArray(column_result.column.value.c_str(), column_result.column.value.length()));
    }
    else {
        // undefined we assume empty...
        value.setNullValue();
    }
    if(column_result.column.__isset.timestamp) {
        value.assignTimestamp(column_result.column.timestamp);
    }
    if(column_result.column.__isset.ttl) {
        value.setTtl(column_result.column.ttl);
    }
}

int32_t QCassandraPrivate::getCellCount(const QString& table_name, const QByteArray& row_key, const QCassandraColumnPredicate& column_predicate)
{
    mustBeConnected();

    std::string key(row_key.data(), row_key.size());

    org::apache::cassandra::ColumnParent column_parent;
    column_parent.__set_column_family(table_name.toUtf8().data());
    // no super column support here

    org::apache::cassandra::SlicePredicate slice_predicate;
    column_predicate.toPredicate(&slice_predicate);

    // our consistency level is 100% based on the Thrift consistency level
    // a cast is enough to get the value we want to get
    // (see the QCassandraValue.cpp file)

    consistency_level_t consistency_level = column_predicate.consistencyLevel();
    if(consistency_level == CONSISTENCY_LEVEL_DEFAULT) {
        consistency_level = f_parent->defaultConsistencyLevel();
    }

    return f_client->get_count(key, column_parent, slice_predicate, static_cast<org::apache::cassandra::ConsistencyLevel::type>(static_cast<cassandra_consistency_level_t>(consistency_level)));
}

void QCassandraPrivate::getColumnSlice(QCassandraTable& table, const QByteArray& row_key, const QCassandraColumnPredicate& column_predicate)
{
    mustBeConnected();

    std::string key(row_key.data(), row_key.size());

    org::apache::cassandra::ColumnParent column_parent;
    column_parent.__set_column_family(table.tableName().toUtf8().data());
    // no super column support here

    org::apache::cassandra::SlicePredicate slice_predicate;
    column_predicate.toPredicate(&slice_predicate);

    typedef std::vector<org::apache::cassandra::ColumnOrSuperColumn> column_vector_t;
    column_vector_t results;

    // our consistency level is 100% based on the Thrift consistency level
    // a cast is enough to get the value we want to get
    // (see the QCassandraValue.cpp file)

    consistency_level_t consistency_level = column_predicate.consistencyLevel();
    if(consistency_level == CONSISTENCY_LEVEL_DEFAULT) {
        consistency_level = f_parent->defaultConsistencyLevel();
    }

    f_client->get_slice(results, key, column_parent, slice_predicate, static_cast<org::apache::cassandra::ConsistencyLevel::type>(static_cast<cassandra_consistency_level_t>(consistency_level)));

    // we got results, copy the data to the table cache
    for(column_vector_t::iterator it = results.begin(); it != results.end(); ++it) {
        // transform the value of the cell to a QCassandraValue
        QCassandraValue value;
        if(it->column.__isset.value) {
            value.setBinaryValue(QByteArray(it->column.value.c_str(), it->column.value.length()));
        }
        if(it->column.__isset.timestamp) {
            value.assignTimestamp(it->column.timestamp);
        }
        if(it->column.__isset.ttl) {
            value.setTtl(it->column.ttl);
        }

        // save the cell in the corresponding table, row, cell
        table.assignRow(row_key, it->column.name.c_str(), value);
    }
}

void QCassandraPrivate::remove(const QString& table_name, const QByteArray& row_key, const QByteArray& column_key, int64_t timestamp, consistency_level_t consistency_level)
{
    mustBeConnected();

    std::string key(row_key.data(), row_key.size());

    org::apache::cassandra::ColumnPath column_path;
    column_path.__set_column_family(table_name.toUtf8().data());
    // no super column support here
    if(column_key.size() > 0) {
        // when column_key is empty we want to remove all the columns!
        std::string ckey(column_key.data(), column_key.size());
        column_path.__set_column(ckey);
    }

    if(consistency_level == CONSISTENCY_LEVEL_DEFAULT) {
        consistency_level = f_parent->defaultConsistencyLevel();
    }

    f_client->remove(key, column_path, timestamp, static_cast<org::apache::cassandra::ConsistencyLevel::type>(static_cast<cassandra_consistency_level_t>(consistency_level)));
}

uint32_t QCassandraPrivate::getRowSlices(QCassandraTable& table, const QCassandraRowPredicate& row_predicate)
{
    mustBeConnected();

    org::apache::cassandra::ColumnParent column_parent;
    column_parent.__set_column_family(table.tableName().toUtf8().data());
    // no super column support here

    QSharedPointer<QCassandraColumnPredicate> column_predicate(row_predicate.columnPredicate());
    org::apache::cassandra::SlicePredicate slice_predicate;
    column_predicate->toPredicate(&slice_predicate);

    org::apache::cassandra::KeyRange key_range;
    row_predicate.toPredicate(&key_range);

    typedef std::vector<org::apache::cassandra::KeySlice> key_slice_vector_t;
    key_slice_vector_t results;

    // our consistency level is 100% based on the Thrift consistency level
    // a cast is enough to get the value we want to get
    // (see the QCassandraValue.cpp file)

    consistency_level_t consistency_level = column_predicate->consistencyLevel();
    if(consistency_level == CONSISTENCY_LEVEL_DEFAULT) {
        consistency_level = f_parent->defaultConsistencyLevel();
    }

    f_client->get_range_slices(results, column_parent, slice_predicate, key_range, static_cast<org::apache::cassandra::ConsistencyLevel::type>(static_cast<cassandra_consistency_level_t>(consistency_level)));

    // we got results, copy the data to the table cache
    for(key_slice_vector_t::iterator it = results.begin(); it != results.end(); ++it) {
        QByteArray row_key(it->key.c_str(), it->key.size());
        typedef std::vector<org::apache::cassandra::ColumnOrSuperColumn> column_vector_t;
        for(column_vector_t::iterator jt = it->columns.begin(); jt != it->columns.end(); ++jt) {
            // transform the value of the cell to a QCassandraValue
            QCassandraValue value;
            if(jt->column.__isset.value) {
                value.setBinaryValue(QByteArray(jt->column.value.c_str(), jt->column.value.length()));
            }
            if(jt->column.__isset.timestamp) {
                value.assignTimestamp(jt->column.timestamp);
            }
            if(jt->column.__isset.ttl) {
                value.setTtl(jt->column.ttl);
            }

            // save the cell in the corresponding table, row, cell
            table.assignRow(row_key, jt->column.name.c_str(), value);
        }
    }

    return results.size();
}

} // namespace QtCassandra
// vim: ts=4 sw=4 et