import ballerinax/cassandra as c;
import ballerina/io;
import ballerina/jsonutils;
import ballerina/xmlutils;

type Person record {
    int id;
    string name;
    float salary;
};

public function main() {
    c:Client conn = new ({
        host: "localhost",
        port: 9042,
        username: "cassandra",
        password: "cassandra",
        options: {
            queryOptionsConfig: {consistencyLevel: "ONE", defaultIdempotence: false},
            protocolOptionsConfig: {sslEnabled: false},
            socketOptionsConfig: {connectTimeoutMillis: 500, readTimeoutMillis: 1000},
            poolingOptionsConfig: {maxConnectionsPerHostLocal: 5, newConnectionThresholdLocal: 10}
        }
    });

    var returned = conn->update("CREATE KEYSPACE testballerina  WITH replication = {'class':'SimpleStrategy'," +
        "'replication_factor' : 3}");
    handleUpdate(returned, "Keyspace testballerina creation");

    returned = conn->update("CREATE TABLE testballerina.person(id int PRIMARY KEY,name text,salary float,income double," +
        "married boolean)");
    handleUpdate(returned, "Table person creation");

    c:Parameter pID = {cqlType: c:TYPE_INT, value: 1};
    c:Parameter pName = {cqlType: c:TYPE_TEXT, value: "Anupama"};
    c:Parameter pSalary = {cqlType: c:TYPE_FLOAT, value: 100.5};
    c:Parameter pIncome = {cqlType: c:TYPE_DOUBLE, value: 1000.5};
    c:Parameter pMarried = {cqlType: c:TYPE_BOOLEAN, value: true};
    returned = conn->update("INSERT INTO testballerina.person(id, name, salary, income, married) values (?,?,?,?,?)",
        pID, pName, pSalary, pIncome, pMarried);
    handleUpdate(returned, "Insert One Row to Table person");

    var selectRet = conn->query("select id, name, salary from testballerina.person where id = ?", Person, pID);

    if (selectRet is table<Person>) {
        foreach var row in selectRet {
            io:println("Person:" + row.id.toString() + "|" + row.name + "|" + row.salary.toString());
        }
    } 

    selectRet = conn->query("select id, name, salary from testballerina.person where id = ? and name = ?" +
        "ALLOW FILTERING", Person, pID, pName);

    if (selectRet is table<record {}>) {
        json jsonRet = jsonutils:fromTable(selectRet);
        io:println("JSON: ", jsonRet);
    } else {
        io:println("Select data from person table failed: " + selectRet.message());
    }

    selectRet = conn->query("select id, name, salary from testballerina.person where salary = ? ALLOW FILTERING",
        Person, pSalary);
    if (selectRet is table<record {}>) {
        xml xmlRet = xmlutils:fromTable(selectRet);
        io:println("XML: ", xmlRet);
    } else {
        io:println("Select data from person table failed: " + selectRet.message());
    }

    returned = conn->update("DROP KEYSPACE testballerina");
    handleUpdate(returned, "Drop keyspace testballerina");

    conn.stop();
    io:println("Connection closed.");
}

function handleUpdate(()|error returned, string message) {
    if (returned is ()) {
        io:println(message + " success ");
    } else {
        io:println(message + " failed: " + returned.message());
    }
}