chbatey

CREATE TABLE if NOT EXISTS customer_events ( 
    customer_id text, 
    statff_id text, 
    store_type text, 
    time timeuuid , 
    event_type text, 
    PRIMARY KEY (customer_id, time))

CREATE TABLE if NOT EXISTS customer_events_by_staff ( 
    customer_id text, 

chbatey

private final static String insertByCustomerIdCql = "INSERT INTO events.customer_events (customer_id, time , event_type , statff_id , store_type ) VALUES ( ?, ?, ?, ?, ?)";
private final static String insertByStaffIdCql = "INSERT INTO events.customer_events_by_staff_id (customer_id, time , event_type , statff_id , store_type ) VALUES ( ?, ?, ?, ?, ?)";

public CustomerEventDao() {
    cluster = Cluster.builder().addContactPoint("localhost").build();
    session = cluster.connect("events");
    insertByCustomerId = session.prepare(insertByCustomerIdCql);
    insertByStaffId = session.prepare(insertByStaffIdCql);
}

chbatey

public void storeEvents(String customerId, ConsistencyLevel consistencyLevel, CustomerEvent... events) {
  ...
}

chbatey

public void storeEvents(ConsistencyLevel consistencyLevel, CustomerEvent... events) {
    BatchStatement batchStatement = new BatchStatement(BatchStatement.Type.UNLOGGED);
    batchStatement.enableTracing();

    for (CustomerEvent event : events) {
        batchStatement.add(createBoundStatement(consistencyLevel, event));
    }

    ResultSet execute = session.execute(batchStatement);
    logTraceInfo(execute.getExecutionInfo());

chbatey

public void storeEvents(ConsistencyLevel consistencyLevel, CustomerEvent... events) {
        for (CustomerEvent event : events) {
            storeEvent(consistencyLevel, event);
        }
}

chbatey

public void storeEvents(ConsistencyLevel consistencyLevel, CustomerEvent... events) {
    ...
 }

chbatey

package info.batey.examples.cassandra;

import com.datastax.driver.core.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;

public class CustomerEventDao implements Closeable {

chbatey

CREATE TABLE if NOT EXISTS customer_events ( 
    customer_id text, 
    statff_id text, 
    store_type text, 
    time timeuuid , 
    event_type text, 
    PRIMARY KEY (customer_id, time))

chbatey

String query = "select name from people";
Map<String, String> row = ImmutableMap.of("name", "Christopher");
Map<String, String> row = ImmutableMap.of("name", "Alexandra");
Map<String, ColumnTypes> columnTypes = ImmutableMap.of("name", ColumnTypes.Varchar);
PrimingRequest prime = PrimingRequest.queryBuilder()
    .withQuery(query)
    .withRows(row)
    .withColumnTypes(columnTypes)
    .build();
primingClient.primeQuery(prime);

chbatey

public class KafkaEventStoreTest {

    @Rule
    public KafkaUnitRule kafkaUnit = new KafkaUnitRule(5000, 5001);

    private static String topicName = "Events";

    private KafkaEventStore underTest;

    @Before