package com.tradeshift.reaktive.replication;


import com.tradeshift.reaktive.cassandra.CassandraSession;
import com.typesafe.config.Config;

import akka.actor.ActorSystem;
import akka.persistence.cassandra.CassandraPluginConfig;
import io.vavr.collection.Vector;
import scala.collection.JavaConverters;

public class VisibilityCassandraSession extends CassandraSession {
    private final String keyspace;

    public VisibilityCassandraSession(ActorSystem system, String metricsCategory) {
        super(system, metricsCategory, initialStatements(system.settings().config().getConfig("ts-reaktive.replication.cassandra")));
        this.keyspace = system.settings().config().getString("ts-reaktive.replication.cassandra.keyspace");
    }

    public String getKeyspace() {
        return keyspace;
    }

    private static Vector<String> initialStatements(Config config) {
        String replStrategy = CassandraPluginConfig.getReplicationStrategy(
            config.getString("replication-strategy"),
            config.getInt("replication-factor"),
            JavaConverters.asScalaBuffer(config.getStringList("data-center-replication-factors")));
        String keyspace = config.getString("keyspace");

        return Vector.of(
            "CREATE KEYSPACE IF NOT EXISTS " + keyspace + " WITH REPLICATION = { 'class' : " + replStrategy + " } ",
            "CREATE TABLE IF NOT EXISTS " + keyspace + ".meta (datacenter text, tag text, lastEventOffset bigint, PRIMARY KEY(datacenter, tag))",
            "CREATE TABLE IF NOT EXISTS " + keyspace + ".visibility (persistenceid text PRIMARY KEY, master boolean, datacenters set<text>)"
        );
    }
}

@@ -40,16 +40,87 @@
 tion;%0A%0A%0A
+import static scala.collection.JavaConverters.asScalaBufferConverter;%0A%0A
 import c
@@ -331,48 +331,8 @@
 tor;
-%0Aimport scala.collection.JavaConverters;
 %0A%0Apu
@@ -1091,31 +1091,16 @@
         
-JavaConverters.
 asScalaB
@@ -1104,16 +1104,25 @@
 laBuffer
+Converter
 (config.
@@ -1170,16 +1170,26 @@
 ctors%22))
+.asScala()
 );%0A     

Revert part of e9d0552 since it breaks 2.11 compatibility