//设置package包名称以及导入依赖的类
package com.bwsw.tstreams.velocity

import java.net.InetSocketAddress
import java.util.UUID

import com.datastax.driver.core.Cluster

import scala.collection.mutable.ListBuffer

object Validator {
  def isSorted(list: ListBuffer[UUID]): Boolean = {
    if (list.isEmpty)
      return true
    var curVal = list.head
    var v = 0
    list foreach { el =>
      if (el.timestamp() < curVal.timestamp()) {
        println(s"value=$v")
        return false
      }
      if (el.timestamp() > curVal.timestamp())
        curVal = el
      v += 1
    }
    true
  }

  def main(args: Array[String]) {
    //    if (args.length != 1)
    //      throw new IllegalArgumentException("specify [keyspace]")
    //    val keyspace = args(0)

    val cluster = Cluster.builder().addContactPointsWithPorts(new InetSocketAddress("176.120.27.82", 9042)).build()
    val session = cluster.connect()

    val set = session.execute(s"select * from velocity.commit_log").all()
    val it = set.iterator()
    val buffers = scala.collection.mutable.Map[Int, ListBuffer[UUID]]()

    while (it.hasNext) {
      val row = it.next()
      val partition = row.getInt("partition")
      val uuid = row.getUUID("transaction")
      if (!buffers.contains(partition)) {
        buffers(partition) = ListBuffer(uuid)
      } else {
        buffers(partition) += uuid
      }
    }

    val checkVal = buffers.map(x => isSorted(x._2)).reduceLeft((a, b) => a & b)

    if (checkVal)
      println("sorted")
    else
      println("not sorted")

    cluster.close()
    session.close()
  }
} 

//设置package包名称以及导入依赖的类
package connector

import com.datastax.driver.core.{Session, Cluster}
import com.websudos.phantom.connectors.{KeySpace, SessionProvider}


trait CassandraConnector extends SessionProvider {

  implicit val space: KeySpace = Connector.keySpace
  val cluster = Connector.cluster
  override implicit lazy val session = Connector.session
}

object Connector {
  val keySpace = KeySpace("MovieLens")
  val cluster = Cluster.builder().addContactPoint("192.168.99.100").build()
  val session = cluster.connect(keySpace.name)
} 

//设置package包名称以及导入依赖的类
package co.verdigris.spark.connector.cql

import com.datastax.driver.core.Cluster

class AwsS3USWest1ConnectionFactoryTest extends ConnectionFactorySpec {
  override def beforeAll {
    super.beforeAll

    factory = AwsS3USWest1ConnectionFactory
  }

  describe(".clusterBuilder") {
    it("should return a new Cluster.Builder instance") {
      factory.clusterBuilder(cassandraConf) shouldBe a [Cluster.Builder]
    }
  }

  describe(".createCluster") {
    it("should return a new Cluster instance") {
      factory.createCluster(cassandraConf) shouldBe a [Cluster]
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.bwsw.tstreams.velocity

import com.bwsw.tstreams.common.CassandraHelper
import com.datastax.driver.core.Cluster

object MetadataCreator {
  def main(args: Array[String]) {
    import Common._
    val cluster = Cluster.builder().addContactPoint("localhost").build()
    val session = cluster.connect()
    try {
      session.execute(s"DROP KEYSPACE $keyspace")
    }
    catch {
      case e: Exception => println(s"msg=${e.getMessage}")
    }
    CassandraHelper.createKeyspace(session, keyspace)
    CassandraHelper.createMetadataTables(session, keyspace)
    cluster.close()
    session.close()
  }
} 

//设置package包名称以及导入依赖的类
package conf.connection

import java.net.InetAddress

import com.datastax.driver.core.Cluster
import com.typesafe.config.ConfigFactory
import com.websudos.phantom.connectors.{KeySpace, SessionProvider}
import com.websudos.phantom.dsl.Session

import scala.collection.JavaConversions._



object Config {
  val config = ConfigFactory.load()

}

trait DefaultsConnector extends SessionProvider {
  val config = ConfigFactory.load()
  implicit val space: KeySpace = DataConnection.keySpace
  val cluster = DataConnection.cluster
  override implicit lazy val session: Session = DataConnection.session
}

object DataConnection {
  val config = ConfigFactory.load()
  val hosts: Seq[String] = Config.config.getStringList("cassandra.host").toList
  val inets = hosts.map(InetAddress.getByName)
  val keySpace: KeySpace = KeySpace(Config.config.getString("cassandra.keyspace"))
  val cluster =
    Cluster.builder()
      .addContactPoints(inets)
      .withClusterName(Config.config.getString("cassandra.cluster"))
//      .withCredentials(config.getString("cassandra.username"), config.getString("cassandra.password"))
      .build()
  val session: Session = cluster.connect(keySpace.name)
} 

//设置package包名称以及导入依赖的类
package conf.connection

import java.net.InetAddress

import com.datastax.driver.core.Cluster
import com.mongodb.casbah.{MongoDB, MongoClientURI, MongoClient}
import com.typesafe.config.ConfigFactory
import com.websudos.phantom.connectors.{KeySpace, SessionProvider}
import com.websudos.phantom.dsl.Session

import scala.collection.JavaConversions._



object Config {
  val config = ConfigFactory.load()

}

object HashDB {
  def getConnection(): MongoDB={
    val hosts = Config.config.getString("mongodb.host")
    val database = Config.config.getString("mongodb.database")
    val uri = MongoClientURI(hosts)
    MongoClient(uri)(database)
  }
}

trait DefaultsConnector extends SessionProvider {
  val config = ConfigFactory.load()
  implicit val space: KeySpace = DataConnection.keySpace
  val cluster = DataConnection.cluster
  override implicit lazy val session: Session = DataConnection.session
}

object DataConnection {
  val config = ConfigFactory.load()
  val hosts: Seq[String] = Config.config.getStringList("cassandra.host").toList
  val inets = hosts.map(InetAddress.getByName)
  val keySpace: KeySpace = KeySpace(Config.config.getString("cassandra.keyspace"))

  val cluster =
    Cluster.builder()
      .addContactPoints(inets)
      .withClusterName(Config.config.getString("cassandra.cluster"))
//      .withCredentials(config.getString("cassandra.username"), config.getString("cassandra.password"))
      .build()
  val session: Session = cluster.connect(keySpace.name)
} 

//设置package包名称以及导入依赖的类
package data

import javax.inject.{Inject, Singleton}

import com.datastax.driver.core.{Cluster, Session}
import com.typesafe.config.Config
import io.getquill.{CassandraAsyncContext, SnakeCase}
import play.api.inject.ApplicationLifecycle
import play.api.{Configuration, Environment, Logger}
import util.FutureUtils.toFutureUnit

import scala.concurrent.ExecutionContext

@Singleton
class CassandraAsyncContextImpl @Inject()(cassandra: CassandraCluster, conf: CassandraConfig)
  extends CassandraAsyncContext[SnakeCase](cassandra.cluster, conf.keySpace, 100L)

@Singleton
class CassandraCluster @Inject()(conf: CassandraConfig, appLifecycle: ApplicationLifecycle)(implicit executor: ExecutionContext) {

  import conf._

  private[data] val cluster =
    Cluster.builder()
      .addContactPoints(hosts: _*)
      .withPort(port)
      .build()

  private[data] def noKeySpaceSession: Session = cluster.connect()

  private[data] def stop() = toFutureUnit(cluster.closeAsync())

  Logger.info(s"Cassandra host to be used: '${hosts.mkString(",")}' with port:$port")
  appLifecycle.addStopHook(() => stop())
}

@Singleton
class CassandraConfig @Inject()(configuration: Configuration, environment: Environment) {
  val config: Config = configuration.underlying

  val keySpace = config.getString("devgym.db.cassandra.keyspace")
  val port = config.getInt("devgym.db.cassandra.port")

  val hosts: Seq[String] =
    configuration.getStringSeq("devgym.db.cassandra.hosts").get
} 

//设置package包名称以及导入依赖的类
package hmda.query.cassandra

import com.datastax.driver.core.{ Cluster, Session }
import org.scalatest.{ BeforeAndAfterAll, MustMatchers, WordSpec }
import org.cassandraunit.CQLDataLoader
import org.cassandraunit.dataset.cql.ClassPathCQLDataSet
import org.cassandraunit.utils.EmbeddedCassandraServerHelper

class CassandraSpec extends WordSpec with MustMatchers with BeforeAndAfterAll {

  var cluster: Cluster = _
  var session: Session = _

  override def beforeAll(): Unit = {
    EmbeddedCassandraServerHelper.startEmbeddedCassandra(60000L)
    cluster = EmbeddedCassandraServerHelper.getCluster
    session = cluster.connect()
    loadData()
  }

  override def afterAll(): Unit = {
    EmbeddedCassandraServerHelper.cleanEmbeddedCassandra()
  }

  def loadData(): Unit = {
    val dataLoader = new CQLDataLoader(session)
    dataLoader.load(new ClassPathCQLDataSet("simple.cql", "hmda_query"))
  }

  "Cassandra" must {
    "Select from table" in {
      val resultSet = session.execute("select * from myTable where id = 'myKey01'")
      resultSet.iterator().next().getString("value") mustBe "myValue01"
    }
  }

} 

//设置package包名称以及导入依赖的类
package io.gzet.story.web.dao

import com.datastax.driver.core.Cluster
import io.gzet.story.model.Article
import io.gzet.story.util.SimhashUtils._
import io.gzet.story.web.SimpleConfig

import scala.collection.JavaConversions._
import scala.language.postfixOps

class CassandraDao() extends SimpleConfig {

  private val cluster = Cluster.builder().addContactPoint(cassandraHost).withPort(cassandraPort).build()
  val session = cluster.connect()

  def count(): Long = {
    val stmt = s"SELECT count(*) FROM $cassandraKeyspace.$cassandraTable;"
    val results = session.execute(stmt).all()
    results map { row =>
      row.getLong(0)
    } head
  }

  def findDuplicates(hash: Int): List[Article] = {
    searchmasks flatMap { mask =>
      val searchHash = mask ^ hash
      val stmt = s"SELECT hash, url, title, body FROM $cassandraKeyspace.$cassandraTable WHERE hash = $searchHash;"
      val results = session.execute(stmt).all()
      results map { row =>
        Article(row.getInt("hash"), row.getString("body"), row.getString("title"), row.getString("url"))
      }
    } toList
  }

} 

//设置package包名称以及导入依赖的类
package conf.connection

import java.net.InetAddress

import com.datastax.driver.core.Cluster
import com.typesafe.config.ConfigFactory
import com.websudos.phantom.connectors.{KeySpace, SessionProvider}
import com.websudos.phantom.dsl.Session

import scala.collection.JavaConversions._



object Config {
  val config = ConfigFactory.load()

}

trait DefaultsConnector extends SessionProvider {
  val config = ConfigFactory.load()
  implicit val space: KeySpace = DataConnection.keySpace
  val cluster = DataConnection.cluster
  override implicit lazy val session: Session = DataConnection.session
}

object DataConnection {
  val config = ConfigFactory.load()
  val hosts: Seq[String] = Config.config.getStringList("cassandra.host").toList
  val inets = hosts.map(InetAddress.getByName)
  val keySpace: KeySpace = KeySpace(Config.config.getString("cassandra.keyspace"))

  val cluster =
    Cluster.builder()
      .addContactPoints(inets)
      .withClusterName(Config.config.getString("cassandra.cluster"))
//      .withCredentials(config.getString("cassandra.username"), config.getString("cassandra.password"))
      .build()
  val session: Session = cluster.connect(keySpace.name)
} 

//设置package包名称以及导入依赖的类
package com.knoldus

import com.datastax.driver.core.{Cluster, ResultSet, Session}
import com.knoldus.utils.ConfigReader

import scala.collection.JavaConversions
import scala.util.{Failure, Success, Try}




object Launcher extends App {

  Try {
    CassandraConnector.getCasssandraBuilder
  } match {
    case Success(cluster) =>
      val session = getCassandraSession(ConfigReader.getKeyspaceName, cluster)
      QueryHandler.runQuery(session)
      session.close()
      cluster.close()
    case Failure(exception) => println("Unable to Connect to Cassandra" + exception)
  }

  private def getCassandraSession(keyspace: String, cluster: Cluster): Session = {
    Try{
      CassandraConnector.getSession(keyspace, cluster)
    } match {
      case Success(session) => session
      case Failure(exception) => throw new Exception("Unable to connect to keyspace" + exception)
    }
  }


} 

//设置package包名称以及导入依赖的类
package com.yper.utils

import com.datastax.driver.core.{Cluster, Session}


object DatabaseUtil {

  val keyspace = "yper"

  def getSession: Session = DatabaseUtil.synchronized {
    try {
      val builder = Cluster
        .builder()
        .addContactPoint("localhost")
        .withPort(9042)
        .build()
      builder.connect(keyspace)
    } catch {
      case e: Exception => {
        println(e.getMessage)
        throw e
      }
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.bwsw.sj.common.utils

import java.net.InetSocketAddress

import com.bwsw.tstreams.common.{CassandraConnectorConf, CassandraHelper}
import com.bwsw.tstreams.data.cassandra.Factory
import com.bwsw.tstreams.metadata.MetadataStorageFactory
import com.datastax.driver.core.{Cluster, Session}

class CassandraFactory {

  import scala.collection.JavaConverters._

  private var cluster: Cluster = null
  private var session: Session = null
  private var cassandraConnectorConf: CassandraConnectorConf = null
  private val metadataStorageFactory = new MetadataStorageFactory()
  private val dataStorageFactory = new Factory()

  def open(hosts: Set[(String, Int)]) = {
    val cassandraHosts = hosts.map(s => new InetSocketAddress(s._1, s._2))
    cluster = Cluster.builder().addContactPointsWithPorts(cassandraHosts.toList.asJava).build()
    session = cluster.connect()
    cassandraConnectorConf = CassandraConnectorConf.apply(cassandraHosts)
  }

  def getDataStorage(keyspace: String) = {
    dataStorageFactory.getInstance(
      cassandraConnectorConf,
      keyspace = keyspace)
  }

  def getMetadataStorage(keyspace: String) = {
    metadataStorageFactory.getInstance(
      cassandraConnectorConf,
      keyspace = keyspace)
  }

  def createKeyspace(keyspace: String) = {
    CassandraHelper.createKeyspace(session, keyspace)
  }

  def createDataTable(keyspace: String) = {
    CassandraHelper.createDataTable(session, keyspace)
  }

  def createMetadataTables(keyspace: String) = {
    CassandraHelper.createMetadataTables(session, keyspace)
  }

  def dropKeyspace(keyspace: String) = {
    CassandraHelper.dropKeyspace(session, keyspace)
  }

  def close() = {
    metadataStorageFactory.closeFactory()
    session.close()
    cluster.close()
  }
} 

//设置package包名称以及导入依赖的类
package demo6

import java.util.UUID
import com.datastax.driver.core.{Row, Cluster, Session}
import troy.dsl._
import troy.driver.DSL._

import scala.concurrent.Await
import scala.concurrent.duration.Duration

case class Post(id: UUID, title: String)

object Main extends App {
  import scala.concurrent.ExecutionContext.Implicits.global

  val port: Int = 9042
  val host: String = "127.0.0.1"

  private val cluster =
    new Cluster.Builder().addContactPoints(host).withPort(port).build()

  implicit val session: Session = cluster.connect()

  val create = withSchema {
    (authorId: String, title: String) =>
      cql"""
         INSERT INTO test.posts (author_id , post_id , post_title )
         VALUES ( $authorId, now(), $title);
       """.prepared.executeAsync
  }

  val listByAuthor = withSchema {
    (authorId: String) =>
      cql"""
         SELECT post_id, post_title
         FROM test.posts
         WHERE author_id = $authorId
       """
        .prepared
        .executeAsync
        .as(Post)
  }

  println(Await.result(create("test", "title"), Duration(1, "second")))
  println(Await.result(listByAuthor("test"), Duration(1, "second")))

  session.close()
  cluster.close()
} 

//设置package包名称以及导入依赖的类
package demo2

import java.util.UUID
import com.datastax.driver.core.{Row, Cluster, Session}
import troy.dsl._
import troy.driver.DSL._

import scala.concurrent.Await
import scala.concurrent.duration.Duration

case class Post(id: UUID, title: String)

object Main extends App {
  import scala.concurrent.ExecutionContext.Implicits.global

  val port: Int = 9042
  val host: String = "127.0.0.1"

  private val cluster =
    new Cluster.Builder().addContactPoints(host).withPort(port).build()

  implicit val session: Session = cluster.connect()

  val listByAuthor = withSchema {
    (authorId: String) =>
      cql"""
         SELECT post_id, post_title
         FROM test.posts
         WHERE author_id = $authorId
       """
        .prepared
        .executeAsync
        .as(Post)
  }

  val result = listByAuthor("test")
  println(Await.result(result, Duration(1, "second")))

  session.close()
  cluster.close()
} 

//设置package包名称以及导入依赖的类
package demo1

import com.datastax.driver.core.{Cluster, Session}
import troy.dsl._
import troy.driver.DSL._
import scala.concurrent.ExecutionContext.Implicits.global

object Main extends App {
  val port: Int = 9042
  val host: String = "127.0.0.1"

  private val cluster =
    new Cluster.Builder().addContactPoints(host).withPort(port).build()

  implicit val session: Session = cluster.connect()

  val listByAuthor = withSchema {
    (authorId: String) =>
      cql"""
         SELECT post_id, post_title
         FROM test.posts
         WHERE author_id = $authorId
       """.prepared.executeAsync
  }


  println(listByAuthor("test"))

  session.close()
  cluster.close()
} 

//设置package包名称以及导入依赖的类
package demo4

import java.util.UUID
import com.datastax.driver.core.{Cluster, Session}
import troy.dsl._
import scala.concurrent.Await
import scala.concurrent.duration.Duration

case class Post(id: UUID, title: String)

object Main extends App {
  import scala.concurrent.ExecutionContext.Implicits.global

  val port: Int = 9042
  val host: String = "127.0.0.1"

  private val cluster =
    new Cluster.Builder().addContactPoints(host).withPort(port).build()

  implicit val session: Session = cluster.connect()

  val listByTag = withSchema {
    (tag: String) =>
      cql"""
         SELECT post_id, post_title
         FROM test.posts
         WHERE post_tags CONTAINS $tag
       """.prepared.as(Post)
  }

  val titlesF = listByTag("test").map(_.map(_.title))
  val titles = Await.result(titlesF, Duration(1, "second"))
  println("Matching titles:")
  titles.foreach(println)

  session.close()
  cluster.close()
} 

//设置package包名称以及导入依赖的类
package demo5

import java.util.UUID
import com.datastax.driver.core.{Cluster, Session}
import troy.dsl._
import scala.concurrent.Await
import scala.concurrent.duration.Duration

case class Post(id: UUID, comments: Map[Int, String])

object Main extends App {
  import scala.concurrent.ExecutionContext.Implicits.global

  val port: Int = 9042
  val host: String = "127.0.0.1"

  private val cluster =
    new Cluster.Builder().addContactPoints(host).withPort(port).build()

  implicit val session: Session = cluster.connect()

  val getCommentsByLine = withSchema {
    (authorId: String, postId: UUID, line: Int) =>
      cql"""
         SELECT post_id, comments
         FROM test.posts
         WHERE author_id = $authorId
           AND post_id = $postId
           AND comments CONTAINS KEY $line
       """.prepared.as(Post)
  }

  val postId = UUID.fromString("a4a70900-24e1-11df-8924-001ff3591711")
  println(Await.result(getCommentsByLine("test", postId, 5), Duration(1, "second")))

  session.close()
  cluster.close()
} 

//设置package包名称以及导入依赖的类
package demo3

import java.util.UUID
import com.datastax.driver.core.{Row, Cluster, Session}
import troy.dsl._
import troy.driver.DSL._

import scala.concurrent.Await
import scala.concurrent.duration.Duration

case class Post(id: UUID, title: String)

object Main extends App {
  import scala.concurrent.ExecutionContext.Implicits.global

  val port: Int = 9042
  val host: String = "127.0.0.1"

  private val cluster =
    new Cluster.Builder().addContactPoints(host).withPort(port).build()

  implicit val session: Session = cluster.connect()

  val get = withSchema {
    (authorId: String, postId: UUID) =>
      cql"""
         SELECT post_id, post_title
         FROM test.posts
         WHERE author_id = $authorId AND post_id = $postId
       """
        .prepared
        .executeAsync
        .oneOption
        .as(Post)
  }

  val result = get("test", UUID.fromString("a4a70900-24e1-11df-8924-001ff3591711"))
  val maybePost: Option[Post] = Await.result(result, Duration(1, "second"))
  println(maybePost.map(_.title).getOrElse("Post not found"))

  session.close()
  cluster.close()
} 

//设置package包名称以及导入依赖的类
package demo1

import java.util.UUID
import com.datastax.driver.core.{Cluster, Session}
import troy.dsl._
import scala.concurrent.Await
import scala.concurrent.duration.Duration

case class Post(id: UUID, img: String) // Map doesn't work with Primitives yet. See: https://github.com/cassandra-scala/troy/issues/18

object Main extends App {
  import scala.concurrent.ExecutionContext.Implicits.global

  val port: Int = 9042
  val host: String = "127.0.0.1"

  private val cluster =
    new Cluster.Builder().addContactPoints(host).withPort(port).build()

  implicit val session: Session = cluster.connect()

  val getCommentsByLine = withSchema.minVersion(2).maxVersion(2) {
    (authorId: String, postId: UUID) =>
      cql"""
         SELECT post_id, post_img, foo
         FROM test.posts
         WHERE author_id = $authorId
           AND post_id = $postId
       """.prepared.as(Post)
  }

  val postId = UUID.fromString("a4a70900-24e1-11df-8924-001ff3591711")
  println(Await.result(getCommentsByLine("test", postId), Duration(1, "second")))

  session.close()
  cluster.close()
}