经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Scala » 查看文章
PICE（3）：CassandraStreaming - gRPC-CQL Service
来源：cnblogs　　作者：雪川大虫　　时间：2018/9/25 20:28:19　　对本文有异议

  在上一篇博文里我们介绍了通过gRPC实现JDBC数据库的streaming，这篇我们介绍关于cassandra的streaming实现方式。如果我们需要从一个未部署cassandra的节点或终端上读取cassandra数据，可以用gRPC来搭建一个数据桥梁来连接这两端。这时cassandra这端就是gRPC-Server端，由它提供cassandra的数据服务。

在前面sdp系列讨论里我们已经实现了Cassandra-Engine。它的运作原理还是通过某种Context把指令提交给cassandra去执行。我们先设计一个创建库表的例子。CQL语句和Cassandra-Engine程序代码如下，这是客户端部分：

  1.   val dropCQL = "DROP TABLE IF EXISTS testdb.AQMRPT"
  3.   val createCQL ="""
  4.   CREATE TABLE testdb.AQMRPT (
  5.      rowid bigint primary key,
  6.      measureid bigint,
  7.      statename text,
  8.      countyname text,
  9.      reportyear int,
  10.      value int,
  11.      created timestamp
  12.   )"""
  14.   val cqlddl = CQLUpdate(statements = Seq(dropCQL,createCQL))
  15.   def createTbl: Source[CQLResult,NotUsed] = {
  16.     log.info(s"running createTbl ...")
  17.     Source
  18.       .single(cqlddl)
  19.       .via(stub.runDDL)
  20.   }

首先，我们在CQLUpdate这个protobuf对应Context里传入两条指令dropCQL和createCQL，可以预计这会是一种批次型batch方式。然后一如既往，我们使用了streaming编程模式。在.proto文件里用DDL来对应Context和Service：

  1. message CQLUpdate {
  2.     repeated string statements = 1;
  3.     bytes parameters = 2;
  4.     google.protobuf.Int32Value consistency = 3;
  5.     google.protobuf.BoolValue batch = 4;
  6. }
  8. service CQLServices {
  9.   rpc runDDL(CQLUpdate) returns (CQLResult) {}
  10. }

服务函数runDDL程序实现如下：

  1.  override def runDDL: Flow[CQLUpdate, CQLResult, NotUsed] = {
  2.     Flow[CQLUpdate]
  3.       .flatMapConcat { context =>
  4.         //unpack CQLUpdate and construct the context
  5.         val ctx = CQLContext(context.statements)
  6.         log.info(s"**** CQLContext => ${ctx} ***")
  8.         Source
  9.           .fromFuture(cqlExecute(ctx))
  10.           .map { r => CQLResult(marshal(r)) }
  11.       }
  12.   }

这里我们调用了Cassandra-Engine的cqlExecute(ctx)函数：

  1.   def cqlExecute(ctx: CQLContext)(
  2.     implicit session: Session, ec: ExecutionContext): Future[Boolean] = {
  4.     var invalidBat = false
  5.     if ( ctx.batch ) {
  6.       if (ctx.parameters == Nil)
  7.         invalidBat = true
  8.       else if (ctx.parameters.size < 2)
  9.         invalidBat = true;
  10.     }
  11.     if (!ctx.batch || invalidBat) {
  12.       if(invalidBat)
  13.        log.warn(s"cqlExecute> batch update must have at least 2 sets of parameters! change to single-command.")
  15.       if (ctx.statements.size == 1) {
  16.         var param: Seq[Object] = Nil
  17.         if (ctx.parameters != Nil) param =  ctx.parameters.head
  18.         log.info(s"cqlExecute>  single-command: statement: ${ctx.statements.head} parameters: ${param}")
  19.         cqlSingleUpdate(ctx.consistency, ctx.statements.head, param)
  20.       }
  21.       else {
  22.         var params: Seq[Seq[Object]] = Nil
  23.         if (ctx.parameters == Nil)
  24.           params = Seq.fill(ctx.statements.length)(Nil)
  25.         else {
  26.           if (ctx.statements.size > ctx.parameters.size) {
  27.             log.warn(s"cqlExecute> fewer parameters than statements! pad with 'Nil'.")
  28.             val nils = Seq.fill(ctx.statements.size - ctx.parameters.size)(Nil)
  29.             params = ctx.parameters ++ nils
  31.           }
  32.           else
  33.             params = ctx.parameters
  34.         }
  36.         val commands: Seq[(String,Seq[Object])] = ctx.statements zip params
  37.         log.info(s"cqlExecute>  multi-commands: ${commands}")
  38. /*
  39.         //using sequence to flip List[Future[Boolean]] => Future[List[Boolean]]
  40.         //therefore, make sure no command replies on prev command effect
  41.         val lstCmds: List[Future[Boolean]] = commands.map { case (stmt,param) =>
  42.           cqlSingleUpdate(ctx.consistency, stmt, param)
  43.         }.toList
  45.         val futList = lstCmds.sequence.map(_ => true)   //must map to execute
  46.         */
  47. /*
  48.         //using traverse to have some degree of parallelism = max(runtimes)
  49.         //therefore, make sure no command replies on prev command effect
  50.         val futList = Future.traverse(commands) { case (stmt,param)  =>
  51.           cqlSingleUpdate(ctx.consistency, stmt, param)
  52.         }.map(_ => true)
  54.         Await.result(futList, 3 seconds)
  55.         Future.successful(true)
  56. */
  57.         // run sync directly
  58.         Future {
  59.           commands.foreach { case (stm, pars) =>
  60.             cqlExecuteSync(ctx.consistency, stm, pars)
  61.           }
  62.           true
  63.         }
  64.       }
  65.     }
  66.     else
  67.       cqlBatchUpdate(ctx)
  68.   }

特别展示了这个函数的代码是因为对于一批次多条指令可能会涉及到non-blocking和并行计算。可参考上面代码标注段落里函数式方法（cats）sequence,traverse如何实现对一串Future的运算。

下一个例子是用流方式把JDBC数据库数据并入cassandra数据库里。.proto DDL内容如下：

  1. message ProtoDate {
  2.   int32 yyyy = 1;
  3.   int32 mm   = 2;
  4.   int32 dd   = 3;
  5. }
  7. message ProtoTime {
  8.   int32 hh   = 1;
  9.   int32 mm   = 2;
  10.   int32 ss   = 3;
  11.   int32 nnn  = 4;
  12. }
  14. message ProtoDateTime {
  15.    ProtoDate date = 1;
  16.    ProtoTime time = 2;
  17. }
  19. message AQMRPTRow {
  20.     int64 rowid = 1;
  21.     string countyname = 2;
  22.     string statename = 3;
  23.     int64 measureid = 4;
  24.     int32 reportyear = 5;
  25.     int32 value = 6;
  26.     ProtoDateTime created = 7;
  27. }
  29. message CQLResult {
  30.   bytes result = 1;
  31. }
  33. message CQLUpdate {
  34.     repeated string statements = 1;
  35.     bytes parameters = 2;
  36.     google.protobuf.Int32Value consistency = 3;
  37.     google.protobuf.BoolValue batch = 4;
  38. }
  41. service CQLServices {
  42.   rpc transferRows(stream AQMRPTRow) returns (stream CQLResult) {}
  43.   rpc runDDL(CQLUpdate) returns (CQLResult) {}
  44. }

下面是服务函数的实现：

  1.  val toParams: AQMRPTRow => Seq[Object] = row => Seq[Object](
  2.     row.rowid.asInstanceOf[Object],
  3.     row.measureid.asInstanceOf[Object],
  4.     row.statename,
  5.     row.countyname,
  6.     row.reportyear.asInstanceOf[Object],
  7.     row.value.asInstanceOf[Object],
  8.     CQLDateTimeNow
  9.   )
  10.   val cqlInsert ="""
  11.                    |insert into testdb.AQMRPT(
  12.                    | rowid,
  13.                    | measureid,
  14.                    | statename,
  15.                    | countyname,
  16.                    | reportyear,
  17.                    | value,
  18.                    | created)
  19.                    | values(?,?,?,?,?,?,?)
  20.                  """.stripMargin
  21.   
  22.   val cqlActionStream = CassandraActionStream(cqlInsert,toParams).setParallelism(2)
  23.     .setProcessOrder(false)
  25. /*
  26.   val cqlActionFlow: Flow[AQMRPTRow,AQMRPTRow,NotUsed] =
  27.     Flow[AQMRPTRow]
  28.       .via(cqlActionStream.performOnRow)
  29. */
  31.   val cqlActionFlow: Flow[AQMRPTRow,CQLResult,NotUsed] = {
  32.     Flow[AQMRPTRow]
  33.         .mapAsync(cqlActionStream.parallelism){ row =>
  34.           if (IfExists(row.rowid))
  35.             Future.successful(CQLResult(marshal(0)))
  36.           else
  37.             cqlActionStream.perform(row).map {_ => CQLResult(marshal(1))}
  38.         }
  39.   }
  41.   override def transferRows: Flow[AQMRPTRow, CQLResult, NotUsed] = {
  42.     Flow[AQMRPTRow]
  43.       .via(cqlActionFlow)
  44.   }
  46.   private def IfExists(rowid: Long): Boolean = {
  47.     val cql = "SELECT * FROM testdb.AQMRPT WHERE ROWID = ? ALLOW FILTERING"
  48.     val param = Seq(rowid.asInstanceOf[Object])
  49.     val toRowId: Row => Long = r => r.getLong("rowid")
  50.     val ctx = CQLQueryContext(cql,param)
  51.     val src: Source[Long,NotUsed] = cassandraStream(ctx,toRowId)
  52.     val fut = src.toMat(Sink.headOption)(Keep.right).run()
  54.     val result = Await.result(fut,3 seconds)
  56.     log.info(s"checking existence: ${result}")
  57.     result match {
  58.       case Some(x) => true
  59.       case None => false
  60.     }
  61.   }

在上面的代码里我们调用了Cassandra-Engine的CassandraActionStream类型的流处理方法。值得注意的是这里我们尝试在stream Flow里运算另一个Flow，如：IfExists函数里运算一个Source来确定rowid是否存在。不要在意这个函数的实际应用，它只是一个人为的例子。另外，rowid：Long这样的定义是硬性规定的。cassandra对数据类型的匹配要求很弱智，没有提供任何自然转换。所以，Int <> Long被视为类型错误，而且无法catch任何明白的错误信息。

这项服务的客户端调用如下：

  1.   val stub = CqlGrpcAkkaStream.stub(channel)
  3.   val jdbcRows2transfer = JDBCQueryContext[AQMRPTRow](
  4.     dbName = 'h2,
  5.     statement = "select * from AQMRPT where statename='Arkansas'"
  6.   )
  8.   def toAQMRPTRow: WrappedResultSet => AQMRPTRow = rs => AQMRPTRow(
  9.     rowid = rs.long("ROWID"),
  10.     measureid = rs.long("MEASUREID"),
  11.     statename = rs.string("STATENAME"),
  12.     countyname = rs.string("COUNTYNAME"),
  13.     reportyear = rs.int("REPORTYEAR"),
  14.     value = rs.int("VALUE"),
  15.     created = Some(ProtoDateTime(Some(ProtoDate(1990, 8, 12)), Some(ProtoTime(23, 56, 23, 0))))
  16.   )
  18.   import scala.concurrent.duration._
  20.   def transferRows: Source[CQLResult, NotUsed] = {
  21.     log.info(s"**** calling transferRows ****")
  22.     jdbcAkkaStream(jdbcRows2transfer, toAQMRPTRow)
  23.       //      .throttle(1, 500.millis, 1, ThrottleMode.shaping)
  24.       .via(stub.transferRows)
  25.   }

注意：JDBC在客户端本地，cassandra是远程服务。

最后我们示范一下cassandra Query。.proto DDL 定义：

  1. message CQLQuery {
  2.     string statement = 1;
  3.     bytes parameters = 2;
  4.     google.protobuf.Int32Value consistency = 3;
  5.     google.protobuf.Int32Value fetchSize = 4;
  6. }
  8. service CQLServices {
  9.   rpc transferRows(stream AQMRPTRow) returns (stream CQLResult) {}
  10.   rpc runQuery(CQLQuery) returns (stream AQMRPTRow) {}
  11.   rpc runDDL(CQLUpdate) returns (CQLResult) {}
  12. }

服务函数代码如下：

  1.  def toCQLTimestamp(rs: Row) = {
  2.     try {
  3.       val tm = rs.getTimestamp("CREATED")
  4.       if (tm == null) None
  5.       else {
  6.         val localdt = cqlGetTimestamp(tm)
  7.         Some(ProtoDateTime(Some(ProtoDate(localdt.getYear, localdt.getMonthValue, localdt.getDayOfMonth)),
  8.           Some(ProtoTime(localdt.getHour, localdt.getMinute, localdt.getSecond, localdt.getNano))))
  9.       }
  10.     }
  11.     catch {
  12.       case e: Exception => None
  13.     }
  14.   }
  16.   val toAQMRow: Row => AQMRPTRow = rs=> AQMRPTRow(
  17.     rowid = rs.getLong("ROWID"),
  18.     measureid = rs.getLong("MEASUREID"),
  19.     statename = rs.getString("STATENAME"),
  20.     countyname = rs.getString("COUNTYNAME"),
  21.     reportyear = rs.getInt("REPORTYEAR"),
  22.     value = rs.getInt("VALUE"),
  23.     created = toCQLTimestamp(rs)
  24.   )
  25.   override def runQuery: Flow[CQLQuery, AQMRPTRow, NotUsed] = {
  26.     log.info("**** runQuery called on service side ***")
  27.     Flow[CQLQuery]
  28.       .flatMapConcat { q =>
  29.         //unpack JDBCQuery and construct the context
  30.         var params: Seq[Object] =  Nil
  31.         if (q.parameters != _root_.com.google.protobuf.ByteString.EMPTY)
  32.           params = unmarshal[Seq[Object]](q.parameters)
  33.         log.info(s"**** query parameters: ${params} ****")
  34.         val ctx = CQLQueryContext(q.statement,params)
  35.         CQLEngine.cassandraStream(ctx,toAQMRow)
  36.       }
  37.   }

这里值得看看的一是日期转换，二是对于cassandra parameter Seq[Object]的marshal和unmarshal。客户端代码：

  1.   val query = CQLQuery(
  2.     statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE > ? ALLOW FILTERING;",
  3.     parameters = marshal(Seq("Arkansas", 0.toInt))
  4.   )
  5.   val query2 = CQLQuery (
  6.     statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE = ?",
  7.     parameters = marshal(Seq("Colorado", 3.toInt))
  8.   )
  9.   val query3= CQLQuery (
  10.     statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE = ?",
  11.     parameters = marshal(Seq("Arkansas", 8.toInt))
  12.   )
  13.   def queryRows: Source[AQMRPTRow,NotUsed] = {
  14.     log.info(s"running queryRows ...")
  15.     Source
  16.       .single(query)
  17.       .via(stub.runQuery)
  18.   }

这段相对直白。

下面就是本次讨论涉及的完整源代码：

project/scalapb.sbt

 

  1. addSbtPlugin("com.thesamet" % "sbt-protoc" % "0.99.18")
  3. resolvers += Resolver.bintrayRepo("beyondthelines", "maven")
  5. libraryDependencies ++= Seq(
  6.   "com.thesamet.scalapb" %% "compilerplugin" % "0.7.4",
  7.   "beyondthelines"         %% "grpcakkastreamgenerator" % "0.0.5"
  8. )

 

build.sbt

  1. import scalapb.compiler.Version.scalapbVersion
  2. import scalapb.compiler.Version.grpcJavaVersion
  4. name := "gRPCCassandra"
  6. version := "0.1"
  8. scalaVersion := "2.12.6"
  10. resolvers += Resolver.bintrayRepo("beyondthelines", "maven")
  12. scalacOptions += "-Ypartial-unification"
  14. libraryDependencies := Seq(
  15.   "com.thesamet.scalapb" %% "scalapb-runtime" % scalapbVersion % "protobuf",
  16.   "io.grpc" % "grpc-netty" % grpcJavaVersion,
  17.   "com.thesamet.scalapb" %% "scalapb-runtime-grpc" % scalapbVersion,
  18.   "io.monix" %% "monix" % "2.3.0",
  19.   // for GRPC Akkastream
  20.   "beyondthelines"         %% "grpcakkastreamruntime" % "0.0.5",
  21.   // for scalikejdbc
  22.   "org.scalikejdbc" %% "scalikejdbc"       % "3.2.1",
  23.   "org.scalikejdbc" %% "scalikejdbc-test"   % "3.2.1"   % "test",
  24.   "org.scalikejdbc" %% "scalikejdbc-config"  % "3.2.1",
  25.   "org.scalikejdbc" %% "scalikejdbc-streams" % "3.2.1",
  26.   "org.scalikejdbc" %% "scalikejdbc-joda-time" % "3.2.1",
  27.   "com.h2database"  %  "h2"                % "1.4.196",
  28.   "mysql" % "mysql-connector-java" % "6.0.6",
  29.   "org.postgresql" % "postgresql" % "42.2.0",
  30.   "commons-dbcp" % "commons-dbcp" % "1.4",
  31.   "org.apache.tomcat" % "tomcat-jdbc" % "9.0.2",
  32.   "com.zaxxer" % "HikariCP" % "2.7.4",
  33.   "com.jolbox" % "bonecp" % "0.8.0.RELEASE",
  34.   "com.typesafe.slick" %% "slick" % "3.2.1",
  35.   //for cassandra  340
  36.   "com.datastax.cassandra" % "cassandra-driver-core" % "3.4.0",
  37.   "com.datastax.cassandra" % "cassandra-driver-extras" % "3.4.0",
  38.   "com.typesafe.akka" %% "akka-stream" % "2.5.13",
  39.   "com.lightbend.akka" %% "akka-stream-alpakka-cassandra" % "0.19",
  40.   "ch.qos.logback"  %  "logback-classic"   % "1.2.3",
  41.   "org.typelevel" %% "cats-core" % "1.1.0"
  42. )
  44. PB.targets in Compile := Seq(
  45.   scalapb.gen() -> (sourceManaged in Compile).value,
  46.   // generate the akka stream files
  47.   grpc.akkastreams.generators.GrpcAkkaStreamGenerator() -> (sourceManaged in Compile).value
  48. )

main/resources/application.conf

  1. # JDBC settings
  2. test {
  3.   db {
  4.     h2 {
  5.       driver = "org.h2.Driver"
  6.       url = "jdbc:h2:tcp://localhost/~/slickdemo"
  7.       user = ""
  8.       password = ""
  9.       poolInitialSize = 5
  10.       poolMaxSize = 7
  11.       poolConnectionTimeoutMillis = 1000
  12.       poolValidationQuery = "select 1 as one"
  13.       poolFactoryName = "commons-dbcp2"
  14.     }
  15.   }
  17.   db.mysql.driver = "com.mysql.cj.jdbc.Driver"
  18.   db.mysql.url = "jdbc:mysql://localhost:3306/testdb"
  19.   db.mysql.user = "root"
  20.   db.mysql.password = "123"
  21.   db.mysql.poolInitialSize = 5
  22.   db.mysql.poolMaxSize = 7
  23.   db.mysql.poolConnectionTimeoutMillis = 1000
  24.   db.mysql.poolValidationQuery = "select 1 as one"
  25.   db.mysql.poolFactoryName = "bonecp"
  27.   # scallikejdbc Global settings
  28.   scalikejdbc.global.loggingSQLAndTime.enabled = true
  29.   scalikejdbc.global.loggingSQLAndTime.logLevel = info
  30.   scalikejdbc.global.loggingSQLAndTime.warningEnabled = true
  31.   scalikejdbc.global.loggingSQLAndTime.warningThresholdMillis = 1000
  32.   scalikejdbc.global.loggingSQLAndTime.warningLogLevel = warn
  33.   scalikejdbc.global.loggingSQLAndTime.singleLineMode = false
  34.   scalikejdbc.global.loggingSQLAndTime.printUnprocessedStackTrace = false
  35.   scalikejdbc.global.loggingSQLAndTime.stackTraceDepth = 10
  36. }
  37. dev {
  38.   db {
  39.     h2 {
  40.       driver = "org.h2.Driver"
  41.       url = "jdbc:h2:tcp://localhost/~/slickdemo"
  42.       user = ""
  43.       password = ""
  44.       poolFactoryName = "hikaricp"
  45.       numThreads = 10
  46.       maxConnections = 12
  47.       minConnections = 4
  48.       keepAliveConnection = true
  49.     }
  50.     mysql {
  51.       driver = "com.mysql.cj.jdbc.Driver"
  52.       url = "jdbc:mysql://localhost:3306/testdb"
  53.       user = "root"
  54.       password = "123"
  55.       poolInitialSize = 5
  56.       poolMaxSize = 7
  57.       poolConnectionTimeoutMillis = 1000
  58.       poolValidationQuery = "select 1 as one"
  59.       poolFactoryName = "bonecp"
  61.     }
  62.     postgres {
  63.       driver = "org.postgresql.Driver"
  64.       url = "jdbc:postgresql://localhost:5432/testdb"
  65.       user = "root"
  66.       password = "123"
  67.       poolFactoryName = "hikaricp"
  68.       numThreads = 10
  69.       maxConnections = 12
  70.       minConnections = 4
  71.       keepAliveConnection = true
  72.     }
  73.   }
  74.   # scallikejdbc Global settings
  75.   scalikejdbc.global.loggingSQLAndTime.enabled = true
  76.   scalikejdbc.global.loggingSQLAndTime.logLevel = info
  77.   scalikejdbc.global.loggingSQLAndTime.warningEnabled = true
  78.   scalikejdbc.global.loggingSQLAndTime.warningThresholdMillis = 1000
  79.   scalikejdbc.global.loggingSQLAndTime.warningLogLevel = warn
  80.   scalikejdbc.global.loggingSQLAndTime.singleLineMode = false
  81.   scalikejdbc.global.loggingSQLAndTime.printUnprocessedStackTrace = false
  82.   scalikejdbc.global.loggingSQLAndTime.stackTraceDepth = 10
  83. }

main/resources/logback.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <configuration>
  3.  
  4.     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
  5.         <layout class="ch.qos.logback.classic.PatternLayout">
  6.             <Pattern>
  7.                 %d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n
  8.             </Pattern>
  9.         </layout>
  10.     </appender>
  11.  
  12.     <logger name="sdp.cql" level="info"
  13.             additivity="false">
  14.         <appender-ref ref="STDOUT" />
  15.     </logger>
  16.  
  17.     <logger name="demo.sdp.grpc.cql" level="info"
  18.             additivity="false">
  19.         <appender-ref ref="STDOUT" />
  20.     </logger>
  21.  
  22.     <root level="error">
  23.         <appender-ref ref="STDOUT" />
  24.     </root>
  25.  
  26. </configuration>

main/protobuf/cql.proto

  1. syntax = "proto3";
  3. import "google/protobuf/wrappers.proto";
  4. import "google/protobuf/any.proto";
  5. import "scalapb/scalapb.proto";
  7. option (scalapb.options) = {
  8.   // use a custom Scala package name
  9.   // package_name: "io.ontherocks.introgrpc.demo"
  11.   // don't append file name to package
  12.   flat_package: true
  13.  
  14.   // generate one Scala file for all messages (services still get their own file)
  15.   single_file: true
  16.  
  17.   // add imports to generated file
  18.   // useful when extending traits or using custom types
  19.   // import: "io.ontherocks.hellogrpc.RockingMessage"
  21.   // code to put at the top of generated file
  22.   // works only with `single_file: true`
  23.   //preamble: "sealed trait SomeSealedTrait"
  24. };
  26. /*
  27.  * Demoes various customization options provided by ScalaPBs.
  28.  */
  30. package sdp.grpc.services;
  32. message ProtoDate {
  33.   int32 yyyy = 1;
  34.   int32 mm   = 2;
  35.   int32 dd   = 3;
  36. }
  38. message ProtoTime {
  39.   int32 hh   = 1;
  40.   int32 mm   = 2;
  41.   int32 ss   = 3;
  42.   int32 nnn  = 4;
  43. }
  45. message ProtoDateTime {
  46.    ProtoDate date = 1;
  47.    ProtoTime time = 2;
  48. }
  50. message AQMRPTRow {
  51.     int64 rowid = 1;
  52.     string countyname = 2;
  53.     string statename = 3;
  54.     int64 measureid = 4;
  55.     int32 reportyear = 5;
  56.     int32 value = 6;
  57.     ProtoDateTime created = 7;
  58. }
  60. message CQLResult {
  61.   bytes result = 1;
  62. }
  64. message CQLQuery {
  65.     string statement = 1;
  66.     bytes parameters = 2;
  67.     google.protobuf.Int32Value consistency = 3;
  68.     google.protobuf.Int32Value fetchSize = 4;
  69. }
  71. message CQLUpdate {
  72.     repeated string statements = 1;
  73.     bytes parameters = 2;
  74.     google.protobuf.Int32Value consistency = 3;
  75.     google.protobuf.BoolValue batch = 4;
  76. }
  78. message HelloMsg {
  79.   string hello = 1;
  80. }
  82. service CQLServices {
  83.   rpc clientStreaming(stream HelloMsg) returns (stream HelloMsg) {}
  84.   rpc transferRows(stream AQMRPTRow) returns (stream CQLResult) {}
  85.   rpc runQuery(CQLQuery) returns (stream AQMRPTRow) {}
  86.   rpc runDDL(CQLUpdate) returns (CQLResult) {}
  87. }

logging/log.scala

  1. package sdp.logging
  3. import org.slf4j.Logger
  5. /**
  6.   * Logger which just wraps org.slf4j.Logger internally.
  7.   *
  8.   * @param logger logger
  9.   */
  10. class Log(logger: Logger) {
  12.   // use var consciously to enable squeezing later
  13.   var isDebugEnabled: Boolean = logger.isDebugEnabled
  14.   var isInfoEnabled: Boolean = logger.isInfoEnabled
  15.   var isWarnEnabled: Boolean = logger.isWarnEnabled
  16.   var isErrorEnabled: Boolean = logger.isErrorEnabled
  18.   def withLevel(level: Symbol)(msg: => String, e: Throwable = null): Unit = {
  19.     level match {
  20.       case 'debug | 'DEBUG => debug(msg)
  21.       case 'info | 'INFO => info(msg)
  22.       case 'warn | 'WARN => warn(msg)
  23.       case 'error | 'ERROR => error(msg)
  24.       case _ => // nothing to do
  25.     }
  26.   }
  28.   def debug(msg: => String): Unit = {
  29.     if (isDebugEnabled && logger.isDebugEnabled) {
  30.       logger.debug(msg)
  31.     }
  32.   }
  34.   def debug(msg: => String, e: Throwable): Unit = {
  35.     if (isDebugEnabled && logger.isDebugEnabled) {
  36.       logger.debug(msg, e)
  37.     }
  38.   }
  40.   def info(msg: => String): Unit = {
  41.     if (isInfoEnabled && logger.isInfoEnabled) {
  42.       logger.info(msg)
  43.     }
  44.   }
  46.   def info(msg: => String, e: Throwable): Unit = {
  47.     if (isInfoEnabled && logger.isInfoEnabled) {
  48.       logger.info(msg, e)
  49.     }
  50.   }
  52.   def warn(msg: => String): Unit = {
  53.     if (isWarnEnabled && logger.isWarnEnabled) {
  54.       logger.warn(msg)
  55.     }
  56.   }
  58.   def warn(msg: => String, e: Throwable): Unit = {
  59.     if (isWarnEnabled && logger.isWarnEnabled) {
  60.       logger.warn(msg, e)
  61.     }
  62.   }
  64.   def error(msg: => String): Unit = {
  65.     if (isErrorEnabled && logger.isErrorEnabled) {
  66.       logger.error(msg)
  67.     }
  68.   }
  70.   def error(msg: => String, e: Throwable): Unit = {
  71.     if (isErrorEnabled && logger.isErrorEnabled) {
  72.       logger.error(msg, e)
  73.     }
  74.   }
  76. }

logging/LogSupport.scala

  1. package sdp.logging
  3. import org.slf4j.LoggerFactory
  5. trait LogSupport {
  7.   /**
  8.     * Logger
  9.     */
  10.   protected val log = new Log(LoggerFactory.getLogger(this.getClass))
  12. }

filestreaming/FileStreaming.scala

  1. package sdp.file
  3. import java.io.{ByteArrayInputStream, InputStream}
  4. import java.nio.ByteBuffer
  5. import java.nio.file.Paths
  7. import akka.stream.Materializer
  8. import akka.stream.scaladsl.{FileIO, StreamConverters}
  9. import akka.util._
  11. import scala.concurrent.Await
  12. import scala.concurrent.duration._
  14. object Streaming {
  15.   def FileToByteBuffer(fileName: String, timeOut: FiniteDuration = 60 seconds)(
  16.     implicit mat: Materializer):ByteBuffer = {
  17.     val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
  18.       hd ++ bs
  19.     }
  20.     (Await.result(fut, timeOut)).toByteBuffer
  21.   }
  23.   def FileToByteArray(fileName: String, timeOut: FiniteDuration = 60 seconds)(
  24.     implicit mat: Materializer): Array[Byte] = {
  25.     val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
  26.       hd ++ bs
  27.     }
  28.     (Await.result(fut, timeOut)).toArray
  29.   }
  31.   def FileToInputStream(fileName: String, timeOut: FiniteDuration = 60 seconds)(
  32.     implicit mat: Materializer): InputStream = {
  33.     val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) =>
  34.       hd ++ bs
  35.     }
  36.     val buf = (Await.result(fut, timeOut)).toArray
  37.     new ByteArrayInputStream(buf)
  38.   }
  40.   def ByteBufferToFile(byteBuf: ByteBuffer, fileName: String)(
  41.     implicit mat: Materializer) = {
  42.     val ba = new Array[Byte](byteBuf.remaining())
  43.     byteBuf.get(ba,0,ba.length)
  44.     val baInput = new ByteArrayInputStream(ba)
  45.     val source = StreamConverters.fromInputStream(() => baInput)  //ByteBufferInputStream(bytes))
  46.     source.runWith(FileIO.toPath(Paths.get(fileName)))
  47.   }
  49.   def ByteArrayToFile(bytes: Array[Byte], fileName: String)(
  50.     implicit mat: Materializer) = {
  51.     val bb = ByteBuffer.wrap(bytes)
  52.     val baInput = new ByteArrayInputStream(bytes)
  53.     val source = StreamConverters.fromInputStream(() => baInput) //ByteBufferInputStream(bytes))
  54.     source.runWith(FileIO.toPath(Paths.get(fileName)))
  55.   }
  57.   def InputStreamToFile(is: InputStream, fileName: String)(
  58.     implicit mat: Materializer) = {
  59.     val source = StreamConverters.fromInputStream(() => is)
  60.     source.runWith(FileIO.toPath(Paths.get(fileName)))
  61.   }
  63. }

jdbc/JDBCConfig.scala

  1. package sdp.jdbc.config
  2. import scala.collection.mutable
  3. import scala.concurrent.duration.Duration
  4. import scala.language.implicitConversions
  5. import com.typesafe.config._
  6. import java.util.concurrent.TimeUnit
  7. import java.util.Properties
  8. import scalikejdbc.config._
  9. import com.typesafe.config.Config
  10. import com.zaxxer.hikari._
  11. import scalikejdbc.ConnectionPoolFactoryRepository
  13. /** Extension methods to make Typesafe Config easier to use */
  14. class ConfigExtensionMethods(val c: Config) extends AnyVal {
  15.   import scala.collection.JavaConverters._
  17.   def getBooleanOr(path: String, default: => Boolean = false) = if(c.hasPath(path)) c.getBoolean(path) else default
  18.   def getIntOr(path: String, default: => Int = 0) = if(c.hasPath(path)) c.getInt(path) else default
  19.   def getStringOr(path: String, default: => String = null) = if(c.hasPath(path)) c.getString(path) else default
  20.   def getConfigOr(path: String, default: => Config = ConfigFactory.empty()) = if(c.hasPath(path)) c.getConfig(path) else default
  22.   def getMillisecondsOr(path: String, default: => Long = 0L) = if(c.hasPath(path)) c.getDuration(path, TimeUnit.MILLISECONDS) else default
  23.   def getDurationOr(path: String, default: => Duration = Duration.Zero) =
  24.     if(c.hasPath(path)) Duration(c.getDuration(path, TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS) else default
  26.   def getPropertiesOr(path: String, default: => Properties = null): Properties =
  27.     if(c.hasPath(path)) new ConfigExtensionMethods(c.getConfig(path)).toProperties else default
  29.   def toProperties: Properties = {
  30.     def toProps(m: mutable.Map[String, ConfigValue]): Properties = {
  31.       val props = new Properties(null)
  32.       m.foreach { case (k, cv) =>
  33.         val v =
  34.           if(cv.valueType() == ConfigValueType.OBJECT) toProps(cv.asInstanceOf[ConfigObject].asScala)
  35.           else if(cv.unwrapped eq null) null
  36.           else cv.unwrapped.toString
  37.         if(v ne null) props.put(k, v)
  38.       }
  39.       props
  40.     }
  41.     toProps(c.root.asScala)
  42.   }
  44.   def getBooleanOpt(path: String): Option[Boolean] = if(c.hasPath(path)) Some(c.getBoolean(path)) else None
  45.   def getIntOpt(path: String): Option[Int] = if(c.hasPath(path)) Some(c.getInt(path)) else None
  46.   def getStringOpt(path: String) = Option(getStringOr(path))
  47.   def getPropertiesOpt(path: String) = Option(getPropertiesOr(path))
  48. }
  50. object ConfigExtensionMethods {
  51.   @inline implicit def configExtensionMethods(c: Config): ConfigExtensionMethods = new ConfigExtensionMethods(c)
  52. }
  54. trait HikariConfigReader extends TypesafeConfigReader {
  55.   self: TypesafeConfig =>      // with TypesafeConfigReader => //NoEnvPrefix =>
  57.   import ConfigExtensionMethods.configExtensionMethods
  59.   def getFactoryName(dbName: Symbol): String = {
  60.     val c: Config = config.getConfig(envPrefix + "db." + dbName.name)
  61.     c.getStringOr("poolFactoryName", ConnectionPoolFactoryRepository.COMMONS_DBCP)
  62.   }
  64.   def hikariCPConfig(dbName: Symbol): HikariConfig = {
  66.     val hconf = new HikariConfig()
  67.     val c: Config = config.getConfig(envPrefix + "db." + dbName.name)
  69.     // Connection settings
  70.     if (c.hasPath("dataSourceClass")) {
  71.       hconf.setDataSourceClassName(c.getString("dataSourceClass"))
  72.     } else {
  73.       Option(c.getStringOr("driverClassName", c.getStringOr("driver"))).map(hconf.setDriverClassName _)
  74.     }
  75.     hconf.setJdbcUrl(c.getStringOr("url", null))
  76.     c.getStringOpt("user").foreach(hconf.setUsername)
  77.     c.getStringOpt("password").foreach(hconf.setPassword)
  78.     c.getPropertiesOpt("properties").foreach(hconf.setDataSourceProperties)
  80.     // Pool configuration
  81.     hconf.setConnectionTimeout(c.getMillisecondsOr("connectionTimeout", 1000))
  82.     hconf.setValidationTimeout(c.getMillisecondsOr("validationTimeout", 1000))
  83.     hconf.setIdleTimeout(c.getMillisecondsOr("idleTimeout", 600000))
  84.     hconf.setMaxLifetime(c.getMillisecondsOr("maxLifetime", 1800000))
  85.     hconf.setLeakDetectionThreshold(c.getMillisecondsOr("leakDetectionThreshold", 0))
  86.     hconf.setInitializationFailFast(c.getBooleanOr("initializationFailFast", false))
  87.     c.getStringOpt("connectionTestQuery").foreach(hconf.setConnectionTestQuery)
  88.     c.getStringOpt("connectionInitSql").foreach(hconf.setConnectionInitSql)
  89.     val numThreads = c.getIntOr("numThreads", 20)
  90.     hconf.setMaximumPoolSize(c.getIntOr("maxConnections", numThreads * 5))
  91.     hconf.setMinimumIdle(c.getIntOr("minConnections", numThreads))
  92.     hconf.setPoolName(c.getStringOr("poolName", dbName.name))
  93.     hconf.setRegisterMbeans(c.getBooleanOr("registerMbeans", false))
  95.     // Equivalent of ConnectionPreparer
  96.     hconf.setReadOnly(c.getBooleanOr("readOnly", false))
  97.     c.getStringOpt("isolation").map("TRANSACTION_" + _).foreach(hconf.setTransactionIsolation)
  98.     hconf.setCatalog(c.getStringOr("catalog", null))
  100.     hconf
  102.   }
  103. }
  105. import scalikejdbc._
  106. trait ConfigDBs {
  107.   self: TypesafeConfigReader with TypesafeConfig with HikariConfigReader =>
  109.   def setup(dbName: Symbol = ConnectionPool.DEFAULT_NAME): Unit = {
  110.     getFactoryName(dbName) match {
  111.       case "hikaricp" => {
  112.         val hconf = hikariCPConfig(dbName)
  113.         val hikariCPSource = new HikariDataSource(hconf)
  114.         case class HikariDataSourceCloser(src: HikariDataSource) extends DataSourceCloser {
  115.           var closed = false
  116.           override def close(): Unit = src.close()
  117.         }
  118.         if (hconf.getDriverClassName != null && hconf.getDriverClassName.trim.nonEmpty) {
  119.           Class.forName(hconf.getDriverClassName)
  120.         }
  121.         ConnectionPool.add(dbName, new DataSourceConnectionPool(dataSource = hikariCPSource,settings = DataSourceConnectionPoolSettings(),
  122.           closer = HikariDataSourceCloser(hikariCPSource)))
  123.       }
  124.       case _ => {
  125.         val JDBCSettings(url, user, password, driver) = readJDBCSettings(dbName)
  126.         val cpSettings = readConnectionPoolSettings(dbName)
  127.         if (driver != null && driver.trim.nonEmpty) {
  128.           Class.forName(driver)
  129.         }
  130.         ConnectionPool.add(dbName, url, user, password, cpSettings)
  131.       }
  132.     }
  133.   }
  135.   def setupAll(): Unit = {
  136.     loadGlobalSettings()
  137.     dbNames.foreach { dbName => setup(Symbol(dbName)) }
  138.   }
  140.   def close(dbName: Symbol = ConnectionPool.DEFAULT_NAME): Unit = {
  141.     ConnectionPool.close(dbName)
  142.   }
  144.   def closeAll(): Unit = {
  145.     ConnectionPool.closeAll
  146.   }
  148. }
  151. object ConfigDBs extends ConfigDBs
  152.   with TypesafeConfigReader
  153.   with StandardTypesafeConfig
  154.   with HikariConfigReader
  156. case class ConfigDBsWithEnv(envValue: String) extends ConfigDBs
  157.   with TypesafeConfigReader
  158.   with StandardTypesafeConfig
  159.   with HikariConfigReader
  160.   with EnvPrefix {
  162.   override val env = Option(envValue)
  163. }

jdbc/JDBCEngine.scala

  1. package sdp.jdbc.engine
  2. import java.sql.PreparedStatement
  3. import scala.collection.generic.CanBuildFrom
  4. import akka.stream.scaladsl._
  5. import scalikejdbc._
  6. import scalikejdbc.streams._
  7. import akka.NotUsed
  8. import akka.stream._
  9. import java.time._
  10. import scala.concurrent.duration._
  11. import scala.concurrent._
  12. import sdp.file.Streaming._
  14. import scalikejdbc.TxBoundary.Try._
  16. import scala.concurrent.ExecutionContextExecutor
  17. import java.io.InputStream
  19. import sdp.logging.LogSupport
  21. object JDBCContext {
  22.   type SQLTYPE = Int
  23.   val SQL_EXEDDL= 1
  24.   val SQL_UPDATE = 2
  25.   val RETURN_GENERATED_KEYVALUE = true
  26.   val RETURN_UPDATED_COUNT = false
  28. }
  30. case class JDBCQueryContext[M](
  31.                                 dbName: Symbol,
  32.                                 statement: String,
  33.                                 parameters: Seq[Any] = Nil,
  34.                                 fetchSize: Int = 100,
  35.                                 autoCommit: Boolean = false,
  36.                                 queryTimeout: Option[Int] = None)
  39. case class JDBCContext (
  40.                         dbName: Symbol,
  41.                         statements: Seq[String] = Nil,
  42.                         parameters: Seq[Seq[Any]] = Nil,
  43.                         fetchSize: Int = 100,
  44.                         queryTimeout: Option[Int] = None,
  45.                         queryTags: Seq[String] = Nil,
  46.                         sqlType: JDBCContext.SQLTYPE = JDBCContext.SQL_UPDATE,
  47.                         batch: Boolean = false,
  48.                         returnGeneratedKey: Seq[Option[Any]] = Nil,
  49.                         // no return: None, return by index: Some(1), by name: Some("id")
  50.                         preAction: Option[PreparedStatement => Unit] = None,
  51.                         postAction: Option[PreparedStatement => Unit] = None)
  52.               extends LogSupport {
  54.   ctx =>
  55.  
  56.   //helper functions
  58.   def appendTag(tag: String): JDBCContext = ctx.copy(queryTags = ctx.queryTags :+ tag)
  60.   def appendTags(tags: Seq[String]): JDBCContext = ctx.copy(queryTags = ctx.queryTags ++ tags)
  62.   def setFetchSize(size: Int): JDBCContext = ctx.copy(fetchSize = size)
  64.   def setQueryTimeout(time: Option[Int]): JDBCContext = ctx.copy(queryTimeout = time)
  66.   def setPreAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
  67.     if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
  68.       !ctx.batch && ctx.statements.size == 1) {
  69.       val nc = ctx.copy(preAction = action)
  70.       log.info("setPreAction> set")
  71.       nc
  72.     }
  73.     else {
  74.       log.info("setPreAction> JDBCContex setting error: preAction not supported!")
  75.       throw new IllegalStateException("JDBCContex setting error: preAction not supported!")
  76.     }
  77.   }
  79.   def setPostAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
  80.     if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
  81.       !ctx.batch && ctx.statements.size == 1) {
  82.       val nc = ctx.copy(postAction = action)
  83.       log.info("setPostAction> set")
  84.       nc
  85.     }
  86.     else {
  87.       log.info("setPreAction> JDBCContex setting error: postAction not supported!")
  88.       throw new IllegalStateException("JDBCContex setting error: postAction not supported!")
  89.     }
  90.   }
  92.   def appendDDLCommand(_statement: String, _parameters: Any*): JDBCContext = {
  93.     if (ctx.sqlType == JDBCContext.SQL_EXEDDL) {
  94.       log.info(s"appendDDLCommand> appending: statement: ${_statement}, parameters: ${_parameters}")
  95.       val nc = ctx.copy(
  96.         statements = ctx.statements ++ Seq(_statement),
  97.         parameters = ctx.parameters ++ Seq(Seq(_parameters))
  98.       )
  99.       log.info(s"appendDDLCommand> appended: statement: ${nc.statements}, parameters: ${nc.parameters}")
  100.       nc
  101.     } else {
  102.       log.info(s"appendDDLCommand> JDBCContex setting error: option not supported!")
  103.       throw new IllegalStateException("JDBCContex setting error: option not supported!")
  104.     }
  105.   }
  107.   def appendUpdateCommand(_returnGeneratedKey: Boolean, _statement: String,_parameters: Any*): JDBCContext = {
  108.     if (ctx.sqlType == JDBCContext.SQL_UPDATE && !ctx.batch) {
  109.       log.info(s"appendUpdateCommand> appending: returnGeneratedKey: ${_returnGeneratedKey}, statement: ${_statement}, parameters: ${_parameters}")
  110.       val nc = ctx.copy(
  111.         statements = ctx.statements ++ Seq(_statement),
  112.         parameters = ctx.parameters ++ Seq(_parameters),
  113.         returnGeneratedKey = ctx.returnGeneratedKey ++ (if (_returnGeneratedKey) Seq(Some(1)) else Seq(None))
  114.       )
  115.       log.info(s"appendUpdateCommand> appended: statement: ${nc.statements}, parameters: ${nc.parameters}")
  116.       nc
  117.     } else {
  118.       log.info(s"appendUpdateCommand> JDBCContex setting error: option not supported!")
  119.       throw new IllegalStateException("JDBCContex setting error: option not supported!")
  120.     }
  121.   }
  123.   def appendBatchParameters(_parameters: Any*): JDBCContext = {
  124.     log.info(s"appendBatchParameters> appending:  parameters: ${_parameters}")
  125.     if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch) {
  126.       log.info(s"appendBatchParameters> JDBCContex setting error: batch parameters only supported for SQL_UPDATE and batch = true!")
  127.       throw new IllegalStateException("JDBCContex setting error: batch parameters only supported for SQL_UPDATE and batch = true!")
  128.     }
  129.     var matchParams = true
  130.     if (ctx.parameters != Nil)
  131.       if (ctx.parameters.head.size != _parameters.size)
  132.         matchParams = false
  133.     if (matchParams) {
  134.       val nc = ctx.copy(
  135.         parameters = ctx.parameters ++ Seq(_parameters)
  136.       )
  137.       log.info(s"appendBatchParameters> appended: statement: ${nc.statements}, parameters: ${nc.parameters}")
  138.       nc
  139.     } else {
  140.       log.info(s"appendBatchParameters> JDBCContex setting error: batch command parameters not match!")
  141.       throw new IllegalStateException("JDBCContex setting error: batch command parameters not match!")
  142.     }
  143.   }
  146.   def setBatchReturnGeneratedKeyOption(returnKey: Boolean): JDBCContext = {
  147.     if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch)
  148.       throw new IllegalStateException("JDBCContex setting error: only supported in batch update commands!")
  149.     ctx.copy(
  150.       returnGeneratedKey = if (returnKey) Seq(Some(1)) else Nil
  151.     )
  152.   }
  154.   def setDDLCommand(_statement: String, _parameters: Any*): JDBCContext = {
  155.     log.info(s"setDDLCommand> setting: statement: ${_statement}, parameters: ${_parameters}")
  156.     val nc = ctx.copy(
  157.       statements = Seq(_statement),
  158.       parameters = Seq(_parameters),
  159.       sqlType = JDBCContext.SQL_EXEDDL,
  160.       batch = false
  161.     )
  162.     log.info(s"setDDLCommand> set: statement: ${nc.statements}, parameters: ${nc.parameters}")
  163.     nc
  164.   }
  166.   def setUpdateCommand(_returnGeneratedKey: Boolean, _statement: String,_parameters: Any*): JDBCContext = {
  167.     log.info(s"setUpdateCommand> setting: returnGeneratedKey: ${_returnGeneratedKey}, statement: ${_statement}, parameters: ${_parameters}")
  168.     val nc = ctx.copy(
  169.       statements = Seq(_statement),
  170.       parameters = Seq(_parameters),
  171.       returnGeneratedKey = if (_returnGeneratedKey) Seq(Some(1)) else Seq(None),
  172.       sqlType = JDBCContext.SQL_UPDATE,
  173.       batch = false
  174.     )
  175.     log.info(s"setUpdateCommand> set: statement: ${nc.statements}, parameters: ${nc.parameters}")
  176.     nc
  177.   }
  178.   def setBatchCommand(_statement: String): JDBCContext = {
  179.     log.info(s"setBatchCommand> appending: statement: ${_statement}")
  180.     val nc = ctx.copy (
  181.       statements = Seq(_statement),
  182.       sqlType = JDBCContext.SQL_UPDATE,
  183.       batch = true
  184.     )
  185.     log.info(s"setBatchCommand> set: statement: ${nc.statements}, parameters: ${nc.parameters}")
  186.     nc
  187.   }
  189. }
  191. object JDBCEngine extends LogSupport {
  192.   import JDBCContext._
  194.   type JDBCDate = LocalDate
  195.   type JDBCDateTime = LocalDateTime
  196.   type JDBCTime = LocalTime
  198.   def jdbcSetDate(yyyy: Int, mm: Int, dd: Int) = LocalDate.of(yyyy,mm,dd)
  199.   def jdbcSetTime(hh: Int, mm: Int, ss: Int, nn: Int) = LocalTime.of(hh,mm,ss,nn)
  200.   def jdbcSetDateTime(date: JDBCDate, time: JDBCTime) =  LocalDateTime.of(date,time)
  201.   def jdbcSetNow = LocalDateTime.now()
  203.   def jdbcGetDate(sqlDate: java.sql.Date): java.time.LocalDate = sqlDate.toLocalDate
  204.   def jdbcGetTime(sqlTime: java.sql.Time): java.time.LocalTime = sqlTime.toLocalTime
  205.   def jdbcGetTimestamp(sqlTimestamp: java.sql.Timestamp): java.time.LocalDateTime =
  206.                   sqlTimestamp.toLocalDateTime
  209.   type JDBCBlob = InputStream
  211.   def fileToJDBCBlob(fileName: String, timeOut: FiniteDuration = 60 seconds)(
  212.     implicit mat: Materializer) = FileToInputStream(fileName,timeOut)
  214.   def jdbcBlobToFile(blob: JDBCBlob, fileName: String)(
  215.     implicit mat: Materializer) =  InputStreamToFile(blob,fileName)
  218.   private def noExtractor(message: String): WrappedResultSet => Nothing = { (rs: WrappedResultSet) =>
  219.     throw new IllegalStateException(message)
  220.   }
  222.   def jdbcAkkaStream[A](ctx: JDBCQueryContext[A],extractor: WrappedResultSet => A)
  223.                        (implicit ec: ExecutionContextExecutor): Source[A,NotUsed] = {
  224.       val publisher: DatabasePublisher[A] = NamedDB(ctx.dbName) readOnlyStream {
  225.       val rawSql = new SQLToCollectionImpl[A, NoExtractor](ctx.statement, ctx.parameters)(noExtractor(""))
  226.       ctx.queryTimeout.foreach(rawSql.queryTimeout(_))
  227.       val sql: SQL[A, HasExtractor] = rawSql.map(extractor)
  228.       sql.iterator
  229.         .withDBSessionForceAdjuster(session => {
  230.           session.connection.setAutoCommit(ctx.autoCommit)
  231.           session.fetchSize(ctx.fetchSize)
  232.         })
  233.     }
  234.     log.info(s"jdbcAkkaStream> Source: db: ${ctx.dbName}, statement: ${ctx.statement}, parameters: ${ctx.parameters}")
  235.     Source.fromPublisher[A](publisher)
  236.   }
  239.   def jdbcQueryResult[C[_] <: TraversableOnce[_], A](ctx: JDBCQueryContext[A],
  240.                                                      extractor: WrappedResultSet => A)(
  241.                                                       implicit cbf: CanBuildFrom[Nothing, A, C[A]]): C[A] = {
  242.     val rawSql = new SQLToCollectionImpl[A, NoExtractor](ctx.statement, ctx.parameters)(noExtractor(""))
  243.     ctx.queryTimeout.foreach(rawSql.queryTimeout(_))
  244.     rawSql.fetchSize(ctx.fetchSize)
  245.     try {
  246.       implicit val session = NamedAutoSession(ctx.dbName)
  247.       log.info(s"jdbcQueryResult> Source: db: ${ctx.dbName}, statement: ${ctx.statement}, parameters: ${ctx.parameters}")
  248.       val sql: SQL[A, HasExtractor] = rawSql.map(extractor)
  249.       sql.collection.apply[C]()
  250.     } catch {
  251.       case e: Exception =>
  252.         log.error(s"jdbcQueryResult> runtime error: ${e.getMessage}")
  253.         throw new RuntimeException(s"jdbcQueryResult> Error: ${e.getMessage}")
  255.     }
  257.   }
  259.   def jdbcExecuteDDL(ctx: JDBCContext)(implicit ec: ExecutionContextExecutor): Future[String] = {
  260.     if (ctx.sqlType != SQL_EXEDDL) {
  261.       log.info(s"jdbcExecuteDDL> JDBCContex setting error: sqlType must be 'SQL_EXEDDL'!")
  262.       Future.failed(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_EXEDDL'!"))
  263.     }
  264.     else {
  265.       log.info(s"jdbcExecuteDDL> Source: db: ${ctx.dbName}, statement: ${ctx.statements}, parameters: ${ctx.parameters}")
  266.       Future {
  267.         NamedDB(ctx.dbName) localTx { implicit session =>
  268.           ctx.statements.foreach { stm =>
  269.             val ddl = new SQLExecution(statement = stm, parameters = Nil)(
  270.               before = WrappedResultSet => {})(
  271.               after = WrappedResultSet => {})
  273.             ddl.apply()
  274.           }
  275.           "SQL_EXEDDL executed succesfully."
  276.         }
  277.       }
  278.     }
  279.   }
  281.   def jdbcBatchUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  282.     implicit ec: ExecutionContextExecutor,
  283.              cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  284.     if (ctx.statements == Nil) {
  285.       log.info(s"jdbcBatchUpdate> JDBCContex setting error: statements empty!")
  286.       Future.failed(new IllegalStateException("JDBCContex setting error: statements empty!"))
  287.     }
  288.     if (ctx.sqlType != SQL_UPDATE) {
  289.       log.info(s"jdbcBatchUpdate> JDBCContex setting error: sqlType must be 'SQL_UPDATE'!")
  290.       Future.failed(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!"))
  291.     }
  292.     else {
  293.       if (ctx.batch) {
  294.         if (noReturnKey(ctx)) {
  295.           log.info(s"jdbcBatchUpdate> batch updating no return: db: ${ctx.dbName}, statement: ${ctx.statements}, parameters: ${ctx.parameters}")
  296.           val usql = SQL(ctx.statements.head)
  297.             .tags(ctx.queryTags: _*)
  298.             .batch(ctx.parameters: _*)
  299.           Future {
  300.             NamedDB(ctx.dbName) localTx { implicit session =>
  301.               ctx.queryTimeout.foreach(session.queryTimeout(_))
  302.               usql.apply[Seq]()
  303.               Seq.empty[Long].to[C]
  304.             }
  305.           }
  306.         } else {
  307.           log.info(s"jdbcBatchUpdate> batch updating return genkey: db: ${ctx.dbName}, statement: ${ctx.statements}, parameters: ${ctx.parameters}")
  308.           val usql = new SQLBatchWithGeneratedKey(ctx.statements.head, ctx.parameters, ctx.queryTags)(None)
  309.           Future {
  310.             NamedDB(ctx.dbName) localTx { implicit session =>
  311.               ctx.queryTimeout.foreach(session.queryTimeout(_))
  312.               usql.apply[C]()
  313.             }
  314.           }
  315.         }
  317.       } else {
  318.         log.info(s"jdbcBatchUpdate> JDBCContex setting error: must set batch = true !")
  319.         Future.failed(new IllegalStateException("JDBCContex setting error: must set batch = true !"))
  320.       }
  321.     }
  322.   }
  323.   private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  324.           implicit ec: ExecutionContextExecutor,
  325.                     cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  326.     val Some(key) :: xs = ctx.returnGeneratedKey
  327.     val params: Seq[Any] = ctx.parameters match {
  328.       case Nil => Nil
  329.       case p@_ => p.head
  330.     }
  331.     log.info(s"singleTxUpdateWithReturnKey> updating: db: ${ctx.dbName}, statement: ${ctx.statements}, parameters: ${ctx.parameters}")
  332.     val usql = new SQLUpdateWithGeneratedKey(ctx.statements.head, params, ctx.queryTags)(key)
  333.     Future {
  334.       NamedDB(ctx.dbName) localTx { implicit session =>
  335.         session.fetchSize(ctx.fetchSize)
  336.         ctx.queryTimeout.foreach(session.queryTimeout(_))
  337.         val result = usql.apply()
  338.         Seq(result).to[C]
  339.       }
  340.     }
  341.   }
  343.   private def singleTxUpdateNoReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  344.           implicit ec: ExecutionContextExecutor,
  345.                    cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  346.     val params: Seq[Any] = ctx.parameters match {
  347.       case Nil => Nil
  348.       case p@_ => p.head
  349.     }
  350.     val before = ctx.preAction match {
  351.       case None => pstm: PreparedStatement => {}
  352.       case Some(f) => f
  353.     }
  354.     val after = ctx.postAction match {
  355.       case None => pstm: PreparedStatement => {}
  356.       case Some(f) => f
  357.     }
  358.     log.info(s"singleTxUpdateNoReturnKey> updating: db: ${ctx.dbName}, statement: ${ctx.statements}, parameters: ${ctx.parameters}")
  359.     val usql = new SQLUpdate(ctx.statements.head,params,ctx.queryTags)(before)(after)
  360.     Future {
  361.       NamedDB(ctx.dbName) localTx {implicit session =>
  362.         session.fetchSize(ctx.fetchSize)
  363.         ctx.queryTimeout.foreach(session.queryTimeout(_))
  364.         val result = usql.apply()
  365.         Seq(result.toLong).to[C]
  366.       }
  367.     }
  369.   }
  371.   private def singleTxUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  372.     implicit ec: ExecutionContextExecutor,
  373.              cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  374.     if (noReturnKey(ctx))
  375.       singleTxUpdateNoReturnKey(ctx)
  376.     else
  377.       singleTxUpdateWithReturnKey(ctx)
  378.   }
  380.   private def noReturnKey(ctx: JDBCContext): Boolean = {
  381.     if (ctx.returnGeneratedKey != Nil) {
  382.       val k :: xs = ctx.returnGeneratedKey
  383.       k match {
  384.         case None => true
  385.         case Some(k) => false
  386.       }
  387.     } else true
  388.   }
  390.   def noActon: PreparedStatement=>Unit = pstm => {}
  392.   def multiTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  393.     implicit ec: ExecutionContextExecutor,
  394.              cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  396.     val keys: Seq[Option[Any]] = ctx.returnGeneratedKey match {
  397.       case Nil => Seq.fill(ctx.statements.size)(None)
  398.       case k@_ => k
  399.     }
  400.     val sqlcmd = ctx.statements zip ctx.parameters zip keys
  401.     log.info(s"multiTxUpdates> updating: db: ${ctx.dbName}, SQL Commands: ${sqlcmd}")
  402.     Future {
  403.       NamedDB(ctx.dbName) localTx { implicit session =>
  404.         session.fetchSize(ctx.fetchSize)
  405.         ctx.queryTimeout.foreach(session.queryTimeout(_))
  406.         val results = sqlcmd.map { case ((stm, param), key) =>
  407.           key match {
  408.             case None =>
  409.               new SQLUpdate(stm, param, Nil)(noActon)(noActon).apply().toLong
  410.             case Some(k) =>
  411.               new SQLUpdateWithGeneratedKey(stm, param, Nil)(k).apply().toLong
  412.           }
  413.         }
  414.         results.to[C]
  415.       }
  416.     }
  417.   }
  420.   def jdbcTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  421.     implicit ec: ExecutionContextExecutor,
  422.              cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  423.     if (ctx.statements == Nil) {
  424.       log.info(s"jdbcTxUpdates> JDBCContex setting error: statements empty!")
  425.       Future.failed(new IllegalStateException("JDBCContex setting error: statements empty!"))
  426.     }
  427.     if (ctx.sqlType != SQL_UPDATE) {
  428.       log.info(s"jdbcTxUpdates> JDBCContex setting error: sqlType must be 'SQL_UPDATE'!")
  429.       Future.failed(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!"))
  430.     }
  431.     else {
  432.       if (!ctx.batch) {
  433.         if (ctx.statements.size == 1)
  434.           singleTxUpdate(ctx)
  435.         else
  436.           multiTxUpdates(ctx)
  437.       } else {
  438.         log.info(s"jdbcTxUpdates> JDBCContex setting error: must set batch = false !")
  439.         Future.failed(new IllegalStateException("JDBCContex setting error: must set batch = false !"))
  440.       }
  441.     }
  442.   }
  444.   case class JDBCActionStream[R](dbName: Symbol, parallelism: Int = 1, processInOrder: Boolean = true,
  445.                                  statement: String, prepareParams: R => Seq[Any]) extends LogSupport {
  446.     jas =>
  447.     def setDBName(db: Symbol): JDBCActionStream[R] = jas.copy(dbName = db)
  448.     def setParallelism(parLevel: Int): JDBCActionStream[R] = jas.copy(parallelism = parLevel)
  449.     def setProcessOrder(ordered: Boolean): JDBCActionStream[R] = jas.copy(processInOrder = ordered)
  451.     private def perform(r: R)(implicit ec: ExecutionContextExecutor) = {
  452.       import scala.concurrent._
  453.       val params = prepareParams(r)
  454.       log.info(s"JDBCActionStream.perform>  db: ${dbName}, statement: ${statement}, parameters: ${params}")
  455.       Future {
  456.         NamedDB(dbName) autoCommit { session =>
  457.           session.execute(statement, params: _*)
  458.         }
  459.         r
  460.       }
  461.     }
  462.     def performOnRow(implicit ec: ExecutionContextExecutor): Flow[R, R, NotUsed] =
  463.       if (processInOrder)
  464.         Flow[R].mapAsync(parallelism)(perform)
  465.       else
  466.         Flow[R].mapAsyncUnordered(parallelism)(perform)
  468.   }
  470.   object JDBCActionStream {
  471.     def apply[R](_dbName: Symbol, _statement: String, params: R => Seq[Any]): JDBCActionStream[R] =
  472.       new JDBCActionStream[R](dbName = _dbName, statement=_statement, prepareParams = params)
  473.   }
  475. }

cql/CassandraEngine.scala

  1. package sdp.cql.engine
  3. import akka.NotUsed
  4. import akka.stream.alpakka.cassandra.scaladsl._
  5. import akka.stream.scaladsl._
  6. import com.datastax.driver.core._
  7. import com.google.common.util.concurrent.{FutureCallback, Futures, ListenableFuture}
  9. import scala.collection.JavaConverters._
  10. import scala.collection.generic.CanBuildFrom
  11. import scala.concurrent._
  12. import scala.concurrent.duration.Duration
  13. import sdp.logging.LogSupport
  15. object CQLContext {
  16.   // Consistency Levels
  17.   type CONSISTENCY_LEVEL = Int
  18.   val ANY: CONSISTENCY_LEVEL          =                                        0x0000
  19.   val ONE: CONSISTENCY_LEVEL          =                                        0x0001
  20.   val TWO: CONSISTENCY_LEVEL          =                                        0x0002
  21.   val THREE: CONSISTENCY_LEVEL        =                                        0x0003
  22.   val QUORUM : CONSISTENCY_LEVEL      =                                        0x0004
  23.   val ALL: CONSISTENCY_LEVEL          =                                        0x0005
  24.   val LOCAL_QUORUM: CONSISTENCY_LEVEL =                                        0x0006
  25.   val EACH_QUORUM: CONSISTENCY_LEVEL  =                                        0x0007
  26.   val LOCAL_ONE: CONSISTENCY_LEVEL    =                                        0x000A
  27.   val LOCAL_SERIAL: CONSISTENCY_LEVEL =                                        0x000B
  28.   val SERIAL: CONSISTENCY_LEVEL       =                                        0x000C
  30.   def apply(): CQLContext = CQLContext(statements = Nil)
  32.   def consistencyLevel: CONSISTENCY_LEVEL => ConsistencyLevel = consistency => {
  33.     consistency match {
  34.       case ALL => ConsistencyLevel.ALL
  35.       case ONE => ConsistencyLevel.ONE
  36.       case TWO => ConsistencyLevel.TWO
  37.       case THREE => ConsistencyLevel.THREE
  38.       case ANY => ConsistencyLevel.ANY
  39.       case EACH_QUORUM => ConsistencyLevel.EACH_QUORUM
  40.       case LOCAL_ONE => ConsistencyLevel.LOCAL_ONE
  41.       case QUORUM => ConsistencyLevel.QUORUM
  42.       case SERIAL => ConsistencyLevel.SERIAL
  43.       case LOCAL_SERIAL => ConsistencyLevel.LOCAL_SERIAL
  45.     }
  46.   }
  48. }
  49. case class CQLQueryContext(
  50.                                statement: String,
  51.                                parameter: Seq[Object] = Nil,
  52.                                consistency: Option[CQLContext.CONSISTENCY_LEVEL] = None,
  53.                                fetchSize: Int = 100
  54.                              ) { ctx =>
  55.   def setConsistencyLevel(_consistency: CQLContext.CONSISTENCY_LEVEL): CQLQueryContext =
  56.     ctx.copy(consistency = Some(_consistency))
  57.   def setFetchSize(pageSize: Int): CQLQueryContext =
  58.     ctx.copy(fetchSize = pageSize)
  59.   def setParameters(param: Seq[Object]): CQLQueryContext =
  60.     ctx.copy(parameter = param)
  61. }
  62. object CQLQueryContext {
  63.   def apply[M](stmt: String, param: Seq[Object]): CQLQueryContext = new CQLQueryContext(statement = stmt, parameter = param)
  64. }
  66. case class CQLContext(
  67.                        statements: Seq[String],
  68.                        parameters: Seq[Seq[Object]] = Nil,
  69.                        consistency: Option[CQLContext.CONSISTENCY_LEVEL] = None,
  70.                        batch: Boolean = false
  71.                      ) extends LogSupport { ctx =>
  72.   def setBatch(bat: Boolean) = ctx.copy(batch = bat)
  73.   def setConsistencyLevel(_consistency: CQLContext.CONSISTENCY_LEVEL): CQLContext =
  74.     ctx.copy(consistency = Some(_consistency))
  75.   def setCommand(_statement: String, _parameters: Object*): CQLContext = {
  76.     log.info(s"setCommand> setting: statement: ${_statement}, parameters: ${_parameters}")
  77.     val nc = ctx.copy(statements = Seq(_statement), parameters = Seq(_parameters))
  78.     log.info(s"setCommand> set: statements: ${nc.statements}, parameters: ${nc.parameters}")
  79.     nc
  80.   }
  81.   def appendCommand(_statement: String, _parameters: Object*): CQLContext = {
  82.     log.info(s"appendCommand> appending: statement: ${_statement}, parameters: ${_parameters}")
  83.     val nc = ctx.copy(statements = ctx.statements :+ _statement,
  84.       parameters = ctx.parameters ++ Seq(_parameters))
  85.     log.info(s"appendCommand> appended: statements: ${nc.statements}, parameters: ${nc.parameters}")
  86.     nc
  87.   }
  88. }
  90. object CQLEngine extends LogSupport {
  91.   import CQLContext._
  92.   import CQLHelpers._
  94.   import cats._, cats.data._, cats.implicits._
  95.   import scala.concurrent.{Await, Future}
  96.   import scala.concurrent.duration._
  98.   def fetchResultPage[C[_] <: TraversableOnce[_],A](ctx: CQLQueryContext, pageSize: Int = 100
  99.                                                    ,extractor: Row => A)(
  100.     implicit session: Session, cbf: CanBuildFrom[Nothing, A, C[A]]): (ResultSet, C[A])= {
  102.     val prepStmt = session.prepare(ctx.statement)
  104.     var boundStmt =  prepStmt.bind()
  105.     var params: Seq[Object] = Nil
  106.     if (ctx.parameter != Nil) {
  107.       params = processParameters(ctx.parameter)
  108.       boundStmt = prepStmt.bind(params:_*)
  109.     }
  110.     log.info(s"fetchResultPage>  statement: ${prepStmt.getQueryString}, parameters: ${params}")
  112.     ctx.consistency.foreach {consistency =>
  113.       boundStmt.setConsistencyLevel(consistencyLevel(consistency))}
  115.     val resultSet = session.execute(boundStmt.setFetchSize(pageSize))
  116.     (resultSet,(resultSet.asScala.view.map(extractor)).to[C])
  117.   }
  118.   def fetchMorePages[C[_] <: TraversableOnce[_],A](resultSet: ResultSet, timeOut: Duration)(
  119.     extractor: Row => A)(implicit cbf: CanBuildFrom[Nothing, A, C[A]]): (ResultSet,Option[C[A]]) =
  120.     if (resultSet.isFullyFetched) {
  121.       (resultSet, None)
  122.     } else {
  123.       try {
  124.         val result = Await.result(resultSet.fetchMoreResults(), timeOut)
  125.         (result, Some((result.asScala.view.map(extractor)).to[C]))
  126.       } catch { case e: Throwable => (resultSet, None) }
  127.     }
  129.   def cqlExecute(ctx: CQLContext)(
  130.     implicit session: Session, ec: ExecutionContext): Future[Boolean] = {
  132.     var invalidBat = false
  133.     if ( ctx.batch ) {
  134.       if (ctx.parameters == Nil)
  135.         invalidBat = true
  136.       else if (ctx.parameters.size < 2)
  137.         invalidBat = true;
  138.     }
  139.     if (!ctx.batch || invalidBat) {
  140.       if(invalidBat)
  141.        log.warn(s"cqlExecute> batch update must have at least 2 sets of parameters! change to single-command.")
  143.       if (ctx.statements.size == 1) {
  144.         var param: Seq[Object] = Nil
  145.         if (ctx.parameters != Nil) param =  ctx.parameters.head
  146.         log.info(s"cqlExecute>  single-command: statement: ${ctx.statements.head} parameters: ${param}")
  147.         cqlSingleUpdate(ctx.consistency, ctx.statements.head, param)
  148.       }
  149.       else {
  150.         var params: Seq[Seq[Object]] = Nil
  151.         if (ctx.parameters == Nil)
  152.           params = Seq.fill(ctx.statements.length)(Nil)
  153.         else {
  154.           if (ctx.statements.size > ctx.parameters.size) {
  155.             log.warn(s"cqlExecute> fewer parameters than statements! pad with 'Nil'.")
  156.             val nils = Seq.fill(ctx.statements.size - ctx.parameters.size)(Nil)
  157.             params = ctx.parameters ++ nils
  159.           }
  160.           else
  161.             params = ctx.parameters
  162.         }
  164.         val commands: Seq[(String,Seq[Object])] = ctx.statements zip params
  165.         log.info(s"cqlExecute>  multi-commands: ${commands}")
  166. /*
  167.         //using sequence to flip List[Future[Boolean]] => Future[List[Boolean]]
  168.         //therefore, make sure no command replies on prev command effect
  169.         val lstCmds: List[Future[Boolean]] = commands.map { case (stmt,param) =>
  170.           cqlSingleUpdate(ctx.consistency, stmt, param)
  171.         }.toList
  173.         val futList = lstCmds.sequence.map(_ => true)   //must map to execute
  174.         */
  175. /*
  176.         //using traverse to have some degree of parallelism = max(runtimes)
  177.         //therefore, make sure no command replies on prev command effect
  178.         val futList = Future.traverse(commands) { case (stmt,param)  =>
  179.           cqlSingleUpdate(ctx.consistency, stmt, param)
  180.         }.map(_ => true)
  182.         Await.result(futList, 3 seconds)
  183.         Future.successful(true)
  184. */
  185.         // run sync directly
  186.         Future {
  187.           commands.foreach { case (stm, pars) =>
  188.             cqlExecuteSync(ctx.consistency, stm, pars)
  189.           }
  190.           true
  191.         }
  192.       }
  193.     }
  194.     else
  195.       cqlBatchUpdate(ctx)
  196.   }
  197.   def cqlSingleUpdate(cons: Option[CQLContext.CONSISTENCY_LEVEL],stmt: String, params: Seq[Object])(
  198.          implicit session: Session, ec: ExecutionContext): Future[Boolean] = {
  200.     val prepStmt = session.prepare(stmt)
  202.     var boundStmt = prepStmt.bind()
  203.     var pars: Seq[Object] = Nil
  204.     if (params != Nil) {
  205.       pars = processParameters(params)
  206.       boundStmt = prepStmt.bind(pars: _*)
  207.     }
  208.     log.info(s"cqlSingleUpdate>  statement: ${prepStmt.getQueryString}, parameters: ${pars}")
  210.     cons.foreach { consistency =>
  211.       boundStmt.setConsistencyLevel(consistencyLevel(consistency))
  212.     }
  213.     session.executeAsync(boundStmt).map(_.wasApplied())
  214.   }
  216.   def cqlExecuteSync(cons: Option[CQLContext.CONSISTENCY_LEVEL],stmt: String, params: Seq[Object])(
  217.     implicit session: Session, ec: ExecutionContext): Boolean = {
  219.     val prepStmt = session.prepare(stmt)
  221.     var boundStmt = prepStmt.bind()
  222.     var pars: Seq[Object] = Nil
  223.     if (params != Nil) {
  224.       pars = processParameters(params)
  225.       boundStmt = prepStmt.bind(pars: _*)
  226.     }
  227.     log.info(s"cqlExecuteSync>  statement: ${prepStmt.getQueryString}, parameters: ${pars}")
  229.     cons.foreach { consistency =>
  230.       boundStmt.setConsistencyLevel(consistencyLevel(consistency))
  231.     }
  232.     session.execute(boundStmt).wasApplied()
  234.   }
  236.   def cqlBatchUpdate(ctx: CQLContext)(
  237.     implicit session: Session, ec: ExecutionContext): Future[Boolean] = {
  238.     var params: Seq[Seq[Object]] = Nil
  239.     if (ctx.parameters == Nil)
  240.       params = Seq.fill(ctx.statements.length)(Nil)
  241.     else
  242.       params = ctx.parameters
  243.     log.info(s"cqlBatchUpdate>  statement: ${ctx.statements.head}, parameters: ${params}")
  245.     val prepStmt = session.prepare(ctx.statements.head)
  247.     var batch = new BatchStatement()
  248.     params.foreach { p =>
  249.       log.info(s"cqlBatchUpdate>  batch with raw parameter: ${p}")
  250.       val pars = processParameters(p)
  251.       log.info(s"cqlMultiUpdate>  batch with cooked parameters: ${pars}")
  252.       batch.add(prepStmt.bind(pars: _*))
  253.     }
  254.     ctx.consistency.foreach { consistency =>
  255.       batch.setConsistencyLevel(consistencyLevel(consistency))
  256.     }
  257.     session.executeAsync(batch).map(_.wasApplied())
  258.   }
  260.   def cassandraStream[A](ctx: CQLQueryContext,extractor: Row => A)
  261.                         (implicit session: Session, ec: ExecutionContextExecutor): Source[A,NotUsed] = {
  263.     val prepStmt = session.prepare(ctx.statement)
  264.     var boundStmt =  prepStmt.bind()
  265.     val params = processParameters(ctx.parameter)
  266.     boundStmt = prepStmt.bind(params:_*)
  267.     ctx.consistency.foreach {consistency =>
  268.       boundStmt.setConsistencyLevel(consistencyLevel(consistency))}
  270.     log.info(s"cassandraStream>  statement: ${prepStmt.getQueryString}, parameters: ${params}")
  271.     CassandraSource(boundStmt.setFetchSize(ctx.fetchSize)).map(extractor)
  272.   }
  274.   case class CassandraActionStream[R](parallelism: Int = 1, processInOrder: Boolean = true,
  275.                                       statement: String, prepareParams: R => Seq[Object],
  276.                                       consistency: Option[CQLContext.CONSISTENCY_LEVEL] = None){ cas =>
  277.     def setParallelism(parLevel: Int): CassandraActionStream[R] = cas.copy(parallelism=parLevel)
  278.     def setProcessOrder(ordered: Boolean): CassandraActionStream[R] = cas.copy(processInOrder = ordered)
  279.     def setConsistencyLevel(_consistency: CQLContext.CONSISTENCY_LEVEL): CassandraActionStream[R] =
  280.       cas.copy(consistency = Some(_consistency))
  282.     def perform(r: R)(implicit session: Session, ec: ExecutionContext) = {
  283.       var prepStmt = session.prepare(statement)
  284.       var boundStmt =  prepStmt.bind()
  285.       val params = processParameters(prepareParams(r))
  286.       boundStmt = prepStmt.bind(params: _*)
  287.       consistency.foreach { cons =>
  288.         boundStmt.setConsistencyLevel(CQLContext.consistencyLevel(cons))
  289.       }
  290.       log.info(s"CassandraActionStream.perform>  statement: ${prepStmt.getQueryString}, parameters: ${params}")
  291.       session.executeAsync(boundStmt).map(_ => r)
  292.     }
  294.     def performOnRow(implicit session: Session, ec: ExecutionContext): Flow[R, R, NotUsed] =
  295.       if (processInOrder)
  296.         Flow[R].mapAsync(parallelism)(perform)
  297.       else
  298.         Flow[R].mapAsyncUnordered(parallelism)(perform)
  300.     def unloggedBatch[K](statementBinder: (
  301.       R, PreparedStatement) => BoundStatement,partitionKey: R => K)(
  302.       implicit session: Session, ec: ExecutionContext): Flow[R, R, NotUsed] = {
  303.       val preparedStatement = session.prepare(statement)
  304.       log.info(s"CassandraActionStream.unloggedBatch>  statement: ${preparedStatement.getQueryString}")
  305.       CassandraFlow.createUnloggedBatchWithPassThrough[R, K](
  306.         parallelism,
  307.         preparedStatement,
  308.         statementBinder,
  309.         partitionKey)
  310.     }
  312.   }
  313.   object CassandraActionStream {
  314.     def apply[R](_statement: String, params: R => Seq[Object]): CassandraActionStream[R] =
  315.       new CassandraActionStream[R]( statement=_statement, prepareParams = params)
  316.   }
  319. }
  321. object CQLHelpers extends LogSupport {
  322.   import java.nio.ByteBuffer
  323.   import java.io._
  324.   import java.nio.file._
  325.   import com.datastax.driver.core.LocalDate
  326.   import com.datastax.driver.extras.codecs.jdk8.InstantCodec
  327.   import java.time.Instant
  328.   import akka.stream.scaladsl._
  329.   import akka.stream._
  331.   implicit def listenableFutureToFuture[T](
  332.                                             listenableFuture: ListenableFuture[T]): Future[T] = {
  333.     val promise = Promise[T]()
  334.     Futures.addCallback(listenableFuture, new FutureCallback[T] {
  335.       def onFailure(error: Throwable): Unit = {
  336.         promise.failure(error)
  337.         ()
  338.       }
  339.       def onSuccess(result: T): Unit = {
  340.         promise.success(result)
  341.         ()
  342.       }
  343.     })
  344.     promise.future
  345.   }
  347.   case class CQLDate(year: Int, month: Int, day: Int)
  348.   case object CQLTodayDate
  349.   case class CQLDateTime(year: Int, Month: Int,
  350.                          day: Int, hour: Int, minute: Int, second: Int, millisec: Int = 0)
  351.   case object CQLDateTimeNow
  353.   def cqlGetDate(dateToConvert: java.util.Date): java.time.LocalDate =
  354.           dateToConvert.toInstant()
  355.            .atZone(java.time.ZoneId.systemDefault())
  356.            .toLocalDate()
  358.   def cqlGetTime(dateToConvert: java.util.Date): java.time.LocalTime =
  359.       dateToConvert.toInstant()
  360.         .atZone(java.time.ZoneId.systemDefault())
  361.         .toLocalTime()
  363.   def cqlGetTimestamp(dateToConvert: java.util.Date): java.time.LocalDateTime=
  364.       new java.sql.Timestamp(
  365.            dateToConvert.getTime()
  366.            ).toLocalDateTime()
  368.   def processParameters(params: Seq[Object]): Seq[Object] = {
  369.     import java.time.{Clock,ZoneId}
  370.     log.info(s"[processParameters] input: ${params}")
  371.     val outParams = params.map { obj =>
  372.       obj match {
  373.         case CQLDate(yy, mm, dd) => LocalDate.fromYearMonthDay(yy, mm, dd)
  374.         case CQLTodayDate =>
  375.           val today = java.time.LocalDate.now()
  376.           LocalDate.fromYearMonthDay(today.getYear, today.getMonth.getValue, today.getDayOfMonth)
  377.         case CQLDateTimeNow => Instant.now(Clock.system(ZoneId.of("EST", ZoneId.SHORT_IDS)))
  378.         case CQLDateTime(yy, mm, dd, hr, ms, sc, mi) =>
  379.           Instant.parse(f"$yy%4d-$mm%2d-$dd%2dT$hr%2d:$ms%2d:$sc%2d$mi%3d")
  380.         case p@_ => p
  381.       }
  382.     }
  383.     log.info(s"[processParameters] output: ${params}")
  384.     outParams
  385.   }
  386.   class ByteBufferInputStream(buf: ByteBuffer) extends InputStream {
  387.     override def read: Int = {
  388.       if (!buf.hasRemaining) return -1
  389.       buf.get
  390.     }
  392.     override def read(bytes: Array[Byte], off: Int, len: Int): Int = {
  393.       val length: Int = Math.min(len, buf.remaining)
  394.       buf.get(bytes, off, length)
  395.       length
  396.     }
  397.   }
  398.   object ByteBufferInputStream {
  399.     def apply(buf: ByteBuffer): ByteBufferInputStream = {
  400.       new ByteBufferInputStream(buf)
  401.     }
  402.   }
  403.   class FixsizedByteBufferOutputStream(buf: ByteBuffer) extends OutputStream {
  405.     override def write(b: Int): Unit = {
  406.       buf.put(b.toByte)
  407.     }
  409.     override def write(bytes: Array[Byte], off: Int, len: Int): Unit = {
  410.       buf.put(bytes, off, len)
  411.     }
  412.   }
  413.   object FixsizedByteBufferOutputStream {
  414.     def apply(buf: ByteBuffer) = new FixsizedByteBufferOutputStream(buf)
  415.   }
  416.   class ExpandingByteBufferOutputStream(var buf: ByteBuffer, onHeap: Boolean) extends OutputStream {
  418.     private val increasing = ExpandingByteBufferOutputStream.DEFAULT_INCREASING_FACTOR
  420.     override def write(b: Array[Byte], off: Int, len: Int): Unit = {
  421.       val position = buf.position
  422.       val limit = buf.limit
  423.       val newTotal: Long = position + len
  424.       if(newTotal > limit){
  425.         var capacity = (buf.capacity * increasing)
  426.         while(capacity <= newTotal){
  427.           capacity = (capacity*increasing)
  428.         }
  429.         increase(capacity.toInt)
  430.       }
  432.       buf.put(b, 0, len)
  433.     }
  435.     override def write(b: Int): Unit= {
  436.       if (!buf.hasRemaining) increase((buf.capacity * increasing).toInt)
  437.       buf.put(b.toByte)
  438.     }
  439.     protected def increase(newCapacity: Int): Unit = {
  440.       buf.limit(buf.position)
  441.       buf.rewind
  442.       val newBuffer =
  443.         if (onHeap) ByteBuffer.allocate(newCapacity)
  444.         else  ByteBuffer.allocateDirect(newCapacity)
  445.       newBuffer.put(buf)
  446.       buf.clear
  447.       buf = newBuffer
  448.     }
  449.     def size: Long = buf.position
  450.     def capacity: Long = buf.capacity
  451.     def byteBuffer: ByteBuffer = buf
  452.   }
  453.   object ExpandingByteBufferOutputStream {
  454.     val DEFAULT_INCREASING_FACTOR = 1.5f
  455.     def apply(size: Int, increasingBy: Float, onHeap: Boolean) = {
  456.       if (increasingBy <= 1) throw new IllegalArgumentException("Increasing Factor must be greater than 1.0")
  457.       val buffer: ByteBuffer =
  458.         if (onHeap) ByteBuffer.allocate(size)
  459.         else ByteBuffer.allocateDirect(size)
  460.       new ExpandingByteBufferOutputStream(buffer,onHeap)
  461.     }
  462.     def apply(size: Int): ExpandingByteBufferOutputStream = {
  463.       apply(size, ExpandingByteBufferOutputStream.DEFAULT_INCREASING_FACTOR, false)
  464.     }
  466.     def apply(size: Int, onHeap: Boolean): ExpandingByteBufferOutputStream = {
  467.       apply(size, ExpandingByteBufferOutputStream.DEFAULT_INCREASING_FACTOR, onHeap)
  468.     }
  470.     def apply(size: Int, increasingBy: Float): ExpandingByteBufferOutputStream = {
  471.       apply(size, increasingBy, false)
  472.     }
  474.   }
  475.   def cqlFileToBytes(fileName: String): ByteBuffer = {
  476.     val fis = new FileInputStream(fileName)
  477.     val b = new Array[Byte](fis.available + 1)
  478.     val length = b.length
  479.     fis.read(b)
  480.     ByteBuffer.wrap(b)
  481.   }
  482.   def cqlBytesToFile(bytes: ByteBuffer, fileName: String)(
  483.     implicit mat: Materializer): Future[IOResult] = {
  484.     val source = StreamConverters.fromInputStream(() => ByteBufferInputStream(bytes))
  485.     source.runWith(FileIO.toPath(Paths.get(fileName)))
  486.   }
  487.   def cqlDateTimeString(date: java.util.Date, fmt: String): String = {
  488.     val outputFormat = new java.text.SimpleDateFormat(fmt)
  489.     outputFormat.format(date)
  490.   }
  491.   def useJava8DateTime(cluster: Cluster) = {
  492.     //for jdk8 datetime format
  493.     cluster.getConfiguration().getCodecRegistry()
  494.       .register(InstantCodec.instance)
  495.   }
  496. }

BytesConverter.scala

  1. package protobuf.bytes
  2. import java.io.{ByteArrayInputStream,ByteArrayOutputStream,ObjectInputStream,ObjectOutputStream}
  3. import com.google.protobuf.ByteString
  4. object Converter {
  6.   def marshal(value: Any): ByteString = {
  7.     val stream: ByteArrayOutputStream = new ByteArrayOutputStream()
  8.     val oos = new ObjectOutputStream(stream)
  9.     oos.writeObject(value)
  10.     oos.close()
  11.     ByteString.copyFrom(stream.toByteArray())
  12.   }
  14.   def unmarshal[A](bytes: ByteString): A = {
  15.     val ois = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray))
  16.     val value = ois.readObject()
  17.     ois.close()
  18.     value.asInstanceOf[A]
  19.   }
  21. }

CQLServices.scala

  1. package demo.sdp.grpc.cql.server
  3. import akka.NotUsed
  4. import akka.stream.scaladsl._
  6. import protobuf.bytes.Converter._
  7. import com.datastax.driver.core._
  9. import scala.concurrent.ExecutionContextExecutor
  10. import sdp.grpc.services._
  11. import sdp.cql.engine._
  12. import CQLEngine._
  13. import CQLHelpers._
  14. import sdp.logging.LogSupport
  15. import scala.concurrent._
  16. import scala.concurrent.duration._
  17. import akka.stream.ActorMaterializer
  20. class CQLStreamingServices(implicit ec: ExecutionContextExecutor,
  21.                            mat: ActorMaterializer,  session: Session)
  22.   extends CqlGrpcAkkaStream.CQLServices with LogSupport{
  24.   val toParams: AQMRPTRow => Seq[Object] = row => Seq[Object](
  25.     row.rowid.asInstanceOf[Object],
  26.     row.measureid.asInstanceOf[Object],
  27.     row.statename,
  28.     row.countyname,
  29.     row.reportyear.asInstanceOf[Object],
  30.     row.value.asInstanceOf[Object],
  31.     CQLDateTimeNow
  32.   )
  33.   val cqlInsert ="""
  34.                    |insert into testdb.AQMRPT(
  35.                    | rowid,
  36.                    | measureid,
  37.                    | statename,
  38.                    | countyname,
  39.                    | reportyear,
  40.                    | value,
  41.                    | created)
  42.                    | values(?,?,?,?,?,?,?)
  43.                  """.stripMargin
  45.   val cqlActionStream = CassandraActionStream(cqlInsert,toParams).setParallelism(2)
  46.     .setProcessOrder(false)
  48. /*
  49.   val cqlActionFlow: Flow[AQMRPTRow,AQMRPTRow,NotUsed] =
  50.     Flow[AQMRPTRow]
  51.       .via(cqlActionStream.performOnRow)
  52. */
  54.   val cqlActionFlow: Flow[AQMRPTRow,CQLResult,NotUsed] = {
  55.     Flow[AQMRPTRow]
  56.         .mapAsync(cqlActionStream.parallelism){ row =>
  57.           if (IfExists(row.rowid))
  58.             Future.successful(CQLResult(marshal(0)))
  59.           else
  60.             cqlActionStream.perform(row).map {_ => CQLResult(marshal(1))}
  61.         }
  62.   }
  64.   override def transferRows: Flow[AQMRPTRow, CQLResult, NotUsed] = {
  65.     Flow[AQMRPTRow]
  66.       .via(cqlActionFlow)
  67.   }
  70.   private def IfExists(rowid: Long): Boolean = {
  72.     val cql = "SELECT * FROM testdb.AQMRPT WHERE ROWID = ? ALLOW FILTERING"
  73.     val param = Seq(rowid.asInstanceOf[Object])
  74.     val toRowId: Row => Long = r => r.getLong("rowid")
  75.     val ctx = CQLQueryContext(cql,param)
  76.     val src: Source[Long,NotUsed] = cassandraStream(ctx,toRowId)
  77.     val fut = src.toMat(Sink.headOption)(Keep.right).run()
  79.     val result = Await.result(fut,3 seconds)
  81.     log.info(s"checking existence: ${result}")
  82.     result match {
  83.       case Some(x) => true
  84.       case None => false
  85.     }
  87.   }
  89.   override def clientStreaming: Flow[HelloMsg, HelloMsg, NotUsed] = {
  90.     Flow[HelloMsg]
  91.       .map {r => println(r) ; r}
  92.   }
  94.   override def runDDL: Flow[CQLUpdate, CQLResult, NotUsed] = {
  95.     Flow[CQLUpdate]
  96.       .flatMapConcat { context =>
  97.         //unpack CQLUpdate and construct the context
  98.         val ctx = CQLContext(context.statements)
  99.         log.info(s"**** CQLContext => ${ctx} ***")
  101.         Source
  102.           .fromFuture(cqlExecute(ctx))
  103.           .map { r => CQLResult(marshal(r)) }
  104.       }
  105.   }
  107.   def toCQLTimestamp(rs: Row) = {
  108.     try {
  109.       val tm = rs.getTimestamp("CREATED")
  110.       if (tm == null) None
  111.       else {
  112.         val localdt = cqlGetTimestamp(tm)
  113.         Some(ProtoDateTime(Some(ProtoDate(localdt.getYear, localdt.getMonthValue, localdt.getDayOfMonth)),
  114.           Some(ProtoTime(localdt.getHour, localdt.getMinute, localdt.getSecond, localdt.getNano))))
  115.       }
  116.     }
  117.     catch {
  118.       case e: Exception => None
  119.     }
  120.   }
  122.   val toAQMRow: Row => AQMRPTRow = rs=> AQMRPTRow(
  123.     rowid = rs.getLong("ROWID"),
  124.     measureid = rs.getLong("MEASUREID"),
  125.     statename = rs.getString("STATENAME"),
  126.     countyname = rs.getString("COUNTYNAME"),
  127.     reportyear = rs.getInt("REPORTYEAR"),
  128.     value = rs.getInt("VALUE"),
  129.     created = toCQLTimestamp(rs)
  130.   )
  131.   override def runQuery: Flow[CQLQuery, AQMRPTRow, NotUsed] = {
  132.     log.info("**** runQuery called on service side ***")
  133.     Flow[CQLQuery]
  134.       .flatMapConcat { q =>
  135.         //unpack JDBCQuery and construct the context
  136.         var params: Seq[Object] =  Nil
  137.         if (q.parameters != _root_.com.google.protobuf.ByteString.EMPTY)
  138.           params = unmarshal[Seq[Object]](q.parameters)
  139.         log.info(s"**** query parameters: ${params} ****")
  140.         val ctx = CQLQueryContext(q.statement,params)
  141.         CQLEngine.cassandraStream(ctx,toAQMRow)
  143.       }
  144.   }
  145.   
  146. }

CQLServer.scala

  1. package demo.sdp.grpc.cql.server
  3. import java.util.logging.Logger
  4. import com.datastax.driver.core._
  5. import akka.actor.ActorSystem
  6. import akka.stream.ActorMaterializer
  7. import io.grpc.Server
  8. import io.grpc.ServerBuilder
  9. import sdp.grpc.services._
  10. import sdp.cql.engine._
  11. import CQLHelpers._
  13. class gRPCServer(server: Server) {
  15.   val logger: Logger = Logger.getLogger(classOf[gRPCServer].getName)
  17.   def start(): Unit = {
  18.     server.start()
  19.     logger.info(s"Server started, listening on ${server.getPort}")
  20.     sys.addShutdownHook {
  21.       // Use stderr here since the logger may has been reset by its JVM shutdown hook.
  22.       System.err.println("*** shutting down gRPC server since JVM is shutting down")
  23.       stop()
  24.       System.err.println("*** server shut down")
  25.     }
  26.     ()
  27.   }
  29.   def stop(): Unit = {
  30.     server.shutdown()
  31.   }
  33.   /**
  34.     * Await termination on the main thread since the grpc library uses daemon threads.
  35.     */
  36.   def blockUntilShutdown(): Unit = {
  37.     server.awaitTermination()
  38.   }
  39. }
  41. object CQLServer extends App {
  42.   implicit val cqlsys = ActorSystem("cqlSystem")
  43.   implicit val mat = ActorMaterializer()
  44.   implicit val ec = cqlsys.dispatcher
  46.   val cluster = new Cluster
  47.   .Builder()
  48.     .addContactPoints("localhost")
  49.     .withPort(9042)
  50.     .build()
  52.   useJava8DateTime(cluster)
  53.   implicit val session = cluster.connect()
  55.   val server = new gRPCServer(
  56.     ServerBuilder
  57.       .forPort(50051)
  58.       .addService(
  59.         CqlGrpcAkkaStream.bindService(
  60.           new CQLStreamingServices
  61.         )
  62.       ).build()
  63.   )
  64.   server.start()
  65.   //  server.blockUntilShutdown()
  66.   scala.io.StdIn.readLine()
  67.   session.close()
  68.   cluster.close()
  69.   mat.shutdown()
  70.   cqlsys.terminate()
  71. }

CQLClient.scala

  1. package demo.sdp.grpc.cql.client
  3. import sdp.grpc.services._
  4. import protobuf.bytes.Converter._
  5. import akka.stream.scaladsl._
  6. import akka.NotUsed
  7. import akka.actor.ActorSystem
  8. import akka.stream.{ActorMaterializer, ThrottleMode}
  9. import io.grpc._
  10. import sdp.logging.LogSupport
  11. import sdp.jdbc.engine._
  12. import JDBCEngine._
  13. import scalikejdbc.WrappedResultSet
  14. import sdp.cql.engine.CQLHelpers.CQLDateTimeNow
  15. import scala.util._
  16. import scala.concurrent.ExecutionContextExecutor
  18. class CQLStreamClient(host: String, port: Int)(
  19.   implicit ec: ExecutionContextExecutor) extends LogSupport {
  21.   val channel = ManagedChannelBuilder
  22.     .forAddress(host, port)
  23.     .usePlaintext(true)
  24.     .build()
  27.   val stub = CqlGrpcAkkaStream.stub(channel)
  29.   val jdbcRows2transfer = JDBCQueryContext[AQMRPTRow](
  30.     dbName = 'h2,
  31.     statement = "select * from AQMRPT where statename='Arkansas'"
  32.   )
  35.   def toAQMRPTRow: WrappedResultSet => AQMRPTRow = rs => AQMRPTRow(
  36.     rowid = rs.long("ROWID"),
  37.     measureid = rs.long("MEASUREID"),
  38.     statename = rs.string("STATENAME"),
  39.     countyname = rs.string("COUNTYNAME"),
  40.     reportyear = rs.int("REPORTYEAR"),
  41.     value = rs.int("VALUE"),
  42.     created = Some(ProtoDateTime(Some(ProtoDate(1990, 8, 12)), Some(ProtoTime(23, 56, 23, 0))))
  43.   )
  46.   import scala.concurrent.duration._
  48.   def transferRows: Source[CQLResult, NotUsed] = {
  49.     log.info(s"**** calling transferRows ****")
  50.     jdbcAkkaStream(jdbcRows2transfer, toAQMRPTRow)
  51.       //      .throttle(1, 500.millis, 1, ThrottleMode.shaping)
  52.       .via(stub.transferRows)
  53.   }
  55.   def echoHello: Source[HelloMsg,NotUsed] = {
  56.     val row = HelloMsg("hello world!")
  57.     val rows = List.fill[HelloMsg](100)(row)
  58.     Source
  59.       .fromIterator(() => rows.iterator)
  60.       .via(stub.clientStreaming)
  61.   }
  62.   val query0 = CQLQuery(
  63.     statement = "select * from testdb.AQMRPT"
  64.   )
  66.   val query = CQLQuery(
  67.     statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE > ? ALLOW FILTERING;",
  68.     parameters = marshal(Seq("Arkansas", 0.toInt))
  69.   )
  70.   val query2 = CQLQuery (
  71.     statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE = ?",
  72.     parameters = marshal(Seq("Colorado", 3.toInt))
  73.   )
  74.   val query3= CQLQuery (
  75.     statement = "select * from testdb.AQMRPT where STATENAME = ? and VALUE = ?",
  76.     parameters = marshal(Seq("Arkansas", 8.toInt))
  77.   )
  79.   def queryRows: Source[AQMRPTRow,NotUsed] = {
  80.     log.info(s"running queryRows ...")
  81.     Source
  82.       .single(query)
  83.       .via(stub.runQuery)
  84.   }
  86.   val dropCQL = "DROP TABLE IF EXISTS testdb.AQMRPT"
  88.   val createCQL ="""
  89.   CREATE TABLE testdb.AQMRPT (
  90.      rowid bigint primary key,
  91.      measureid bigint,
  92.      statename text,
  93.      countyname text,
  94.      reportyear int,
  95.      value int,
  96.      created timestamp
  97.   )"""
  99.   val cqlddl = CQLUpdate(statements = Seq(dropCQL,createCQL))
  100.   def createTbl: Source[CQLResult,NotUsed] = {
  101.     log.info(s"running createTbl ...")
  102.     Source
  103.       .single(cqlddl)
  104.       .via(stub.runDDL)
  105.   }
  106.   
  107. }
  110. object EchoHelloClient extends App {
  111.   implicit val system = ActorSystem("EchoNumsClient")
  112.   implicit val mat = ActorMaterializer.create(system)
  113.   implicit val ec = system.dispatcher
  114.   val client = new CQLStreamClient("localhost", 50051)
  116.   client.echoHello.runForeach(println)
  118.   scala.io.StdIn.readLine()
  119.   mat.shutdown()
  120.   system.terminate()
  122. }
  125. object TransferRows extends App {
  127.   import sdp.jdbc.config._
  129.   implicit val system = ActorSystem("JDBCServer")
  130.   implicit val mat = ActorMaterializer.create(system)
  131.   implicit val ec = system.dispatcher
  133.   ConfigDBsWithEnv("dev").setup('h2)
  134.   ConfigDBsWithEnv("dev").loadGlobalSettings()
  137.   val client = new CQLStreamClient("localhost", 50051)
  138.   val fut = client.transferRows.runFold(0){(a,b) => a + unmarshal[Int](b.result)}
  139.   fut.onComplete {
  140.     case scala.util.Success(cnt) => println(s"done transfer ${cnt} rows.")
  141.     case Failure(e) => println(s"!!!!!streaming error: ${e.getMessage}")
  142.   }
  144.   scala.io.StdIn.readLine()
  145.   ConfigDBsWithEnv("dev").close('h2)
  146.   mat.shutdown()
  147.   system.terminate()
  150. }
  152. object QueryRows extends App {
  153.   implicit val system = ActorSystem("QueryRows")
  154.   implicit val mat = ActorMaterializer.create(system)
  155.   implicit val ec = system.dispatcher
  158.   val client = new CQLStreamClient("localhost", 50051)
  160.   val fut = client.queryRows.runForeach { r => println(r) }
  161.   fut.onComplete {
  162.     case scala.util.Success(d) => println(s"done querying.")
  163.     case Failure(e) => println(s"!!!!!query error: ${e.getMessage}")
  164.   }
  166.   scala.io.StdIn.readLine()
  167.   mat.shutdown()
  168.   system.terminate()
  170. }
  172. object RunDDL extends App {
  173.   implicit val system = ActorSystem("RunDDL")
  174.   implicit val mat = ActorMaterializer.create(system)
  175.   implicit val ec = system.dispatcher
  177.   val client = new CQLStreamClient("localhost", 50051)
  179.   client.createTbl.runForeach { r => println(unmarshal(r.result)) }
  182.   scala.io.StdIn.readLine()
  183.   mat.shutdown()
  184.   system.terminate()
  186. }

 

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Scala热门文章

  • scala (3) Function 和 Method
  • flink 两个datastream实现left_join的两种方法
  • scala case class + BeanProperty+fastjson 快速实现类的序列化与反序列化
  • Scala解析MDF文件
  • flink 如何实现对watermark 的checkpoint,防止数据复写
  • Spark笔记(一):错误总结
  • 踩坑 scala 打包上传 找不到类
  • akka-typed(2) - typed-actor交流方式和交流协议
  • Scala(二)：元组、数组、映射
  • scala 命名规范

    • Scala推荐文章

  • Scala中Array和List的区别说明
  • alpakka-kafka(5)-kafka集群配置与分布式应用部署
  • alpakka-kafka(2)-consumer
  • alpakka-kafka(1)-producer
  • akka-typed(1) - actor生命周期管理
  • akka-typed(2) - typed-actor交流方式和交流协议
  • akka-typed(3) - PersistentActor has EventSourcedBehavior
  • akka-typed(4) - EventSourcedBehavior in action
  • kka-typed(5) - cluster：集群节点状态监视
  • akka-typed(6) - cluster：group router, cluster-load-balancing

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号