StorageClient.java

package com.example.jetpack;

import com.netflix.astyanax.AstyanaxContext;
import com.netflix.astyanax.ColumnListMutation;
import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.NodeDiscoveryType;
import com.netflix.astyanax.connectionpool.OperationResult;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.connectionpool.impl.ConnectionPoolConfigurationImpl;
import com.netflix.astyanax.connectionpool.impl.CountingConnectionPoolMonitor;
import com.netflix.astyanax.impl.AstyanaxConfigurationImpl;
import com.netflix.astyanax.model.ColumnFamily;
import com.netflix.astyanax.serializers.DoubleSerializer;
import com.netflix.astyanax.serializers.IntegerSerializer;
import com.netflix.astyanax.serializers.StringSerializer;
import com.netflix.astyanax.thrift.ThriftFamilyFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

public class StorageClient {

    static final int TTL = 86400;

    final Keyspace keyspace;
    static Logger log = LoggerFactory.getLogger(StorageClient.class);

    ColumnFamily<String, Integer> CF_METRICS =
            new ColumnFamily<String, Integer>(
                    "Metrics",                // Column Family Name
                    StringSerializer.get(),   // Key Serializer
                    IntegerSerializer.get()); // Column Serializer

    public StorageClient() {
        AstyanaxContext<Keyspace> context = new AstyanaxContext.Builder()
                .forCluster("ClusterName")
                .forKeyspace("jetpack")
                .withAstyanaxConfiguration(new AstyanaxConfigurationImpl()
                        .setDiscoveryType(NodeDiscoveryType.RING_DESCRIBE)
                )
                .withConnectionPoolConfiguration(new ConnectionPoolConfigurationImpl("pool")
                        .setPort(9160)
                        .setMaxConnsPerHost(20)
                        .setSeeds("127.0.0.1:9160")
                )
                .withConnectionPoolMonitor(new CountingConnectionPoolMonitor())
                .buildKeyspace(ThriftFamilyFactory.getInstance());

        context.start();
        this.keyspace = context.getEntity();
    }

    public void store(String key, Map<Integer, Double> metrics) {
        MutationBatch m = keyspace.prepareMutationBatch();
        ColumnListMutation<Integer> row = m.withRow(CF_METRICS, key);

        for (Map.Entry<Integer, Double> metric : metrics.entrySet()) {
            row.putColumn(metric.getKey(), metric.getValue(), DoubleSerializer.get(), TTL);
        }

        try {
            OperationResult<Void> result = m.execute();
            log.info("Storage result: " + result.toString());
        } catch (ConnectionException e) {
            log.error("Error connecting to Cassandra.", e);
        }
    }

}