//设置package包名称以及导入依赖的类
package akka.stream.alpakka.cassandra.scaladsl

import akka.Done
import akka.stream.scaladsl.{Flow, Keep, Sink}
import com.datastax.driver.core.{BoundStatement, PreparedStatement, Session}

import scala.concurrent.{ExecutionContext, Future}

import akka.stream.alpakka.cassandra.GuavaFutures._

object CassandraSink {
  def apply[T](
      parallelism: Int,
      statement: PreparedStatement,
      statementBinder: (T, PreparedStatement) => BoundStatement
  )(implicit session: Session, ex: ExecutionContext): Sink[T, Future[Done]] =
    Flow[T]
      .mapAsyncUnordered(parallelism)(t ? session.executeAsync(statementBinder(t, statement)).asScala())
      .toMat(Sink.ignore)(Keep.right)
} 

//设置package包名称以及导入依赖的类
package sample.helloworldconsumer.impl

import akka.Done
import com.datastax.driver.core.{BoundStatement, PreparedStatement}
import com.lightbend.lagom.scaladsl.persistence.ReadSideProcessor.ReadSideHandler
import com.lightbend.lagom.scaladsl.persistence.cassandra.{CassandraReadSide, CassandraSession}
import com.lightbend.lagom.scaladsl.persistence.{AggregateEventTag, EventStreamElement, ReadSideProcessor}

import scala.concurrent.{ExecutionContext, Future}


class MessageEventProcessor(cassandraReadSide: CassandraReadSide, cassandraSession: CassandraSession)
                           (implicit ec: ExecutionContext) extends ReadSideProcessor[MessageEvent] {

  private var insertWordStmt: PreparedStatement = _

  override def buildHandler(): ReadSideHandler[MessageEvent] = {
    cassandraReadSide.builder[MessageEvent]("message_offset")
      .setGlobalPrepare(createTable)
      .setPrepare { tags =>
        prepareStatements
      }
      .setEventHandler[MessageSaved](insertWord)
      .build()
  }

  override def aggregateTags: Set[AggregateEventTag[MessageEvent]] = Set(MessageEventTag.INSTANCE)

  private def createTable(): Future[Done] = {
    for {
      _ <- cassandraSession.executeCreateTable(
        """        CREATE TABLE IF NOT EXISTS wordcounttest (
                      words text,
                      insertion_time timestamp,
                      PRIMARY KEY (words,insertion_time)
                    )WITH CLUSTERING ORDER BY (insertion_time DESC)
        """)
    } yield Done
  }

  private def prepareStatements(): Future[Done] = {
    for {
      insert <- cassandraSession.prepare(
        """insert into wordcounttest(words ,insertion_time) values(? ,toTimestamp(now())) """)
    } yield {
      insertWordStmt = insert
      Done
    }
  }

  private def insertWord(started: EventStreamElement[MessageSaved]): Future[List[BoundStatement]] = {
    Future.successful {
     val words = started.event.msg.replaceAll("[^\\p{L}\\p{Nd}]+", " ").split(" ").toList
     words.map{ word=> insertWordStmt.bind(word) }
    }
  }

} 

//设置package包名称以及导入依赖的类
package akka.stream.alpakka.cassandra.javadsl

import java.util.concurrent.CompletionStage
import java.util.function.BiFunction

import akka.Done
import akka.stream.javadsl.Sink
import com.datastax.driver.core.{BoundStatement, PreparedStatement, Session}
import akka.stream.alpakka.cassandra.scaladsl.{CassandraSink => ScalaCSink}

import scala.compat.java8.FutureConverters._
import scala.concurrent.ExecutionContext

object CassandraSink {

  def create[T](parallelism: Int,
                statement: PreparedStatement,
                statementBinder: BiFunction[T, PreparedStatement, BoundStatement],
                session: Session,
                executionContext: ExecutionContext): Sink[T, CompletionStage[Done]] = {
    val sink =
      ScalaCSink.apply[T](parallelism, statement, (t, p) => statementBinder.apply(t, p))(session, executionContext)

    sink.mapMaterializedValue(_.toJava).asJava
  }

} 

//设置package包名称以及导入依赖的类
package io.digitalcat.publictransportation.services.identity.impl

import java.util.UUID

import akka.Done
import com.datastax.driver.core.PreparedStatement
import com.lightbend.lagom.scaladsl.persistence.ReadSideProcessor.ReadSideHandler
import com.lightbend.lagom.scaladsl.persistence.cassandra.{CassandraReadSide, CassandraSession}
import com.lightbend.lagom.scaladsl.persistence.{AggregateEventTag, EventStreamElement, ReadSideProcessor}

import scala.concurrent.{ExecutionContext, Future}

class IdentityEventProcessor(session: CassandraSession, readSide: CassandraReadSide)(implicit ec: ExecutionContext) extends ReadSideProcessor[IdentityEvent] {
  private var insertUserStatement: PreparedStatement = _
  private var reportIdToReservedUsernamesStatement: PreparedStatement = _
  private var reportIdToReservedEmailsStatement: PreparedStatement = _

  override def buildHandler(): ReadSideHandler[IdentityEvent] = {
    readSide.builder[IdentityEvent]("identityEventOffset")
      .setPrepare { tag =>
        prepareStatements()
      }.setEventHandler[UserCreated](insertUser)
      .build()
  }

  override def aggregateTags: Set[AggregateEventTag[IdentityEvent]] = {
    IdentityEvent.Tag.allTags
  }

  private def prepareStatements(): Future[Done] = {
    for {
      insertUser <- session.prepare("INSERT INTO users(id, client_id, username, email, first_name, last_name, hashed_password) VALUES (?, ?, ?, ?, ?, ?, ?)")
      reportIdToReservedUsernames <- session.prepare("UPDATE reserved_usernames SET user_id = ? WHERE username = ?")
      reportIdToReservedEmails <- session.prepare("UPDATE reserved_emails SET user_id = ? WHERE email = ?")
    } yield {
      insertUserStatement = insertUser
      reportIdToReservedUsernamesStatement = reportIdToReservedUsernames
      reportIdToReservedEmailsStatement = reportIdToReservedEmails
      Done
    }
  }

  private def insertUser(user: EventStreamElement[UserCreated]) = {
    Future.successful(
      List(
        insertUserStatement.bind(
          user.event.userId,
          UUID.fromString(user.entityId),
          user.event.username,
          user.event.email,
          user.event.firstName,
          user.event.lastName,
          user.event.hashedPassword
        ),
        reportIdToReservedUsernamesStatement.bind(user.event.userId, user.event.username),
        reportIdToReservedEmailsStatement.bind(user.event.userId, user.event.email)
      )
    )
  }
} 

//设置package包名称以及导入依赖的类
package com.livelygig.product.message.impl

import akka.Done
import com.datastax.driver.core.PreparedStatement
import com.lightbend.lagom.scaladsl.persistence.ReadSideProcessor
import com.lightbend.lagom.scaladsl.persistence.cassandra.{CassandraReadSide, CassandraSession}
import com.livelygig.product.message.api.Message

import scala.concurrent.{ExecutionContext, Future}


private[impl] class MessageEventProcessor(session: CassandraSession, readSide: CassandraReadSide)(implicit ec: ExecutionContext)
  extends ReadSideProcessor[MessageEvent] {
  private var insertMessageStatement: PreparedStatement = null

  override def buildHandler() = {
    readSide.builder[MessageEvent]("MessageEventOffset")
      .setGlobalPrepare(createTables)
      .setPrepare(_ => preparedStatements())
      .setEventHandler[MessagePosted](e => insertMessage(e.event.message))
      .build
  }

  override def aggregateTags = MessageEvent.Tag.allTags

  def createTables() = {
    for {
      _ <- session.executeCreateTable(
        """
          CREATE TABLE IF NOT EXISTS msg (
          userId UUID,
          id UUID,
          content text,
          posttime timestamp,
          PRIMARY KEY ((userId), posttime)
          ) WITH CLUSTERING ORDER BY (posttime DESC)
        """
      )
    } yield {
      Done
    }
  }

  def preparedStatements() = {
    for {
      insertMessageStmt <- session.prepare("INSERT INTO msg(userId,id,content,posttime) VALUES (?, ?, ? ,?)")
    } yield {
      insertMessageStatement = insertMessageStmt
      Done
    }
  }

  def insertMessage(message: Message) = {
    Future.successful(List(insertMessageStatement.bind(message.userId, message.id, message.content, message.posttime)))
  }
}