PICE（3）：CassandraStreaming，
在上一篇博文里我们介绍了通过gRPC实现JDBC数据库的streaming，这篇我们介绍关于cassandra的streaming实现方式。如果我们需要从一个未部署cassandra的节点或终端上读取cassandra数据，可以用gRPC来搭建一个数据桥梁来连接这两端。这时cassandra这端就是gRPC-Server端，由它提供cassandra的数据服务。
在前面sdp系列讨论里我们已经实现了Cassandra-Engine。它的运作原理还是通过某种Context把指令提交给cassandra去执行。我们先设计一个创建库表的例子。CQL语句和Cassandra-Engine程序代码如下，这是客户端部分：
val dropCQL = "DROP TABLE IF EXISTS testdb.AQMRPT"
val createCQL ="""
CREATE TABLE testdb.AQMRPT (
rowid bigint primary key,
measureid bigint,
statename text,
countyname text,
reportyear int,
value int,
created timestamp
)"""
val cqlddl = CQLUpdate(statements = Seq(dropCQL,createCQL))
def createTbl: Source[CQLResult,NotUsed] = {
log.info(s"running createTbl ...")
Source
.single(cqlddl)
.via(stub.runDDL)
}
首先，我们在CQLUpdate这个protobuf对应Context里传入两条指令dropCQL和createCQL，可以预计这会是一种批次型batch方式。然后一如既往，我们使用了streaming编程模式。在.proto文件里用DDL来对应Context和Service：
message CQLUpdate {
repeated string statements = 1;
bytes parameters = 2;
google.protobuf.Int32Value consistency = 3;
google.protobuf.BoolValue batch = 4;
}
service CQLServices {
rpc runDDL(CQLUpdate) returns (CQLResult) {}
}
runDDL程序实现如下：
override def runDDL: Flow[CQLUpdate, CQLResult, NotUsed] = {
Flow[CQLUpdate]
.flatMapConcat { context =>
//unpack CQLUpdate and construct the context
val ctx = CQLContext(context.statements)
log.info(s"**** CQLContext => ${ctx} ***")
Source
.fromFuture(cqlExecute(ctx))
.map { r => CQLResult(marshal(r)) }
}
}这里我们调用了Cassandra-Engine的cqlExecute(ctx)函数：
def cqlExecute(ctx: CQLContext)(
implicit session: Session, ec: ExecutionContext): Future[Boolean] = {
var invalidBat = false
if ( ctx.batch ) {
if (ctx.parameters == Nil)
invalidBat = true
else if (ctx.parameters.size < 2)
invalidBat = true;
}
if (!ctx.batch || invalidBat) {
if(invalidBat)
log.warn(s"cqlExecute> batch update must have at least 2 sets of parameters! change to single-command.")
if (ctx.statements.size == 1) {
var param: Seq[Object] = Nil
if (ctx.parameters != Nil) param = ctx.parameters.head
log.info(s"cqlExecute> single-command: statement: ${ctx.statements.head} parameters: ${param}")
cqlSingleUpdate(ctx.consistency, ctx.statements.head, param)
}
else {
var params: Seq[Seq[Object]] = Nil
if (ctx.parameters == Nil)
params = Seq.fill(ctx.statements.length)(Nil)
else {
if (ctx.statements.size > ctx.parameters.size) {
log.warn(s"cqlExecute> fewer parameters than statements! pad with 'Nil'.")
val nils = Seq.fill(ctx.statements.size - ctx.parameters.size)(Nil)
params = ctx.parameters ++ nils
}
else
params = ctx.parameters
}
val commands: Seq[(String,Seq[Object])] = ctx.statements zip params
log.info(s"cqlExecute> multi-commands: ${commands}")
/*
//using sequence to flip List[Future[Boolean]] => Future[List[Boolean]]
//therefore, make sure no command replies on prev command effect
val lstCmds: List[Future[Boolean]] = commands.map { case (stmt,param) =>
cqlSingleUpdate(ctx.consistency, stmt, param)
}.toList
val futList = lstCmds.sequence.map(_ => true) //must map to execute
*/
/*
//using traverse to have some degree of parallelism = max(runtimes)
//therefore, make sure no command replies on prev command effect
val futList = Future.traverse(commands) { case (stmt,param) =>
cqlSingleUpdate(ctx.consistency, stmt, param)
}.map(_ => true)
Await.result(futList, 3 seconds)
Future.successful(true)
*/
// run sync directly
Future {
commands.foreach { case (stm, pars) =>
cqlExecuteSync(ctx.consistency, stm, pars)
}
true
}
}
}
else
cqlBatchUpdate(ctx)
}
特别展示了这个函数的代码是因为对于一批次多条指令可能会涉及到non-blocking和并行计算。可参考上面代码标注段落里函数式方法（cats）sequence,traverse如何实现对一串Future的运算。
下一个例子是用流方式把JDBC数据库数据并入cassandra数据库里。.proto DDL内容如下：
message ProtoDate {
int32 yyyy = 1;
int32 mm = 2;
int32 dd = 3;
}
message ProtoTime {
int32 hh = 1;
int32 mm = 2;
int32 ss = 3;
int32 nnn = 4;
}
message ProtoDateTime {
ProtoDate date = 1;
ProtoTime time = 2;
}
message AQMRPTRow {
int64 rowid = 1;
string countyname = 2;
string statename = 3;
int64 measureid = 4;
int32 reportyear = 5;
int32 value = 6;
ProtoDateTime created = 7;
}
message CQLResult {
bytes result = 1;
}
message CQLUpdate {
repeated string statements = 1;
bytes parameters = 2;
google.protobuf.Int32Value consistency = 3;
google.protobuf.BoolValue batch = 4;
}
service CQLServices {
rpc transferRows(stream AQMRPTRow) returns (stream CQLResult) {}
rpc runDDL(CQLUpdate) returns (CQLResult) {}
}
下面是服务函数的实现：
val toParams: AQMRPTRow => Seq[Object] = row => Seq[Object](
row.rowid.asInstanceOf[Object],
row.measureid.asInstanceOf[Object],
row.statename,
row.countyname,
row.reportyear.asInstanceOf[Object],
row.value.asInstanceOf[Object],
CQLDateTimeNow
)
val cqlInsert ="""
|insert into testdb.AQMRPT(
| rowid,
| measureid,
| statename,
| countyname,
| reportyear,
| value,
| created)
| values(?,?,?,?,?,?,?)
""".stripMargin
val cqlActionStream = CassandraActionStream(cqlInsert,toParams).setParallelism(2)
.setProcessOrder(false)
/*
val cqlActionFlow: Flow[AQMRPTRow,AQMRPTRow,NotUsed] =
Flow[AQMRPTRow]
.via(cqlActionStream.performOnRow)
*/
val cqlActionFlow: Flow[AQMRPTRow,CQLResult,NotUsed] = {
Flow[AQMRPTRow]
.mapAsync(cqlActionStream.parallelism){ row =>
if (IfExists(row.rowid))
Future.successful(CQLResult(marshal(0)))
else
cqlActionStream.perform(row).map {_ => CQLResult(marshal(1))}
}
}
override def transferRows: Flow[AQMRPTRow, CQLResult, NotUsed] = {
Flow[AQMRPTRow]
.via(cqlActionFlow)
}
private def IfExists(rowid: Long): Boolean = {
val cql = "SELECT * FROM testdb.AQMRPT WHERE ROWID = ? ALLOW FILTERING"
val param = Seq(rowid.asInstanceOf[Object])
val toRowId: Row => Long = r => r.getLong("rowid")
val ctx = CQLQueryContext(cql,param)
val src: Source[Long,NotUsed] = cassandraStream(ctx,toRowId)
val fut = src.toMat(Sink.headOption)(Keep.right).run()
val result = Await.result(fut,3 seconds)
log.info(s"checking existence: ${result}")
result match {
case Some(x) => true
case None => false
}
}
在上面的代码里我们调用了Cassandra-Engine的CassandraActionStream类型的流处理方法。值得注意的是这里我们尝试在stream Flow里运算另一个Flow，如：IfExists函数里运算一个Source来确定rowid是否存在。不要在意这个函数的实际应用，它只是一个人为的例子。另外，rowid：Long这样的定义是硬性规定的。cassandra对数据类型的匹配要求很弱智，没有提供任何自然转换。所以，Int <> Long被视为类型错误，而且无法catch任何明白的错误信息。
这项服务的客户端调用如下：
val stub = CqlGrpcAkkaStream.stub(channel)
val jdbcRows2transfer = JDBCQueryContext[AQMRPTRow](
dbName = 'h2,
statement = "select * from AQMRPT where statename='Arkansas'"
)
def toAQMRPTRow: WrappedResultSet => AQMRPTRow = rs => AQMRPTRow(
rowid = rs.long("ROWID"),
measureid = rs.long("MEASUREID"),
statename = rs.string("STATENAME"),
countyname = rs.string("COUNTYNAME"),
reportyear = rs.int("REPORTYEAR"),
value = rs.int("VALUE"),
created = Some(ProtoDateTime(Some(ProtoDate(1990, 8, 12)), Some(ProtoTime(23, 56, 23, 0))))
)
import scala.concurrent.duration._
def transferRows: Source[CQLResult, NotUsed] = {
log.info(s"**** calling transferRows ****")
jdbcAkkaStream(jdbcRows2transfer, toAQMRPTRow)
// .throttle(1, 500.millis, 1, ThrottleMode.shaping)
.via(stub.transferRows)
}
注意：JDBC在客户端本地，cassandra是远程服务。
最后我们示范一下cassandra Query。.proto DDL 定义：
message CQLQuery {
string statement = 1;
bytes parameters = 2;
google.protobuf.Int32Value consistency = 3;
google.protobuf.Int32Value fetchSize = 4;
}
service CQLServices {
rpc transferRows(stream AQMRPTRow) returns (stream CQLResult) {}
rpc runQuery(CQLQuery) returns (stream AQMRPTRow) {}
rpc runDDL(CQLUpdate) returns (CQLResult) {}
}
服务函数代码如下：
def toCQLTimestamp(rs: Row) = {
try {
val tm = rs.getTimestamp("CREATED")
if (tm == null) None
else {
val localdt = cqlGetTimestamp(tm)
Some(ProtoDateTime(Some(ProtoDate(localdt.getYear, localdt.getMonthValue, localdt.getDayOfMonth)),
Some(ProtoTime(localdt.getHour, localdt.getMinute, localdt.getSecond, localdt.getNano))))
}
}
catch {
case e: Exception => None
}
}
val toAQMRow: Row => AQMRPTRow = rs=> AQMRPTRow(
rowid = rs.getLong("ROWID"),
measureid = rs.getLong("MEASUREID"),
statename = rs.getString("STATENAME"),
countyname = rs.getString("COUNTYNAME"),
reportyear = rs.getInt("REPORTYEAR"),
value = rs.getInt("VALUE"),
created = toCQLTimestamp(rs)
)
override def runQuery: Flow[CQLQuery, AQMRPTRow, NotUsed] = {
log.info("**** runQuery called on service side ***")
Flow[CQLQuery]
.flatMapConcat { q =>
//unpack JDBCQuery and construct the context
var params: Seq[Object] = Nil
if (q.parameters != _root_.com.google.protobuf.ByteString.EMPTY)
params = unmarshal[Seq[Object]](q.parameters)
log.info(s"**** query parameters: ${params} ****")
val ctx = CQLQueryContext(q.statement,params)
CQLEngine.cassandraStream(ctx,toAQMRow)
}
}
这里值得看看的一是日期转换，二是对于cassandra parameter Seq[Object]的marshal和unmarshal。客户端代码：
val query = CQLQuery(
statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE > ? ALLOW FILTERING;",
parameters = marshal(Seq("Arkansas", 0.toInt))
)
val query2 = CQLQuery (
statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE = ?",
parameters = marshal(Seq("Colorado", 3.toInt))
)
val query3= CQLQuery (
statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE = ?",
parameters = marshal(Seq("Arkansas", 8.toInt))
)
def queryRows: Source[AQMRPTRow,NotUsed] = {
log.info(s"running queryRows ...")
Source
.single(query)
.via(stub.runQuery)
}这段相对直白。
下面就是本次讨论涉及的完整源代码：
project/scalapb.sbt
addSbtPlugin("com.thesamet" % "sbt-protoc" % "0.99.18")
resolvers += Resolver.bintrayRepo("beyondthelines", "maven")
libraryDependencies ++= Seq(
"com.thesamet.scalapb" %% "compilerplugin" % "0.7.4",
"beyondthelines" %% "grpcakkastreamgenerator" % "0.0.5"
)build.sbt
import scalapb.compiler.Version.scalapbVersion
import scalapb.compiler.Version.grpcJavaVersion
name := "gRPCCassandra"
version := "0.1"
scalaVersion := "2.12.6"
resolvers += Resolver.bintrayRepo("beyondthelines", "maven")
scalacOptions += "-Ypartial-unification"
libraryDependencies := Seq(
"com.thesamet.scalapb" %% "scalapb-runtime" % scalapbVersion % "protobuf",
"io.grpc" % "grpc-netty" % grpcJavaVersion,
"com.thesamet.scalapb" %% "scalapb-runtime-grpc" % scalapbVersion,
"io.monix" %% "monix" % "2.3.0",
// for GRPC Akkastream
"beyondthelines" %% "grpcakkastreamruntime" % "0.0.5",
// for scalikejdbc
"org.scalikejdbc" %% "scalikejdbc" % "3.2.1",
"org.scalikejdbc" %% "scalikejdbc-test" % "3.2.1" % "test",
"org.scalikejdbc" %% "scalikejdbc-config" % "3.2.1",
"org.scalikejdbc" %% "scalikejdbc-streams" % "3.2.1",
"org.scalikejdbc" %% "scalikejdbc-joda-time" % "3.2.1",
"com.h2database" % "h2" % "1.4.196",
"mysql" % "mysql-connector-java" % "6.0.6",
"org.postgresql" % "postgresql" % "42.2.0",
"commons-dbcp" % "commons-dbcp" % "1.4",
"org.apache.tomcat" % "tomcat-jdbc" % "9.0.2",
"com.zaxxer" % "HikariCP" % "2.7.4",
"com.jolbox" % "bonecp" % "0.8.0.RELEASE",
"com.typesafe.slick" %% "slick" % "3.2.1",
//for cassandra 340
"com.datastax.cassandra" % "cassandra-driver-core" % "3.4.0",
"com.datastax.cassandra" % "cassandra-driver-extras" % "3.4.0",
"com.typesafe.akka" %% "akka-stream" % "2.5.13",
"com.lightbend.akka" %% "akka-stream-alpakka-cassandra" % "0.19",
"ch.qos.logback" % "logback-classic" % "1.2.3",
"org.typelevel" %% "cats-core" % "1.1.0"
)
PB.targets in Compile := Seq(
scalapb.gen() -> (sourceManaged in Compile).value,
// generate the akka stream files
grpc.akkastreams.generators.GrpcAkkaStreamGenerator() -> (sourceManaged in Compile).value
)
main/resources/application.conf
# JDBC settings
test {
db {
h2 {
driver = "org.h2.Driver"
url = "jdbc:h2:tcp://localhost/~/slickdemo"
user = ""
password = ""
poolInitialSize = 5
poolMaxSize = 7
poolConnectionTimeoutMillis = 1000
poolValidationQuery = "select 1 as one"
poolFactoryName = "commons-dbcp2"
}
}
db.mysql.driver = "com.mysql.cj.jdbc.Driver"
db.mysql.url = "jdbc:mysql://localhost:3306/testdb"
db.mysql.user = "root"
db.mysql.password = "123"
db.mysql.poolInitialSize = 5
db.mysql.poolMaxSize = 7
db.mysql.poolConnectionTimeoutMillis = 1000
db.mysql.poolValidationQuery = "select 1 as one"
db.mysql.poolFactoryName = "bonecp"
# scallikejdbc Global settings
scalikejdbc.global.loggingSQLAndTime.enabled = true
scalikejdbc.global.loggingSQLAndTime.logLevel = info
scalikejdbc.global.loggingSQLAndTime.warningEnabled = true
scalikejdbc.global.loggingSQLAndTime.warningThresholdMillis = 1000
scalikejdbc.global.loggingSQLAndTime.warningLogLevel = warn
scalikejdbc.global.loggingSQLAndTime.singleLineMode = false
scalikejdbc.global.loggingSQLAndTime.printUnprocessedStackTrace = false
scalikejdbc.global.loggingSQLAndTime.stackTraceDepth = 10
}
dev {
db {
h2 {
driver = "org.h2.Driver"
url = "jdbc:h2:tcp://localhost/~/slickdemo"
user = ""
password = ""
poolFactoryName = "hikaricp"
numThreads = 10
maxConnections = 12
minConnections = 4
keepAliveConnection = true
}
mysql {
driver = "com.mysql.cj.jdbc.Driver"
url = "jdbc:mysql://localhost:3306/testdb"
user = "root"
password = "123"
poolInitialSize = 5
poolMaxSize = 7
poolConnectionTimeoutMillis = 1000
poolValidationQuery = "select 1 as one"
poolFactoryName = "bonecp"
}
postgres {
driver = "org.postgresql.Driver"
url = "jdbc:postgresql://localhost:5432/testdb"
user = "root"
password = "123"
poolFactoryName = "hikaricp"
numThreads = 10
maxConnections = 12
minConnections = 4
keepAliveConnection = true
}
}
# scallikejdbc Global settings
scalikejdbc.global.loggingSQLAndTime.enabled = true
scalikejdbc.global.loggingSQLAndTime.logLevel = info
scalikejdbc.global.loggingSQLAndTime.warningEnabled = true
scalikejdbc.global.loggingSQLAndTime.warningThresholdMillis = 1000
scalikejdbc.global.loggingSQLAndTime.warningLogLevel = warn
scalikejdbc.global.loggingSQLAndTime.singleLineMode = false
scalikejdbc.global.loggingSQLAndTime.printUnprocessedStackTrace = false
scalikejdbc.global.loggingSQLAndTime.stackTraceDepth = 10
}main/resources/logback.xml
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<layout class="ch.qos.logback.classic.PatternLayout">
<Pattern>
%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n
</Pattern>
</layout>
</appender>
<logger name="sdp.cql" level="info"
additivity="false">
<appender-ref ref="STDOUT" />
</logger>
<logger name="demo.sdp.grpc.cql" level="info"
additivity="false">
<appender-ref ref="STDOUT" />
</logger>
<root level="error">
<appender-ref ref="STDOUT" />
</root>
</configuration>main/protobuf/cql.proto
syntax = "proto3";
import "google/protobuf/wrappers.proto";
import "google/protobuf/any.proto";
import "scalapb/scalapb.proto";
option (scalapb.options) = {
// use a custom Scala package name
// package_name: "io.ontherocks.introgrpc.demo"
// don't append file name to package
flat_package: true
// generate one Scala file for all messages (services still get their own file)
single_file: true
// add imports to generated file
// useful when extending traits or using custom types
// import: "io.ontherocks.hellogrpc.RockingMessage"
// code to put at the top of generated file
// works only with `single_file: true`
//preamble: "sealed trait SomeSealedTrait"
};
/*
* Demoes various customization options provided by ScalaPBs.
*/
package sdp.grpc.services;
message ProtoDate {
int32 yyyy = 1;
int32 mm = 2;
int32 dd = 3;
}
message ProtoTime {
int32 hh = 1;
int32 mm = 2;
int32 ss = 3;
int32 nnn = 4;
}
message ProtoDateTime {
ProtoDate date = 1;
ProtoTime time = 2;
}
message AQMRPTRow {
int64 rowid = 1;
string countyname = 2;
string statename = 3;
int64 measureid = 4;
int32 reportyear = 5;
int32 value = 6;
ProtoDateTime created = 7;
}
message CQLResult {
bytes result = 1;
}
message CQLQuery {
string statement = 1;
bytes parameters = 2;
google.protobuf.Int32Value consistency = 3;
google.protobuf.Int32Value fetchSize = 4;
}
message CQLUpdate {
repeated string statements = 1;
bytes parameters = 2;
google.protobuf.Int32Value consistency = 3;
google.protobuf.BoolValue batch = 4;
}
message HelloMsg {
string hello = 1;
}
service CQLServices {
rpc clientStreaming(stream HelloMsg) returns (stream HelloMsg) {}
rpc transferRows(stream AQMRPTRow) returns (stream CQLResult) {}
rpc runQuery(CQLQuery) returns (stream AQMRPTRow) {}
rpc runDDL(CQLUpdate) returns (CQLResult) {}
}
logging/log.scala
package sdp.logging
import org.slf4j.Logger
/**
* Logger which just wraps org.slf4j.Logger internally.
*
* @param logger logger
*/
class Log(logger: Logger) {
// use var consciously to enable squeezing later
var isDebugEnabled: Boolean = logger.isDebugEnabled
var isInfoEnabled: Boolean = logger.isInfoEnabled
var isWarnEnabled: Boolean = logger.isWarnEnabled
var isErrorEnabled: Boolean = logger.isErrorEnabled
def withLevel(level: Symbol)(msg: => String, e: Throwable = null): Unit = {
level match {
case 'debug | 'DEBUG => debug(msg)
case 'info | 'INFO => info(msg)
case 'warn | 'WARN => warn(msg)
case 'error | 'ERROR => error(msg)
case _ => // nothing to do
}
}
def debug(msg: => String): Unit = {
if (isDebugEnabled && logger.isDebugEnabled) {
logger.debug(msg)
}
}
def debug(msg: => String, e: Throwable): Unit = {
if (isDebugEnabled && logger.isDebugEnabled) {
logger.debug(msg, e)
}
}
def info(msg: => String): Unit = {
if (isInfoEnabled && logger.isInfoEnabled) {
logger.info(msg)
}
}
def info(msg: => String, e: Throwable): Unit = {
if (isInfoEnabled && logger.isInfoEnabled) {
logger.info(msg, e)
}
}
def warn(msg: => String): Unit = {
if (isWarnEnabled && logger.isWarnEnabled) {
logger.warn(msg)
}
}
def warn(msg: => String, e: Throwable): Unit = {
if (isWarnEnabled && logger.isWarnEnabled) {
logger.warn(msg, e)
}
}
def error(msg: => String): Unit = {
if (isErrorEnabled && logger.isErrorEnabled) {
logger.error(msg)
}
}
def error(msg: => String, e: Throwable): Unit = {
if (isErrorEnabled && logger.isErrorEnabled) {
logger.error(msg, e)
}
}
}
logging/LogSupport.scala
package sdp.logging
import org.slf4j.LoggerFactory
trait LogSupport {
/**
* Logger
*/
protected val log = new Log(LoggerFactory.getLogger(this.getClass))
}filestreaming/FileStreaming.scala
package sdp.file
import java.io.{ByteArrayInputStream, InputStream}
import java.nio.ByteBuffer
import java.nio.file.Paths
import akka.stream.Materializer
import akka.stream.scaladsl.{FileIO, StreamConverters}
import akka.util._
import scala.concurrent.Await
import scala.concurrent.duration._
object Streaming {
def FileToByteBuffer(fileName: String, timeOut: FiniteDuration = 60 seconds)(
implicit mat: Materializer):ByteBuffer = {
val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
hd ++ bs
}
(Await.result(fut, timeOut)).toByteBuffer
}
def FileToByteArray(fileName: String, timeOut: FiniteDuration = 60 seconds)(
implicit mat: Materializer): Array[Byte] = {
val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
hd ++ bs
}
(Await.result(fut, timeOut)).toArray
}
def FileToInputStream(fileName: String, timeOut: FiniteDuration = 60 seconds)(
implicit mat: Materializer): InputStream = {
val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
hd ++ bs
}
val buf = (Await.result(fut, timeOut)).toArray
new ByteArrayInputStream(buf)
}
def ByteBufferToFile(byteBuf: ByteBuffer, fileName: String)(
implicit mat: Materializer) = {
val ba = new Array[Byte](byteBuf.remaining())
byteBuf.get(ba,0,ba.length)
val baInput = new ByteArrayInputStream(ba)
val source = StreamConverters.fromInputStream(() => baInput) //ByteBufferInputStream(bytes))
source.runWith(FileIO.toPath(Paths.get(fileName)))
}
def ByteArrayToFile(bytes: Array[Byte], fileName: String)(
implicit mat: Materializer) = {
val bb = ByteBuffer.wrap(bytes)
val baInput = new ByteArrayInputStream(bytes)
val source = StreamConverters.fromInputStream(() => baInput) //ByteBufferInputStream(bytes))
source.runWith(FileIO.toPath(Paths.get(fileName)))
}
def InputStreamToFile(is: InputStream, fileName: String)(
implicit mat: Materializer) = {
val source = StreamConverters.fromInputStream(() => is)
source.runWith(FileIO.toPath(Paths.get(fileName)))
}
}jdbc/JDBCConfig.scala
package sdp.jdbc.config
import scala.collection.mutable
import scala.concurrent.duration.Duration
import scala.language.implicitConversions
import com.typesafe.config._
import java.util.concurrent.TimeUnit
import java.util.Properties
import scalikejdbc.config._
import com.typesafe.config.Config
import com.zaxxer.hikari._
import scalikejdbc.ConnectionPoolFactoryRepository
/** Extension methods to make Typesafe Config easier to use */
class ConfigExtensionMethods(val c: Config) extends AnyVal {
import scala.collection.JavaConverters._
def getBooleanOr(path: String, default: => Boolean = false) = if(c.hasPath(path)) c.getBoolean(path) else default
def getIntOr(path: String, default: => Int = 0) = if(c.hasPath(path)) c.getInt(path) else default
def getStringOr(path: String, default: => String = null) = if(c.hasPath(path)) c.getString(path) else default
def getConfigOr(path: String, default: => Config = ConfigFactory.empty()) = if(c.hasPath(path)) c.getConfig(path) else default
def getMillisecondsOr(path: String, default: => Long = 0L) = if(c.hasPath(path)) c.getDuration(path, TimeUnit.MILLISECONDS) else default
def getDurationOr(path: String, default: => Duration = Duration.Zero) =
if(c.hasPath(path)) Duration(c.getDuration(path, TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS) else default
def getPropertiesOr(path: String, default: => Properties = null): Properties =
if(c.hasPath(path)) new ConfigExtensionMethods(c.getConfig(path)).toProperties else default
def toProperties: Properties = {
def toProps(m: mutable.Map[String, ConfigValue]): Properties = {
val props = new Properties(null)
m.foreach { case (k, cv) =>
val v =
if(cv.valueType() == ConfigValueType.OBJECT) toProps(cv.asInstanceOf[ConfigObject].asScala)
else if(cv.unwrapped eq null) null
else cv.unwrapped.toString
if(v ne null) props.put(k, v)
}
props
}
toProps(c.root.asScala)
}
def getBooleanOpt(path: String): Option[Boolean] = if(c.hasPath(path)) Some(c.getBoolean(path)) else None
def getIntOpt(path: String): Option[Int] = if(c.hasPath(path)) Some(c.getInt(path)) else None
def getStringOpt(path: String) = Option(getStringOr(path))
def getPropertiesOpt(path: String) = Option(getPropertiesOr(path))
}
object ConfigExtensionMethods {
@inline implicit def configExtensionMethods(c: Config): ConfigExtensionMethods = new ConfigExtensionMethods(c)
}
trait HikariConfigReader extends TypesafeConfigReader {
self: TypesafeConfig => // with TypesafeConfigReader => //NoEnvPrefix =>
import ConfigExtensionMethods.configExtensionMethods
def getFactoryName(dbName: Symbol): String = {
val c: Config = config.getConfig(envPrefix + "db." + dbName.name)
c.getStringOr("poolFactoryName", ConnectionPoolFactoryRepository.COMMONS_DBCP)
}
def hikariCPConfig(dbName: Symbol): HikariConfig = {
val hconf = new HikariConfig()
val c: Config = config.getConfig(envPrefix + "db." + dbName.name)
// Connection settings
if (c.hasPath("dataSourceClass")) {
hconf.setDataSourceClassName(c.getString("dataSourceClass"))
} else {
Option(c.getStringOr("driverClassName", c.getStringOr("driver"))).map(hconf.setDriverClassName _)
}
hconf.setJdbcUrl(c.getStringOr("url", null))
c.getStringOpt("user").foreach(hconf.setUsername)
c.getStringOpt("password").foreach(hconf.setPassword)
c.getPropertiesOpt("properties").foreach(hconf.setDataSourceProperties)
// Pool configuration
hconf.setConnectionTimeout(c.getMillisecondsOr("connectionTimeout", 1000))
hconf.setValidationTimeout(c.getMillisecondsOr("validationTimeout", 1000))
hconf.setIdleTimeout(c.getMillisecondsOr("idleTimeout", 600000))
hconf.setMaxLifetime(c.getMillisecondsOr("maxLifetime", 1800000))
hconf.setLeakDetectionThreshold(c.getMillisecondsOr("leakDetectionThreshold", 0))
hconf.setInitializationFailFast(c.getBooleanOr("initializationFailFast", false))
c.getStringOpt("connectionTestQuery").foreach(hconf.setConnectionTestQuery)
c.getStringOpt("connectionInitSql").foreach(hconf.setConnectionInitSql)
val numThreads = c.getIntOr("numThreads", 20)
hconf.setMaximumPoolSize(c.getIntOr("maxConnections", numThreads * 5))
hconf.setMinimumIdle(c.getIntOr("minConnections", numThreads))
hconf.setPoolName(c.getStringOr("poolName", dbName.name))
hconf.setRegisterMbeans(c.getBooleanOr("registerMbeans", false))
// Equivalent of ConnectionPreparer
hconf.setReadOnly(c.getBooleanOr("readOnly", false))
c.getStringOpt("isolation").map("TRANSACTION_" + _).foreach(hconf.setTransactionIsolation)
hconf.setCatalog(c.getStringOr("catalog", null))
hconf
}
}
import scalikejdbc._
trait ConfigDBs {
self: TypesafeConfigReader with TypesafeConfig with HikariConfigReader =>
def setup(dbName: Symbol = ConnectionPool.DEFAULT_NAME): Unit = {
getFactoryName(dbName) match {
case "hikaricp" => {
val hconf = hikariCPConfig(dbName)
val hikariCPSource = new HikariDataSource(hconf)
case class HikariDataSourceCloser(src: HikariDataSource) extends DataSourceCloser {
var closed = false
override def close(): Unit = src.close()
}
if (hconf.getDriverClassName != null && hconf.getDriverClassName.trim.nonEmpty) {
Class.forName(hconf.getDriverClassName)
}
ConnectionPool.add(dbName, new DataSourceConnectionPool(dataSource = hikariCPSource,settings = DataSourceConnectionPoolSettings(),
closer = HikariDataSourceCloser(hikariCPSource)))
}
case _ => {
val JDBCSettings(url, user, password, driver) = readJDBCSettings(dbName)
val cpSettings = readConnectionPoolSettings(dbName)
if (driver != null && driver.trim.nonEmpty) {
Class.forName(driver)
}
ConnectionPool.add(dbName, url, user, password, cpSettings)
}
}
}
def setupAll(): Unit = {
loadGlobalSettings()
dbNames.foreach { dbName => setup(Symbol(dbName)) }
}
def close(dbName: Symbol = ConnectionPool.DEFAULT_NAME): Unit = {
ConnectionPool.close(dbName)
}
def closeAll(): Unit = {
ConnectionPool.closeAll
}
}
object ConfigDBs extends ConfigDBs
with TypesafeConfigReader
with StandardTypesafeConfig
with HikariConfigReader
case class ConfigDBsWithEnv(envValue: String) extends ConfigDBs
with TypesafeConfigReader
with StandardTypesafeConfig
with HikariConfigReader
with EnvPrefix {
override val env = Option(envValue)
}jdbc/JDBCEngine.scala
package sdp.jdbc.engine
import java.sql.PreparedStatement
import scala.collection.generic.CanBuildFrom
import akka.stream.scaladsl._
import scalikejdbc._
import scalikejdbc.streams._
import akka.NotUsed
import akka.stream._
import java.time._
import scala.concurrent.duration._
import scala.concurrent._
import sdp.file.Streaming._
import scalikejdbc.TxBoundary.Try._
import scala.concurrent.ExecutionContextExecutor
import java.io.InputStream
import sdp.logging.LogSupport
object JDBCContext {
type SQLTYPE = Int
val SQL_EXEDDL= 1
val SQL_UPDATE = 2
val RETURN_GENERATED_KEYVALUE = true
val RETURN_UPDATED_COUNT = false
}
case class JDBCQueryContext[M](
dbName: Symbol,
statement: String,
parameters: Seq[Any] = Nil,
fetchSize: Int = 100,
autoCommit: Boolean = false,
queryTimeout: Option[Int] = None)
case class JDBCContext (
dbName: Symbol,
statements: Seq[String] = Nil,
parameters: Seq[Seq[Any]] = Nil,
fetchSize: Int = 100,
queryTimeout: Option[Int] = None,
queryTags: Seq[String] = Nil,
sqlType: JDBCContext.SQLTYPE = JDBCContext.SQL_UPDATE,
batch: Boolean = false,
returnGeneratedKey: Seq[Option[Any]] = Nil,
// no return: None, return by index: Some(1), by name: Some("id")
preAction: Option[PreparedStatement => Unit] = None,
postAction: Option[PreparedStatement => Unit] = None)
extends LogSupport {
ctx =>
//helper functions
def appendTag(tag: String): JDBCContext = ctx.copy(queryTags = ctx.queryTags :+ tag)
def appendTags(tags: Seq[String]): JDBCContext = ctx.copy(queryTags = ctx.queryTags ++ tags)
def setFetchSize(size: Int): JDBCContext = ctx.copy(fetchSize = size)
def setQueryTimeout(time: Option[Int]): JDBCContext = ctx.copy(queryTimeout = time)
def setPreAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
!ctx.batch && ctx.statements.size == 1) {
val nc = ctx.copy(preAction = action)
log.info("setPreAction> set")
nc
}
else {
log.info("setPreAction> JDBCContex setting error: preAction not supported!")
throw new IllegalStateException("JDBCContex setting error: preAction not supported!")
}
}
def setPostAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
!ctx.batch && ctx.statements.size == 1) {
val nc = ctx.copy(postAction = action)
log.info("setPostAction> set")
nc
}
else {
log.info("setPreAction> JDBCContex setting error: postAction not supported!")
throw new IllegalStateException("JDBCContex setting error: postAction not supported!")
}
}
def appendDDLCommand(_statement: String, _parameters: Any*): JDBCContext = {
if (ctx.sqlType == JDBCContext.SQL_EXEDDL) {
log.info(s"appendDDLCommand> appending: statement: ${_statement}, parameters: ${_parameters}")
val nc = ctx.copy(
statements = ctx.statements ++ Seq(_statement),
parameters = ctx.parameters ++ Seq(Seq(_parameters))
)
log.info(s"appendDDLCommand> appended: statement: ${nc.statements}, parameters: ${nc.parameters}")
nc
} else {
log.info(s"appendDDLCommand> JDBCContex setting error: option not supported!")
throw new IllegalStateException("JDBCContex setting error: option not supported!")
}
}
def appendUpdateCommand(_returnGeneratedKey: Boolean, _statement: String,_parameters: Any*): JDBCContext = {
if (ctx.sqlType == JDBCContext.SQL_UPDATE && !ctx.batch) {
log.info(s"appendUpdateCommand> appending: returnGeneratedKey: ${_returnGeneratedKey}, statement: ${_statement}, parameters: ${_parameters}")
val nc = ctx.copy(
statements = ctx.statements ++ Seq(_statement),
parameters = ctx.parameters ++ Seq(_parameters),
returnGeneratedKey = ctx.returnGeneratedKey ++ (if (_returnGeneratedKey) Seq(Some(1)) else Seq(None))
)
log.info(s"appendUpdateCommand> appended: statement: ${nc.statements}, parameters: ${nc.parameters}")
nc
} else {
log.info(s"appendUpdateCommand> JDBCContex setting error: option not supported!")
throw new IllegalStateException("JDBCContex setting error: option not supported!")
}
}
def appendBatchParameters(_parameters: Any*): JDBCContext = {
log.info(s"appendBatchParameters> appending: parameters: ${_parameters}")
if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch) {
log.info(s"appendBatchParameters> JDBCContex setting error: batch parameters only supported for SQL_UPDATE and batch = true!")
throw new IllegalStateException("JDBCContex setting error: batch parameters only supported for SQL_UPDATE and batch = true!")
}
var matchParams = true
if (ctx.parameters != Nil)
if (ctx.parameters.head.size != _parameters.size)
matchParams = false
if (matchParams) {
val nc = ctx.copy(
parameters = ctx.parameters ++ Seq(_parameters)
)
log.info(s"appendBatchParameters> appended: statement: ${nc.statements}, parameters: ${nc.parameters}")
nc
} else {
log.info(s"appendBatchParameters> JDBCContex setting error: batch command parameters not match!")
throw new IllegalStateException("JDBCContex setting error: batch command parameters not match!")
}
}
def setBatchReturnGeneratedKeyOption(returnKey: Boolean): JDBCContext = {
if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch)
throw new IllegalStateException("JDBCContex setting error: only supported in batch update commands!")
ctx.copy(
returnGeneratedKey = if (returnKey) Seq(Some(1)) else Nil
)
}
def setDDLCommand(_statement: String, _parameters: Any*): JDBCContext = {
log.info(s"setDDLCommand> setting: statement: ${_statement}, parameters: ${_parameters}")
val nc = ctx.copy(
statements = Seq(_statement),
parameters = Seq(_parameters),
sqlType = JDBCContext.SQL_EXEDDL,
batch = false
)
log.info(s"setDDLCommand> set: statement: ${nc.statements}, parameters: ${nc.parameters}")
nc
}
def setUpdateCommand(_returnGeneratedKey: Boolean, _statement: String,_parameters: Any*): JDBCContext = {
log.info(s"setUpdateCommand> setting: returnGeneratedKey: ${_returnGeneratedKey}, statement: ${_statement}, parameters: ${_parameters}")
val nc = ctx.copy(
statements = Seq(_statement),
parameters = Seq(_parameters),
returnGeneratedKey = if (_returnGeneratedKey) Seq(Some(1)) else Seq(None),
sqlType = JDBCContext.SQL_UPDATE,
batch = false
)
log.info(s"setUpdateCommand> set: statement: ${nc.statements}, parameters: ${nc.parameters}")
nc
}
def setBatchCommand(_statement: String): JDBCContext = {
log.info(s"setBatchCommand> appending: statement: ${_statement}")
val nc = ctx.copy (
statements = Seq(_statement),
sqlType = JDBCContext.SQL_UPDATE,
batch = true
)
log.info(s"setBatchCommand> set: statement: ${nc.statements}, parameters: ${nc.parameters}")
nc
}
}
object JDBCEngine extends LogSupport {
import JDBCContext._
type JDBCDate = LocalDate
type JDBCDateTime = LocalDateTime
type JDBCTime = LocalTime
def jdbcSetDate(yyyy: Int, mm: Int, dd: Int) = LocalDate.of(yyyy,mm,dd)
def jdbcSetTime(hh: Int, mm: Int, ss: Int, nn: Int) = LocalTime.of(hh,mm,ss,nn)
def jdbcSetDateTime(date: JDBCDate, time: JDBCTime) = LocalDateTime.of(date,time)
def jdbcSetNow = LocalDateTime.now()
def jdbcGetDate(sqlDate: java.sql.Date): java.time.LocalDate = sqlDate.toLocalDate
def jdbcGetTime(sqlTime: java.sql.Time): java.time.LocalTime = sqlTime.toLocalTime
def jdbcGetTimestamp(sqlTimestamp: java.sql.Timestamp): java.time.LocalDateTime =
sqlTimestamp.toLocalDateTime
type JDBCBlob = InputStream
def fileToJDBCBlob(fileName: String, timeOut: FiniteDuration = 60 seconds)(
implicit mat: Materializer) = FileToInputStream(fileName,timeOut)
def jdbcBlobToFile(blob: JDBCBlob, fileName: String)(
implicit mat: Materializer) = InputStreamToFile(blob,fileName)
private def noExtractor(message: String): WrappedResultSet => Nothing = { (rs: WrappedResultSet) =>
throw new IllegalStateException(message)
}
def jdbcAkkaStream[A](ctx: JDBCQueryContext[A],extractor: WrappedResultSet => A)
(implicit ec: ExecutionContextExecutor): Source[A,NotUsed] = {
val publisher: DatabasePublisher[A] = NamedDB(ctx.dbName) readOnlyStream {
val rawSql = new SQLToCollectionImpl[A, NoExtractor](ctx.statement, ctx.parameters)(noExtractor(""))
ctx.queryTimeout.foreach(rawSql.queryTimeout(_))
val sql: SQL[A, HasExtractor] = rawSql.map(extractor)
sql.iterator
.withDBSessionForceAdjuster(session => {
session.connection.setAutoCommit(ctx.autoCommit)
session.fetchSize(ctx.fetchSize)
})
}
log.info(s"jdbcAkkaStream> Source: db: ${ctx.dbName}, statement: ${ctx.statement}, parameters: ${ctx.parameters}")
Source.fromPublisher[A](publisher)
}
def jdbcQueryResult[C[_] <: TraversableOnce[_], A](ctx: JDBCQueryContext[A],
extractor: WrappedResultSet => A)(
implicit cbf: CanBuildFrom[Nothing, A, C[A]]): C[A] = {
val rawSql = new SQLToCollectionImpl[A, NoExtractor](ctx.statement, ctx.parameters)(noExtractor(""))
ctx.queryTimeout.foreach(rawSql.queryTimeout(_))
rawSql.fetchSize(ctx.fetchSize)
try {
implicit val session = NamedAutoSession(ctx.dbName)
log.info(s"jdbcQueryResult> Source: db: ${ctx.dbName}, statement: ${ctx.statement}, parameters: ${ctx.parameters}")
val sql: SQL[A, HasExtractor] = rawSql.map(extractor)
sql.collection.apply[C]()
} catch {
case e: Exception =>
log.error(s"jdbcQueryResult> runtime error: ${e.getMessage}")
throw new RuntimeException(s"jdbcQueryResult> Error: ${e.getMessage}")
}
}
def jdbcExecuteDDL(ctx: JDBCContext)(implicit ec: ExecutionContextExecutor): Future[String] = {
if (ctx.sqlType != SQL_EXEDDL) {
log.info(s"jdbcExecuteDDL> JDBCContex setting error: sqlType must be 'SQL_EXEDDL'!")
Future.failed(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_EXEDDL'!"))
}
else {
log.info(s"jdbcExecuteDDL> Source: db: ${ctx.dbName}, statement: ${ctx.statements}, parameters: ${ctx.parameters}")
Future {
NamedDB(ctx.dbName) localTx { implicit session =>
ctx.statements.foreach { stm =>
val ddl = new SQLExecution(statement = stm, parameters = Nil)(
before = WrappedResultSet => {})(
after = WrappedResultSet => {})
ddl.apply()
}
"SQL_EXEDDL executed succesfully."
}
}
}
}
def jdbcBatchUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
implicit ec: ExecutionContextExecutor,
cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
if (ctx.statements == Nil) {
log.info(s"jdbcBatchUpdate> JDBCContex setting error: statements empty!")
Future.failed(new IllegalStateException("JDBCContex setting error: statements empty!"))
}
if (ctx.sqlType != SQL_UPDATE) {
log.info(s"jdbcBatchUpdate> JDBCContex setting error: sqlType must be 'SQL_UPDATE'!")
Future.failed(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!"))
}
else {
if (ctx.batch) {
if (noReturnKey(ctx)) {
log.info(s"jdbcBatchUpdate> batch updating no return: db: ${ctx.dbName}, statement: ${ctx.statements}, parameters: ${ctx.parameters}")
val usql = SQL(ctx.statements.head)
.tags(ctx.queryTags: _*)
.batch(ctx.parameters: _*)
Future {
NamedDB(ctx.dbName) localTx { implicit session =>
ctx.queryTimeout.foreach(session.queryTimeout(_))
usql.apply[Seq]()
Seq.empty[Long].to[C]
}
}
} else {
log.info(s"jdbcBatchUpdate> batch updating return genkey: db: ${ctx.dbName}, statement: ${ctx.statements}, parameters: ${ctx.parameters}")
val usql = new SQLBatchWithGeneratedKey(ctx.statements.head, ctx.parameters, ctx.queryTags)(None)
Future {
NamedDB(ctx.dbName) localTx { implicit session =>
ctx.queryTimeout.foreach(session.queryTimeout(_))
usql.apply[C]()
}
}
}
} else {
log.info(s"jdbcBatchUpdate> JDBCContex setting error: must set batch = true !")
Future.failed(new IllegalStateException("JDBCContex setting error: must set batch = true !"))
}
}
}
private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
implicit ec: ExecutionContextExecutor,
cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
val Some(key) :: xs = ctx.returnGeneratedKey
val params: Seq[Any] = ctx.parameters match {
case Nil => Nil
case p@_ => p.head
}
log.info(s"singleTxUpdateWithReturnKey> updating: db: ${ctx.dbName}, statement: ${ctx.statements}, parameters: ${ctx.parameters}")
val usql = new SQLUpdateWithGeneratedKey(ctx.statements.head, params, ctx.queryTags)(key)
Future {
NamedDB(ctx.dbName) localTx { implicit session =>
session.fetchSize(ctx.fetchSize)
ctx.queryTimeout.foreach(session.queryTimeout(_))
val result = usql.apply()
Seq(result).to[C]
}
}
}
private def singleTxUpdateNoReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
implicit ec: ExecutionContextExecutor,
cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
val params: Seq[Any] = ctx.parameters match {
case Nil => Nil
case p@_ => p.head
}
val before = ctx.preAction match {
case None => pstm: PreparedStatement => {}
case Some(f) => f
}
val after = ctx.postAction match {
case None => pstm: PreparedStatement => {}
case Some(f) => f
}
log.info(s"singleTxUpdateNoReturnKey> updating: db: ${ctx.dbName}, statement: ${ctx.statements}, parameters: ${ctx.parameters}")
val usql = new SQLUpdate(ctx.statements.head,params,ctx.queryTags)(before)(after)
Future {
NamedDB(ctx.dbName) localTx {implicit session =>
session.fetchSize(ctx.fetchSize)
ctx.queryTimeout.foreach(session.queryTimeout(_))
val result = usql.apply()
Seq(result.toLong).to[C]
}
}
}
private def singleTxUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
implicit ec: ExecutionContextExecutor,
cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
if (noReturnKey(ctx))
singleTxUpdateNoReturnKey(ctx)
else
singleTxUpdateWithReturnKey(ctx)
}
private def noReturnKey(ctx: JDBCContext): Boolean = {
if (ctx.returnGeneratedKey != Nil) {
val k :: xs = ctx.returnGeneratedKey
k match {
case None => true
case Some(k) => false
}
} else true
}
def noActon: PreparedStatement=>Unit = pstm => {}
def multiTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
implicit ec: ExecutionContextExecutor,
cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
val keys: Seq[Option[Any]] = ctx.returnGeneratedKey match {
case Nil => Seq.fill(ctx.statements.size)(None)
case k@_ => k
}
val sqlcmd = ctx.statements zip ctx.parameters zip keys
log.info(s"multiTxUpdates> updating: db: ${ctx.dbName}, SQL Commands: ${sqlcmd}")
Future {
NamedDB(ctx.dbName) localTx { implicit session =>
session.fetchSize(ctx.fetchSize)
ctx.queryTimeout.foreach(session.queryTimeout(_))
val results = sqlcmd.map { case ((stm, param), key) =>
key match {
case None =>
new SQLUpdate(stm, param, Nil)(noActon)(noActon).apply().toLong
case Some(k) =>
new SQLUpdateWithGeneratedKey(stm, param, Nil)(k).apply().toLong
}
}
results.to[C]
}
}
}
def jdbcTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
implicit ec: ExecutionContextExecutor,
cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
if (ctx.statements == Nil) {
log.info(s"jdbcTxUpdates> JDBCContex setting error: statements empty!")
Future.failed(new IllegalStateException("JDBCContex setting error: statements empty!"))
}
if (ctx.sqlType != SQL_UPDATE) {
log.info(s"jdbcTxUpdates> JDBCContex setting error: sqlType must be 'SQL_UPDATE'!")
Future.failed(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!"))
}
else {
if (!ctx.batch) {
if (ctx.statements.size == 1)
singleTxUpdate(ctx)
else
multiTxUpdates(ctx)
} else {
log.info(s"jdbcTxUpdates> JDBCContex setting error: must set batch = false !")
Future.failed(new IllegalStateException("JDBCContex setting error: must set batch = false !"))
}
}
}
case class JDBCActionStream[R](dbName: Symbol, parallelism: Int = 1, processInOrder: Boolean = true,
statement: String, prepareParams: R => Seq[Any]) extends LogSupport {
jas =>
def setDBName(db: Symbol): JDBCActionStream[R] = jas.copy(dbName = db)
def setParallelism(parLevel: Int): JDBCActionStream[R] = jas.copy(parallelism = parLevel)
def setProcessOrder(ordered: Boolean): JDBCActionStream[R] = jas.copy(processInOrder = ordered)
private def perform(r: R)(implicit ec: ExecutionContextExecutor) = {
import scala.concurrent._
val params = prepareParams(r)
log.info(s"JDBCActionStream.perform> db: ${dbName}, statement: ${statement}, parameters: ${params}")
Future {
NamedDB(dbName) autoCommit { session =>
session.execute(statement, params: _*)
}
r
}
}
def performOnRow(implicit ec: ExecutionContextExecutor): Flow[R, R, NotUsed] =
if (processInOrder)
Flow[R].mapAsync(parallelism)(perform)
else
Flow[R].mapAsyncUnordered(parallelism)(perform)
}
object JDBCActionStream {
def apply[R](_dbName: Symbol, _statement: String, params: R => Seq[Any]): JDBCActionStream[R] =
new JDBCActionStream[R](dbName = _dbName, statement=_statement, prepareParams = params)
}
}cql/CassandraEngine.scala
package sdp.cql.engine
import akka.NotUsed
import akka.stream.alpakka.cassandra.scaladsl._
import akka.stream.scaladsl._
import com.datastax.driver.core._
import com.google.common.util.concurrent.{FutureCallback, Futures, ListenableFuture}
import scala.collection.JavaConverters._
import scala.collection.generic.CanBuildFrom
import scala.concurrent._
import scala.concurrent.duration.Duration
import sdp.logging.LogSupport
object CQLContext {
// Consistency Levels
type CONSISTENCY_LEVEL = Int
val ANY: CONSISTENCY_LEVEL = 0x0000
val ONE: CONSISTENCY_LEVEL = 0x0001
val TWO: CONSISTENCY_LEVEL = 0x0002
val THREE: CONSISTENCY_LEVEL = 0x0003
val QUORUM : CONSISTENCY_LEVEL = 0x0004
val ALL: CONSISTENCY_LEVEL = 0x0005
val LOCAL_QUORUM: CONSISTENCY_LEVEL = 0x0006
val EACH_QUORUM: CONSISTENCY_LEVEL = 0x0007
val LOCAL_ONE: CONSISTENCY_LEVEL = 0x000A
val LOCAL_SERIAL: CONSISTENCY_LEVEL = 0x000B
val SERIAL: CONSISTENCY_LEVEL = 0x000C
def apply(): CQLContext = CQLContext(statements = Nil)
def consistencyLevel: CONSISTENCY_LEVEL => ConsistencyLevel = consistency => {
consistency match {
case ALL => ConsistencyLevel.ALL
case ONE => ConsistencyLevel.ONE
case TWO => ConsistencyLevel.TWO
case THREE => ConsistencyLevel.THREE
case ANY => ConsistencyLevel.ANY
case EACH_QUORUM => ConsistencyLevel.EACH_QUORUM
case LOCAL_ONE => ConsistencyLevel.LOCAL_ONE
case QUORUM => ConsistencyLevel.QUORUM
case SERIAL => ConsistencyLevel.SERIAL
case LOCAL_SERIAL => ConsistencyLevel.LOCAL_SERIAL
}
}
}
case class CQLQueryContext(
statement: String,
parameter: Seq[Object] = Nil,
consistency: Option[CQLContext.CONSISTENCY_LEVEL] = None,
fetchSize: Int = 100
) { ctx =>
def setConsistencyLevel(_consistency: CQLContext.CONSISTENCY_LEVEL): CQLQueryContext =
ctx.copy(consistency = Some(_consistency))
def setFetchSize(pageSize: Int): CQLQueryContext =
ctx.copy(fetchSize = pageSize)
def setParameters(param: Seq[Object]): CQLQueryContext =
ctx.copy(parameter = param)
}
object CQLQueryContext {
def apply[M](stmt: String, param: Seq[Object]): CQLQueryContext = new CQLQueryContext(statement = stmt, parameter = param)
}
case class CQLContext(
statements: Seq[String],
parameters: Seq[Seq[Object]] = Nil,
consistency: Option[CQLContext.CONSISTENCY_LEVEL] = None,
batch: Boolean = false
) extends LogSupport { ctx =>
def setBatch(bat: Boolean) = ctx.copy(batch = bat)
def setConsistencyLevel(_consistency: CQLContext.CONSISTENCY_LEVEL): CQLContext =
ctx.copy(consistency = Some(_consistency))
def setCommand(_statement: String, _parameters: Object*): CQLContext = {
log.info(s"setCommand> setting: statement: ${_statement}, parameters: ${_parameters}")
val nc = ctx.copy(statements = Seq(_statement), parameters = Seq(_parameters))
log.info(s"setCommand> set: statements: ${nc.statements}, parameters: ${nc.parameters}")
nc
}
def appendCommand(_statement: String, _parameters: Object*): CQLContext = {
log.info(s"appendCommand> appending: statement: ${_statement}, parameters: ${_parameters}")
val nc = ctx.copy(statements = ctx.statements :+ _statement,
parameters = ctx.parameters ++ Seq(_parameters))
log.info(s"appendCommand> appended: statements: ${nc.statements}, parameters: ${nc.parameters}")
nc
}
}
object CQLEngine extends LogSupport {
import CQLContext._
import CQLHelpers._
import cats._, cats.data._, cats.implicits._
import scala.concurrent.{Await, Future}
import scala.concurrent.duration._
def fetchResultPage[C[_] <: TraversableOnce[_],A](ctx: CQLQueryContext, pageSize: Int = 100
,extractor: Row => A)(
implicit session: Session, cbf: CanBuildFrom[Nothing, A, C[A]]): (ResultSet, C[A])= {
val prepStmt = session.prepare(ctx.statement)
var boundStmt = prepStmt.bind()
var params: Seq[Object] = Nil
if (ctx.parameter != Nil) {
params = processParameters(ctx.parameter)
boundStmt = prepStmt.bind(params:_*)
}
log.info(s"fetchResultPage> statement: ${prepStmt.getQueryString}, parameters: ${params}")
ctx.consistency.foreach {consistency =>
boundStmt.setConsistencyLevel(consistencyLevel(consistency))}
val resultSet = session.execute(boundStmt.setFetchSize(pageSize))
(resultSet,(resultSet.asScala.view.map(extractor)).to[C])
}
def fetchMorePages[C[_] <: TraversableOnce[_],A](resultSet: ResultSet, timeOut: Duration)(
extractor: Row => A)(implicit cbf: CanBuildFrom[Nothing, A, C[A]]): (ResultSet,Option[C[A]]) =
if (resultSet.isFullyFetched) {
(resultSet, None)
} else {
try {
val result = Await.result(resultSet.fetchMoreResults(), timeOut)
(result, Some((result.asScala.view.map(extractor)).to[C]))
} catch { case e: Throwable => (resultSet, None) }
}
def cqlExecute(ctx: CQLContext)(
implicit session: Session, ec: ExecutionContext): Future[Boolean] = {
var invalidBat = false
if ( ctx.batch ) {
if (ctx.parameters == Nil)
invalidBat = true
else if (ctx.parameters.size < 2)
invalidBat = true;
}
if (!ctx.batch || invalidBat) {
if(invalidBat)
log.warn(s"cqlExecute> batch update must have at least 2 sets of parameters! change to single-command.")
if (ctx.statements.size == 1) {
var param: Seq[Object] = Nil
if (ctx.parameters != Nil) param = ctx.parameters.head
log.info(s"cqlExecute> single-command: statement: ${ctx.statements.head} parameters: ${param}")
cqlSingleUpdate(ctx.consistency, ctx.statements.head, param)
}
else {
var params: Seq[Seq[Object]] = Nil
if (ctx.parameters == Nil)
params = Seq.fill(ctx.statements.length)(Nil)
else {
if (ctx.statements.size > ctx.parameters.size) {
log.warn(s"cqlExecute> fewer parameters than statements! pad with 'Nil'.")
val nils = Seq.fill(ctx.statements.size - ctx.parameters.size)(Nil)
params = ctx.parameters ++ nils
}
else
params = ctx.parameters
}
val commands: Seq[(String,Seq[Object])] = ctx.statements zip params
log.info(s"cqlExecute> multi-commands: ${commands}")
/*
//using sequence to flip List[Future[Boolean]] => Future[List[Boolean]]
//therefore, make sure no command replies on prev command effect
val lstCmds: List[Future[Boolean]] = commands.map { case (stmt,param) =>
cqlSingleUpdate(ctx.consistency, stmt, param)
}.toList
val futList = lstCmds.sequence.map(_ => true) //must map to execute
*/
/*
//using traverse to have some degree of parallelism = max(runtimes)
//therefore, make sure no command replies on prev command effect
val futList = Future.traverse(commands) { case (stmt,param) =>
cqlSingleUpdate(ctx.consistency, stmt, param)
}.map(_ => true)
Await.result(futList, 3 seconds)
Future.successful(true)
*/
// run sync directly
Future {
commands.foreach { case (stm, pars) =>
cqlExecuteSync(ctx.consistency, stm, pars)
}
true
}
}
}
else
cqlBatchUpdate(ctx)
}
def cqlSingleUpdate(cons: Option[CQLContext.CONSISTENCY_LEVEL],stmt: String, params: Seq[Object])(
implicit session: Session, ec: ExecutionContext): Future[Boolean] = {
val prepStmt = session.prepare(stmt)
var boundStmt = prepStmt.bind()
var pars: Seq[Object] = Nil
if (params != Nil) {
pars = processParameters(params)
boundStmt = prepStmt.bind(pars: _*)
}
log.info(s"cqlSingleUpdate> statement: ${prepStmt.getQueryString}, parameters: ${pars}")
cons.foreach { consistency =>
boundStmt.setConsistencyLevel(consistencyLevel(consistency))
}
session.executeAsync(boundStmt).map(_.wasApplied())
}
def cqlExecuteSync(cons: Option[CQLContext.CONSISTENCY_LEVEL],stmt: String, params: Seq[Object])(
implicit session: Session, ec: ExecutionContext): Boolean = {
val prepStmt = session.prepare(stmt)
var boundStmt = prepStmt.bind()
var pars: Seq[Object] = Nil
if (params != Nil) {
pars = processParameters(params)
boundStmt = prepStmt.bind(pars: _*)
}
log.info(s"cqlExecuteSync> statement: ${prepStmt.getQueryString}, parameters: ${pars}")
cons.foreach { consistency =>
boundStmt.setConsistencyLevel(consistencyLevel(consistency))
}
session.execute(boundStmt).wasApplied()
}
def cqlBatchUpdate(ctx: CQLContext)(
implicit session: Session, ec: ExecutionContext): Future[Boolean] = {
var params: Seq[Seq[Object]] = Nil
if (ctx.parameters == Nil)
params = Seq.fill(ctx.statements.length)(Nil)
else
params = ctx.parameters
log.info(s"cqlBatchUpdate> statement: ${ctx.statements.head}, parameters: ${params}")
val prepStmt = session.prepare(ctx.statements.head)
var batch = new BatchStatement()
params.foreach { p =>
log.info(s"cqlBatchUpdate> batch with raw parameter: ${p}")
val pars = processParameters(p)
log.info(s"cqlMultiUpdate> batch with cooked parameters: ${pars}")
batch.add(prepStmt.bind(pars: _*))
}
ctx.consistency.foreach { consistency =>
batch.setConsistencyLevel(consistencyLevel(consistency))
}
session.executeAsync(batch).map(_.wasApplied())
}
def cassandraStream[A](ctx: CQLQueryContext,extractor: Row => A)
(implicit session: Session, ec: ExecutionContextExecutor): Source[A,NotUsed] = {
val prepStmt = session.prepare(ctx.statement)
var boundStmt = prepStmt.bind()
val params = processParameters(ctx.parameter)
boundStmt = prepStmt.bind(params:_*)
ctx.consistency.foreach {consistency =>
boundStmt.setConsistencyLevel(consistencyLevel(consistency))}
log.info(s"cassandraStream> statement: ${prepStmt.getQueryString}, parameters: ${params}")
CassandraSource(boundStmt.setFetchSize(ctx.fetchSize)).map(extractor)
}
case class CassandraActionStream[R](parallelism: Int = 1, processInOrder: Boolean = true,
statement: String, prepareParams: R => Seq[Object],
consistency: Option[CQLContext.CONSISTENCY_LEVEL] = None){ cas =>
def setParallelism(parLevel: Int): CassandraActionStream[R] = cas.copy(parallelism=parLevel)
def setProcessOrder(ordered: Boolean): CassandraActionStream[R] = cas.copy(processInOrder = ordered)
def setConsistencyLevel(_consistency: CQLContext.CONSISTENCY_LEVEL): CassandraActionStream[R] =
cas.copy(consistency = Some(_consistency))
def perform(r: R)(implicit session: Session, ec: ExecutionContext) = {
var prepStmt = session.prepare(statement)
var boundStmt = prepStmt.bind()
val params = processParameters(prepareParams(r))
boundStmt = prepStmt.bind(params: _*)
consistency.foreach { cons =>
boundStmt.setConsistencyLevel(CQLContext.consistencyLevel(cons))
}
log.info(s"CassandraActionStream.perform> statement: ${prepStmt.getQueryString}, parameters: ${params}")
session.executeAsync(boundStmt).map(_ => r)
}
def performOnRow(implicit session: Session, ec: ExecutionContext): Flow[R, R, NotUsed] =
if (processInOrder)
Flow[R].mapAsync(parallelism)(perform)
else
Flow[R].mapAsyncUnordered(parallelism)(perform)
def unloggedBatch[K](statementBinder: (
R, PreparedStatement) => BoundStatement,partitionKey: R => K)(
implicit session: Session, ec: ExecutionContext): Flow[R, R, NotUsed] = {
val preparedStatement = session.prepare(statement)
log.info(s"CassandraActionStream.unloggedBatch> statement: ${preparedStatement.getQueryString}")
CassandraFlow.createUnloggedBatchWithPassThrough[R, K](
parallelism,
preparedStatement,
statementBinder,
partitionKey)
}
}
object CassandraActionStream {
def apply[R](_statement: String, params: R => Seq[Object]): CassandraActionStream[R] =
new CassandraActionStream[R]( statement=_statement, prepareParams = params)
}
}
object CQLHelpers extends LogSupport {
import java.nio.ByteBuffer
import java.io._
import java.nio.file._
import com.datastax.driver.core.LocalDate
import com.datastax.driver.extras.codecs.jdk8.InstantCodec
import java.time.Instant
import akka.stream.scaladsl._
import akka.stream._
implicit def listenableFutureToFuture[T](
listenableFuture: ListenableFuture[T]): Future[T] = {
val promise = Promise[T]()
Futures.addCallback(listenableFuture, new FutureCallback[T] {
def onFailure(error: Throwable): Unit = {
promise.failure(error)
()
}
def onSuccess(result: T): Unit = {
promise.success(result)
()
}
})
promise.future
}
case class CQLDate(year: Int, month: Int, day: Int)
case object CQLTodayDate
case class CQLDateTime(year: Int, Month: Int,
day: Int, hour: Int, minute: Int, second: Int, millisec: Int = 0)
case object CQLDateTimeNow
def cqlGetDate(dateToConvert: java.util.Date): java.time.LocalDate =
dateToConvert.toInstant()
.atZone(java.time.ZoneId.systemDefault())
.toLocalDate()
def cqlGetTime(dateToConvert: java.util.Date): java.time.LocalTime =
dateToConvert.toInstant()
.atZone(java.time.ZoneId.systemDefault())
.toLocalTime()
def cqlGetTimestamp(dateToConvert: java.util.Date): java.time.LocalDateTime=
new java.sql.Timestamp(
dateToConvert.getTime()
).toLocalDateTime()
def processParameters(params: Seq[Object]): Seq[Object] = {
import java.time.{Clock,ZoneId}
log.info(s"[processParameters] input: ${params}")
val outParams = params.map { obj =>
obj match {
case CQLDate(yy, mm, dd) => LocalDate.fromYearMonthDay(yy, mm, dd)
case CQLTodayDate =>
val today = java.time.LocalDate.now()
LocalDate.fromYearMonthDay(today.getYear, today.getMonth.getValue, today.getDayOfMonth)
case CQLDateTimeNow => Instant.now(Clock.system(ZoneId.of("EST", ZoneId.SHORT_IDS)))
case CQLDateTime(yy, mm, dd, hr, ms, sc, mi) =>
Instant.parse(f"$yy%4d-$mm%2d-$dd%2dT$hr%2d:$ms%2d:$sc%2d$mi%3d")
case p@_ => p
}
}
log.info(s"[processParameters] output: ${params}")
outParams
}
class ByteBufferInputStream(buf: ByteBuffer) extends InputStream {
override def read: Int = {
if (!buf.hasRemaining) return -1
buf.get
}
override def read(bytes: Array[Byte], off: Int, len: Int): Int = {
val length: Int = Math.min(len, buf.remaining)
buf.get(bytes, off, length)
length
}
}
object ByteBufferInputStream {
def apply(buf: ByteBuffer): ByteBufferInputStream = {
new ByteBufferInputStream(buf)
}
}
class FixsizedByteBufferOutputStream(buf: ByteBuffer) extends OutputStream {
override def write(b: Int): Unit = {
buf.put(b.toByte)
}
override def write(bytes: Array[Byte], off: Int, len: Int): Unit = {
buf.put(bytes, off, len)
}
}
object FixsizedByteBufferOutputStream {
def apply(buf: ByteBuffer) = new FixsizedByteBufferOutputStream(buf)
}
class ExpandingByteBufferOutputStream(var buf: ByteBuffer, onHeap: Boolean) extends OutputStream {
private val increasing = ExpandingByteBufferOutputStream.DEFAULT_INCREASING_FACTOR
override def write(b: Array[Byte], off: Int, len: Int): Unit = {
val position = buf.position
val limit = buf.limit
val newTotal: Long = position + len
if(newTotal > limit){
var capacity = (buf.capacity * increasing)
while(capacity <= newTotal){
capacity = (capacity*increasing)
}
increase(capacity.toInt)
}
buf.put(b, 0, len)
}
override def write(b: Int): Unit= {
if (!buf.hasRemaining) increase((buf.capacity * increasing).toInt)
buf.put(b.toByte)
}
protected def increase(newCapacity: Int): Unit = {
buf.limit(buf.position)
buf.rewind
val newBuffer =
if (onHeap) ByteBuffer.allocate(newCapacity)
else ByteBuffer.allocateDirect(newCapacity)
newBuffer.put(buf)
buf.clear
buf = newBuffer
}
def size: Long = buf.position
def capacity: Long = buf.capacity
def byteBuffer: ByteBuffer = buf
}
object ExpandingByteBufferOutputStream {
val DEFAULT_INCREASING_FACTOR = 1.5f
def apply(size: Int, increasingBy: Float, onHeap: Boolean) = {
if (increasingBy <= 1) throw new IllegalArgumentException("Increasing Factor must be greater than 1.0")
val buffer: ByteBuffer =
if (onHeap) ByteBuffer.allocate(size)
else ByteBuffer.allocateDirect(size)
new ExpandingByteBufferOutputStream(buffer,onHeap)
}
def apply(size: Int): ExpandingByteBufferOutputStream = {
apply(size, ExpandingByteBufferOutputStream.DEFAULT_INCREASING_FACTOR, false)
}
def apply(size: Int, onHeap: Boolean): ExpandingByteBufferOutputStream = {
apply(size, ExpandingByteBufferOutputStream.DEFAULT_INCREASING_FACTOR, onHeap)
}
def apply(size: Int, increasingBy: Float): ExpandingByteBufferOutputStream = {
apply(size, increasingBy, false)
}
}
def cqlFileToBytes(fileName: String): ByteBuffer = {
val fis = new FileInputStream(fileName)
val b = new Array[Byte](fis.available + 1)
val length = b.length
fis.read(b)
ByteBuffer.wrap(b)
}
def cqlBytesToFile(bytes: ByteBuffer, fileName: String)(
implicit mat: Materializer): Future[IOResult] = {
val source = StreamConverters.fromInputStream(() => ByteBufferInputStream(bytes))
source.runWith(FileIO.toPath(Paths.get(fileName)))
}
def cqlDateTimeString(date: java.util.Date, fmt: String): String = {
val outputFormat = new java.text.SimpleDateFormat(fmt)
outputFormat.format(date)
}
def useJava8DateTime(cluster: Cluster) = {
//for jdk8 datetime format
cluster.getConfiguration().getCodecRegistry()
.register(InstantCodec.instance)
}
}
BytesConverter.scala
package protobuf.bytes
import java.io.{ByteArrayInputStream,ByteArrayOutputStream,ObjectInputStream,ObjectOutputStream}
import com.google.protobuf.ByteString
object Converter {
def marshal(value: Any): ByteString = {
val stream: ByteArrayOutputStream = new ByteArrayOutputStream()
val oos = new ObjectOutputStream(stream)
oos.writeObject(value)
oos.close()
ByteString.copyFrom(stream.toByteArray())
}
def unmarshal[A](bytes: ByteString): A = {
val ois = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray))
val value = ois.readObject()
ois.close()
value.asInstanceOf[A]
}
}CQLServices.scala
package demo.sdp.grpc.cql.server
import akka.NotUsed
import akka.stream.scaladsl._
import protobuf.bytes.Converter._
import com.datastax.driver.core._
import scala.concurrent.ExecutionContextExecutor
import sdp.grpc.services._
import sdp.cql.engine._
import CQLEngine._
import CQLHelpers._
import sdp.logging.LogSupport
import scala.concurrent._
import scala.concurrent.duration._
import akka.stream.ActorMaterializer
class CQLStreamingServices(implicit ec: ExecutionContextExecutor,
mat: ActorMaterializer, session: Session)
extends CqlGrpcAkkaStream.CQLServices with LogSupport{
val toParams: AQMRPTRow => Seq[Object] = row => Seq[Object](
row.rowid.asInstanceOf[Object],
row.measureid.asInstanceOf[Object],
row.statename,
row.countyname,
row.reportyear.asInstanceOf[Object],
row.value.asInstanceOf[Object],
CQLDateTimeNow
)
val cqlInsert ="""
|insert into testdb.AQMRPT(
| rowid,
| measureid,
| statename,
| countyname,
| reportyear,
| value,
| created)
| values(?,?,?,?,?,?,?)
""".stripMargin
val cqlActionStream = CassandraActionStream(cqlInsert,toParams).setParallelism(2)
.setProcessOrder(false)
/*
val cqlActionFlow: Flow[AQMRPTRow,AQMRPTRow,NotUsed] =
Flow[AQMRPTRow]
.via(cqlActionStream.performOnRow)
*/
val cqlActionFlow: Flow[AQMRPTRow,CQLResult,NotUsed] = {
Flow[AQMRPTRow]
.mapAsync(cqlActionStream.parallelism){ row =>
if (IfExists(row.rowid))
Future.successful(CQLResult(marshal(0)))
else
cqlActionStream.perform(row).map {_ => CQLResult(marshal(1))}
}
}
override def transferRows: Flow[AQMRPTRow, CQLResult, NotUsed] = {
Flow[AQMRPTRow]
.via(cqlActionFlow)
}
private def IfExists(rowid: Long): Boolean = {
val cql = "SELECT * FROM testdb.AQMRPT WHERE ROWID = ? ALLOW FILTERING"
val param = Seq(rowid.asInstanceOf[Object])
val toRowId: Row => Long = r => r.getLong("rowid")
val ctx = CQLQueryContext(cql,param)
val src: Source[Long,NotUsed] = cassandraStream(ctx,toRowId)
val fut = src.toMat(Sink.headOption)(Keep.right).run()
val result = Await.result(fut,3 seconds)
log.info(s"checking existence: ${result}")
result match {
case Some(x) => true
case None => false
}
}
override def clientStreaming: Flow[HelloMsg, HelloMsg, NotUsed] = {
Flow[HelloMsg]
.map {r => println(r) ; r}
}
override def runDDL: Flow[CQLUpdate, CQLResult, NotUsed] = {
Flow[CQLUpdate]
.flatMapConcat { context =>
//unpack CQLUpdate and construct the context
val ctx = CQLContext(context.statements)
log.info(s"**** CQLContext => ${ctx} ***")
Source
.fromFuture(cqlExecute(ctx))
.map { r => CQLResult(marshal(r)) }
}
}
def toCQLTimestamp(rs: Row) = {
try {
val tm = rs.getTimestamp("CREATED")
if (tm == null) None
else {
val localdt = cqlGetTimestamp(tm)
Some(ProtoDateTime(Some(ProtoDate(localdt.getYear, localdt.getMonthValue, localdt.getDayOfMonth)),
Some(ProtoTime(localdt.getHour, localdt.getMinute, localdt.getSecond, localdt.getNano))))
}
}
catch {
case e: Exception => None
}
}
val toAQMRow: Row => AQMRPTRow = rs=> AQMRPTRow(
rowid = rs.getLong("ROWID"),
measureid = rs.getLong("MEASUREID"),
statename = rs.getString("STATENAME"),
countyname = rs.getString("COUNTYNAME"),
reportyear = rs.getInt("REPORTYEAR"),
value = rs.getInt("VALUE"),
created = toCQLTimestamp(rs)
)
override def runQuery: Flow[CQLQuery, AQMRPTRow, NotUsed] = {
log.info("**** runQuery called on service side ***")
Flow[CQLQuery]
.flatMapConcat { q =>
//unpack JDBCQuery and construct the context
var params: Seq[Object] = Nil
if (q.parameters != _root_.com.google.protobuf.ByteString.EMPTY)
params = unmarshal[Seq[Object]](q.parameters)
log.info(s"**** query parameters: ${params} ****")
val ctx = CQLQueryContext(q.statement,params)
CQLEngine.cassandraStream(ctx,toAQMRow)
}
}
}
CQLServer.scala
package demo.sdp.grpc.cql.server
import java.util.logging.Logger
import com.datastax.driver.core._
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import io.grpc.Server
import io.grpc.ServerBuilder
import sdp.grpc.services._
import sdp.cql.engine._
import CQLHelpers._
class gRPCServer(server: Server) {
val logger: Logger = Logger.getLogger(classOf[gRPCServer].getName)
def start(): Unit = {
server.start()
logger.info(s"Server started, listening on ${server.getPort}")
sys.addShutdownHook {
// Use stderr here since the logger may has been reset by its JVM shutdown hook.
System.err.println("*** shutting down gRPC server since JVM is shutting down")
stop()
System.err.println("*** server shut down")
}
()
}
def stop(): Unit = {
server.shutdown()
}
/**
* Await termination on the main thread since the grpc library uses daemon threads.
*/
def blockUntilShutdown(): Unit = {
server.awaitTermination()
}
}
object CQLServer extends App {
implicit val cqlsys = ActorSystem("cqlSystem")
implicit val mat = ActorMaterializer()
implicit val ec = cqlsys.dispatcher
val cluster = new Cluster
.Builder()
.addContactPoints("localhost")
.withPort(9042)
.build()
useJava8DateTime(cluster)
implicit val session = cluster.connect()
val server = new gRPCServer(
ServerBuilder
.forPort(50051)
.addService(
CqlGrpcAkkaStream.bindService(
new CQLStreamingServices
)
).build()
)
server.start()
// server.blockUntilShutdown()
scala.io.StdIn.readLine()
session.close()
cluster.close()
mat.shutdown()
cqlsys.terminate()
}CQLClient.scala
package demo.sdp.grpc.cql.client
import sdp.grpc.services._
import protobuf.bytes.Converter._
import akka.stream.scaladsl._
import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, ThrottleMode}
import io.grpc._
import sdp.logging.LogSupport
import sdp.jdbc.engine._
import JDBCEngine._
import scalikejdbc.WrappedResultSet
import sdp.cql.engine.CQLHelpers.CQLDateTimeNow
import scala.util._
import scala.concurrent.ExecutionContextExecutor
class CQLStreamClient(host: String, port: Int)(
implicit ec: ExecutionContextExecutor) extends LogSupport {
val channel = ManagedChannelBuilder
.forAddress(host, port)
.usePlaintext(true)
.build()
val stub = CqlGrpcAkkaStream.stub(channel)
val jdbcRows2transfer = JDBCQueryContext[AQMRPTRow](
dbName = 'h2,
statement = "select * from AQMRPT where statename='Arkansas'"
)
def toAQMRPTRow: WrappedResultSet => AQMRPTRow = rs => AQMRPTRow(
rowid = rs.long("ROWID"),
measureid = rs.long("MEASUREID"),
statename = rs.string("STATENAME"),
countyname = rs.string("COUNTYNAME"),
reportyear = rs.int("REPORTYEAR"),
value = rs.int("VALUE"),
created = Some(ProtoDateTime(Some(ProtoDate(1990, 8, 12)), Some(ProtoTime(23, 56, 23, 0))))
)
import scala.concurrent.duration._
def transferRows: Source[CQLResult, NotUsed] = {
log.info(s"**** calling transferRows ****")
jdbcAkkaStream(jdbcRows2transfer, toAQMRPTRow)
// .throttle(1, 500.millis, 1, ThrottleMode.shaping)
.via(stub.transferRows)
}
def echoHello: Source[HelloMsg,NotUsed] = {
val row = HelloMsg("hello world!")
val rows = List.fill[HelloMsg](100)(row)
Source
.fromIterator(() => rows.iterator)
.via(stub.clientStreaming)
}
val query0 = CQLQuery(
statement = "select * from testdb.AQMRPT"
)
val query = CQLQuery(
statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE > ? ALLOW FILTERING;",
parameters = marshal(Seq("Arkansas", 0.toInt))
)
val query2 = CQLQuery (
statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE = ?",
parameters = marshal(Seq("Colorado", 3.toInt))
)
val query3= CQLQuery (
statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE = ?",
parameters = marshal(Seq("Arkansas", 8.toInt))
)
def queryRows: Source[AQMRPTRow,NotUsed] = {
log.info(s"running queryRows ...")
Source
.single(query)
.via(stub.runQuery)
}
val dropCQL = "DROP TABLE IF EXISTS testdb.AQMRPT"
val createCQL ="""
CREATE TABLE testdb.AQMRPT (
rowid bigint primary key,
measureid bigint,
statename text,
countyname text,
reportyear int,
value int,
created timestamp
)"""
val cqlddl = CQLUpdate(statements = Seq(dropCQL,createCQL))
def createTbl: Source[CQLResult,NotUsed] = {
log.info(s"running createTbl ...")
Source
.single(cqlddl)
.via(stub.runDDL)
}
}
object EchoHelloClient extends App {
implicit val system = ActorSystem("EchoNumsClient")
implicit val mat = ActorMaterializer.create(system)
implicit val ec = system.dispatcher
val client = new CQLStreamClient("localhost", 50051)
client.echoHello.runForeach(println)
scala.io.StdIn.readLine()
mat.shutdown()
system.terminate()
}
object TransferRows extends App {
import sdp.jdbc.config._
implicit val system = ActorSystem("JDBCServer")
implicit val mat = ActorMaterializer.create(system)
implicit val ec = system.dispatcher
ConfigDBsWithEnv("dev").setup('h2)
ConfigDBsWithEnv("dev").loadGlobalSettings()
val client = new CQLStreamClient("localhost", 50051)
val fut = client.transferRows.runFold(0){(a,b) => a + unmarshal[Int](b.result)}
fut.onComplete {
case scala.util.Success(cnt) => println(s"done transfer ${cnt} rows.")
case Failure(e) => println(s"!!!!!streaming error: ${e.getMessage}")
}
scala.io.StdIn.readLine()
ConfigDBsWithEnv("dev").close('h2)
mat.shutdown()
system.terminate()
}
object QueryRows extends App {
implicit val system = ActorSystem("QueryRows")
implicit val mat = ActorMaterializer.create(system)
implicit val ec = system.dispatcher
val client = new CQLStreamClient("localhost", 50051)
val fut = client.queryRows.runForeach { r => println(r) }
fut.onComplete {
case scala.util.Success(d) => println(s"done querying.")
case Failure(e) => println(s"!!!!!query error: ${e.getMessage}")
}
scala.io.StdIn.readLine()
mat.shutdown()
system.terminate()
}
object RunDDL extends App {
implicit val system = ActorSystem("RunDDL")
implicit val mat = ActorMaterializer.create(system)
implicit val ec = system.dispatcher
val client = new CQLStreamClient("localhost", 50051)
client.createTbl.runForeach { r => println(unmarshal(r.result)) }
scala.io.StdIn.readLine()
mat.shutdown()
system.terminate()
}
本站文章为和通数据库网友分享或者投稿，欢迎任何形式的转载，但请务必注明出处.
同时文章内容如有侵犯了您的权益，请联系QQ：970679559，我们会在尽快处理。