jsanda

/*
 * Copyright 2014-2016 Red Hat, Inc. and/or its affiliates
 * and other contributors as indicated by the @author tags.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *

jsanda

   private static Logger logger = Logger.getLogger(RxTest.class);

    private rx.Scheduler tickScheduler;

    private ExecutorService queueExecutor;

    private rx.Scheduler queueScheduler;

    @BeforeClass
    public void initClass() {

jsanda

// Resets keyspace by truncating all tables in keyspace. 
//
// session is an Rx wrapper around com.datastax.driver.core.Session and
// and uses the async methods in Session. The execute method return an
// Observable<ResultSet>.
public void truncateTables(String keyspace) {
    session.execute("select table_name from system_schema.tables where keyspace_name = '" + keyspace + "'")
            .flatMap(Observable::from)
            .flatMap(row -> session.execute("truncate " + row.getString(0)))
            .toCompletable()

jsanda

cqlsh:hawkulartest> select distinct tenant_id, metric from data where tenant_id = 'T2' and type = 0 and metric = 'G488' and dpart = 0;

 tenant_id | metric
-----------+--------

(0 rows)

Tracing session: 3fe7c3d0-6a6a-11e6-ab52-51eaf4bb8e56

 activity                                                                                                                                         | timestamp                  | source    | source_elapsed | client

jsanda

cqlsh:hawkulartest> select distinct tenant_id, type, metric, dpart from data limit 1;

 tenant_id   | type | metric | dpart
-------------+------+--------+-------
 test-tenant |    0 |     m1 |     0

(1 rows)

Tracing session: 11b597d0-6a6a-11e6-ab52-51eaf4bb8e56

jsanda

    @BeforeClass
    public void initClass() {
        dataAccess = new DataAccessImpl(session);

        ConfigurationService configurationService = new ConfigurationService() ;
        configurationService.init(rxSession);

        metricsService = new MetricsServiceImpl();
        metricsService.setDataAccess(dataAccess);
        metricsService.setConfigurationService(configurationService);

jsanda

    public boolean enableAll(MetricRegistry registry)
    {
        boolean enabled = false;
        if (console != null)
        {
            if (enableConsole(registry))
            {
                enabled = true;
            }
        }

jsanda

private void initSchema() {
        doWithLock("cassalog", () -> {
                SchemaService schemaService = new SchemaService();
                schemaService.run(session, keyspace, Boolean.parseBoolean(resetDb));
                session.execute("USE " + keyspace);
        });
}
    private void doWithLock(String key, Runnable runnable) {
        AdvancedCache<String, String> cache = locksCache.getAdvancedCache();
        TransactionManager transactionManager = cache.getTransactionManager();

jsanda

@Listener
public class DistributedLock {

    private static Logger log = Logger.getLogger(DistributedLock.class);

    private static final String KEY = "cassalog";

    private AdvancedCache<String, String> locksCache;

    public DistributedLock(AdvancedCache<String, String> locksCache) {

jsanda

    private void initSchema() {
        doWithLock("cassalog", () -> {
                SchemaService schemaService = new SchemaService();
                schemaService.run(session, keyspace, Boolean.parseBoolean(resetDb));
                session.execute("USE " + keyspace);
        });
    }

    private void doWithLock(String key, Runnable runnable) {
        AdvancedCache<String, String> cache = locksCache.getAdvancedCache();