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

   在一个akka-cluster环境里，从数据调用的角度上，JDBC数据库与集群中其它节点是脱离的。这是因为JDBC数据库不是分布式的，不具备节点位置透明化特性。所以，JDBC数据库服务器必须通过服务方式来向外提供数据操。在这种场景里服务端是JDBC服务，其它节点，包括其它的JDBC数据库节点都是这个JDBC服务的调用客户端。因为我们已经明确选择了在akka-cluster集群环境里实施gRPC服务模式，通过akka-stream的流控制方式实现数据库操作的程序控制，所以在本次讨论里我们将示范说明gRPC-JDBC-Streaming的具体实现和使用方式。

在上次的讨论里我们已经示范了最简单的JDBC-Streaming Unary request/response模式：从客户端向JDBC-Service发送一个JDBCQuery、JDBC服务端运行JDBCQuery后向客户端返回一个数据流DataRows。jdbc.proto文件里用IDL定义的数据和服务类型如下：

  1. message JDBCDataRow {
  2.  string year = 1;
  3.  string state = 2;
  4.  string county = 3;
  5.  string value = 4;
  6. }
  8. message JDBCQuery {
  9.   string dbName = 1;
  10.   string statement = 2;
  11.   bytes parameters = 3;
  12.   google.protobuf.Int32Value fetchSize= 4;
  13.   google.protobuf.BoolValue autoCommit = 5;
  14.   google.protobuf.Int32Value queryTimeout = 6;
  15. }
  17. service JDBCServices {
  18.   rpc runQuery(JDBCQuery) returns (stream JDBCDataRow) {}
  19. }

以上数据类型JDBCDataRow和JDBCQuery分别对应JDBC-Streaming工具的流元素结构和JDBCQueryContext，如下：

  1.   val toRow = (rs: WrappedResultSet) => JDBCDataRow(
  2.     year = rs.string("REPORTYEAR"),
  3.     state = rs.string("STATENAME"),
  4.     county = rs.string("COUNTYNAME"),
  5.     value = rs.string("VALUE")
  6.   )
  8.    val ctx = JDBCQueryContext[JDBCDataRow](
  9.      dbName = Symbol(q.dbName),
  10.      statement = q.statement,
  11.      parameters = params，
  12.      fetchSize = q.fetchSize.getOrElse(100),
  13.      autoCommit = q.autoCommit.getOrElse(false),
  14.      queryTimeout = q.queryTimeout
  15.    )
  16.    jdbcAkkaStream(ctx, toRow)

用scalaPB编译后自动产生服务端和客户端框架代码（boilerplate-code）。我们需要实现具体的JDBC服务：

  1. class JDBCStreamingServices(implicit ec: ExecutionContextExecutor) extends JdbcGrpcAkkaStream.JDBCServices {
  2.   val logger = Logger.getLogger(classOf[JDBCStreamingServices].getName)
  4.   val toRow = (rs: WrappedResultSet) => JDBCDataRow(
  5.     year = rs.string("REPORTYEAR"),
  6.     state = rs.string("STATENAME"),
  7.     county = rs.string("COUNTYNAME"),
  8.     value = rs.string("VALUE")
  9.   )
  10.   override def runQuery: Flow[JDBCQuery, JDBCDataRow, NotUsed] = {
  11.     logger.info("**** runQuery called on service side ***")
  12.     Flow[JDBCQuery]
  13.       .flatMapConcat { q =>
  14.         //unpack JDBCQuery and construct the context
  15.         val params: Seq[Any] = unmarshal[Seq[Any]](q.parameters)
  16.         logger.info(s"**** query parameters: ${params} ****")
  17.         val ctx = JDBCQueryContext[JDBCDataRow](
  18.           dbName = Symbol(q.dbName),
  19.           statement = q.statement,
  20.           parameters = params,
  21.           fetchSize = q.fetchSize.getOrElse(100),
  22.           autoCommit = q.autoCommit.getOrElse(false),
  23.           queryTimeout = q.queryTimeout
  24.         )
  25.         jdbcAkkaStream(ctx, toRow)
  26.       }
  27.   }
  28. }

下面是客户端调用服务示范：

  1.   val query = JDBCQuery (
  2.     dbName = "h2",
  3.     statement = "select * from AQMRPT where STATENAME = ? and VALUE = ?",
  4.     parameters = marshal(Seq("Arizona", 5))
  5.   )
  6.   def queryRows: Source[JDBCDataRow,NotUsed] = {
  7.     logger.info(s"running queryRows ...")
  8.     Source
  9.       .single(query)
  10.       .via(stub.runQuery)
  11.   }

这个程序的运行方式如下：

  1. object QueryRows extends App {
  2.   implicit val system = ActorSystem("QueryRows")
  3.   implicit val mat = ActorMaterializer.create(system)
  5.   val client = new JDBCStreamClient("localhost", 50051)
  7.   client.queryRows.runForeach(println)
  9.   scala.io.StdIn.readLine()
  10.   mat.shutdown()
  11.   system.terminate()
  12. }

那么如果从客户端发出一串的JDBCQuery又如何呢？这也是所谓的BiDi-Streaming模式，在jdbc.proto的服务描述如下：

  1. service JDBCServices {
  2.   rpc runQuery(JDBCQuery) returns (stream JDBCDataRow) {}
  3.   rpc batQuery(stream JDBCQuery) returns (stream JDBCDataRow) {}
  4. }

我们看到batQuery的入参是一个stream。自动产生的服务函数batQuery款式是这样的：

  1.   override def runQuery: Flow[JDBCQuery, JDBCDataRow, NotUsed] = { ... }
  2.   override def batQuery: Flow[JDBCQuery, JDBCDataRow, NotUsed] = runQuery

runQuery和batQuery的函数款式是一样的。这就说明服务端提供的服务模式是一样的。在我们这个例子里它们都是对每个收到的JDBCQuery发还相关的数据流。实际上这两项服务的区别在客户方。下面是scalaPB产生的源代码：

  1.    override def runQuery: Flow[grpc.jdbc.services.JDBCQuery, grpc.jdbc.services.JDBCDataRow, NotUsed] =
  2.       Flow.fromGraph(
  3.         new GrpcGraphStage[grpc.jdbc.services.JDBCQuery, grpc.jdbc.services.JDBCDataRow]({ outputObserver =>
  4.           new StreamObserver[grpc.jdbc.services.JDBCQuery] {
  5.             override def onError(t: Throwable): Unit = ()
  6.             override def onCompleted(): Unit = ()
  7.             override def onNext(request: grpc.jdbc.services.JDBCQuery): Unit =
  8.               ClientCalls.asyncServerStreamingCall(
  9.                 channel.newCall(METHOD_RUN_QUERY, options),
  10.                 request,
  11.                 outputObserver
  12.               )
  13.           }
  14.         })
  15.       )
  16.  ...
  17.       override def batQuery: Flow[grpc.jdbc.services.JDBCQuery, grpc.jdbc.services.JDBCDataRow, NotUsed] =
  18.         Flow.fromGraph(new GrpcGraphStage[grpc.jdbc.services.JDBCQuery, grpc.jdbc.services.JDBCDataRow](outputObserver =>
  19.           ClientCalls.asyncBidiStreamingCall(
  20.             channel.newCall(METHOD_BAT_QUERY, options),
  21.             outputObserver
  22.           )
  23.         ))

所以在客户端我们调用batQuery：

  1.   def batQueryRows: Source[JDBCDataRow,NotUsed] = {
  2.     logger.info(s"running batQueryRows ...")
  3.     Source
  4.       .fromIterator(() => List(query,query2,query3).toIterator)
  5.       .via(stub.batQuery)
  6.   }

JDBC操作除Query之外还应该具备数据更新部分，包括Schema DDL和database-updates。JDBC-update是通过JDBCContext来传递更新要求的：

  1. case class JDBCContext(
  2.                         dbName: Symbol,
  3.                         statements: Seq[String] = Nil,
  4.                         parameters: Seq[Seq[Any]] = Nil,
  5.                         fetchSize: Int = 100,
  6.                         queryTimeout: Option[Int] = None,
  7.                         queryTags: Seq[String] = Nil,
  8.                         sqlType: JDBCContext.SQLTYPE = JDBCContext.SQL_UPDATE,
  9.                         batch: Boolean = false,
  10.                         returnGeneratedKey: Seq[Option[Any]] = Nil,
  11.                         // no return: None, return by index: Some(1), by name: Some("id")
  12.                         preAction: Option[PreparedStatement => Unit] = None,
  13.                         postAction: Option[PreparedStatement => Unit] = None) {... }

这个class对应的protobuf message定义如下：

  1. message JDBCResult {
  2.   bytes result = 1;
  3. }
  5. message JDBCUpdate {
  6.   string dbName = 1;
  7.   repeated string statements = 2;
  8.   bytes parameters = 3;
  9.   google.protobuf.Int32Value fetchSize= 4;
  10.   google.protobuf.Int32Value queryTimeout = 5;
  11.   int32 sqlType = 6;
  12.   google.protobuf.Int32Value batch = 7;
  13.   bytes returnGeneratedKey = 8;
  14. }
  16. service JDBCServices {
  17.   rpc runQuery(JDBCQuery) returns (stream JDBCDataRow) {}
  18.   rpc batQuery(stream JDBCQuery) returns (stream JDBCDataRow) {}
  19.   rpc runDDL(JDBCUpdate) returns (JDBCResult) {}
  20. }

服务函数runDDL返回消息类型JDBCResult： 包嵌一个Seq[Any]类型的返回值。下面是JDBCContext的protobuf message打包、还原使用方法示范，在服务端把JDBCUpdate拆解构建JDBCContext后调用jdbcExecuteDDL：

  1.  override def runDDL: Flow[JDBCUpdate, JDBCResult, NotUsed] = {
  2.     logger.info("**** runDDL called on service side ***")
  3.     Flow[JDBCUpdate]
  4.       .flatMapConcat { context =>
  5.         //unpack JDBCUpdate and construct the context
  7.         val ctx = JDBCContext(
  8.           dbName = Symbol(context.dbName),
  9.           statements = context.statements,
  10.           sqlType = JDBCContext.SQL_EXEDDL,
  11.           queryTimeout = context.queryTimeout
  12.         )
  14.         logger.info(s"**** JDBCContext => ${ctx} ***")
  16.         Source
  17.           .fromFuture(jdbcExecuteDDL(ctx))
  18.           .map { r => JDBCResult(marshal(r)) }
  20.       }
  21.   }

jdbcExecuteDDL返回Future[String]，如下：

  1.   def jdbcExecuteDDL(ctx: JDBCContext)(implicit ec: ExecutionContextExecutor): Future[String] = {
  2.     if (ctx.sqlType != SQL_EXEDDL) {
  3.       Future.failed(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_EXEDDL'!"))
  4.     }
  5.     else {
  6.       Future {
  7.         NamedDB(ctx.dbName) localTx { implicit session =>
  8.           ctx.statements.foreach { stm =>
  9.             val ddl = new SQLExecution(statement = stm, parameters = Nil)(
  10.               before = WrappedResultSet => {})(
  11.               after = WrappedResultSet => {})
  13.             ddl.apply()
  14.           }
  15.           "SQL_EXEDDL executed succesfully."
  16.         }
  17.       }
  18.     }
  19.   }

我们可以用Source.fromFuture(jdbcExecuteDDL(cox))来构建一个akka-stream Source。 在客户端构建一个JDBCUpdate结构传给服务端进行运算：

  1.   val dropSQL: String ="""
  2.       drop table members
  3.     """
  5.   val createSQL: String ="""
  6.     create table members (
  7.       id serial not null primary key,
  8.       name varchar(30) not null,
  9.       description varchar(1000),
  10.       birthday date,
  11.       created_at timestamp not null,
  12.       picture blob
  13.     )"""
  14.   val ctx = JDBCUpdate (
  15.     dbName = "h2",
  16.     sqlType = JDBCContext.SQL_EXEDDL,
  17.     statements = Seq(dropSQL,createSQL)
  18.   )
  20.   def createTbl: Source[JDBCResult,NotUsed] = {
  21.     logger.info(s"running createTbl ...")
  22.     Source
  23.       .single(ctx)
  24.       .via(stub.runDDL)
  25.   }

注意：statements = Seq(dropSQL,createSQL)包含了两个独立的SQL运算。

下面我们示范一下从客户端传送一个数据流（stream MemberRow），由服务端插入数据库操作。DDL数据类型和服务函数定义如下：

  1. message JDBCDate {
  2.   int32 yyyy = 1;
  3.   int32 mm   = 2;
  4.   int32 dd   = 3;
  5. }
  7. message JDBCTime {
  8.   int32 hh   = 1;
  9.   int32 mm   = 2;
  10.   int32 ss   = 3;
  11.   int32 nnn  = 4;
  12. }
  14. message JDBCDateTime {
  15.    JDBCDate date = 1;
  16.    JDBCTime time = 2;
  17. }
  18. message MemberRow {
  19.   string name = 1;
  20.   JDBCDate birthday = 2;
  21.   string description = 3;
  22.   JDBCDateTime created_at = 4;
  23.   bytes picture = 5;
  24. }
  26. service JDBCServices {
  27.   rpc runQuery(JDBCQuery) returns (stream JDBCDataRow) {}
  28.   rpc batQuery(stream JDBCQuery) returns (stream JDBCDataRow) {}
  29.   rpc runDDL(JDBCUpdate) returns (JDBCResult) {}
  30.   rpc insertRows(stream MemberRow) returns(JDBCResult) {}
  31. }

insertRows服务函数的实现如下：

  1.  override def insertRows: Flow[MemberRow, JDBCResult, NotUsed] = {
  2.     logger.info("**** insertRows called on service side ***")
  3.     val insertSQL = """
  4.       insert into members(
  5.         name,
  6.         birthday,
  7.         description,
  8.         created_at
  9.       ) values (?, ?, ?, ?)
  10.     """
  11.     Flow[MemberRow]
  12.       .flatMapConcat { row =>
  13.         val ctx = JDBCContext('h2)
  14.           .setUpdateCommand(true,insertSQL,
  15.              row.name,
  16.              jdbcSetDate(row.birthday.get.yyyy,row.birthday.get.mm,row.birthday.get.dd),
  17.              row.description,
  18.              jdbcSetNow
  19.           )
  21.         logger.info(s"**** JDBCContext => ${ctx} ***")
  23.         Source
  24.           .fromFuture(jdbcTxUpdates[Vector](ctx))
  25.           .map { r => JDBCResult(marshal(r)) }
  26.       }
  27.   }

同样，这个jdbcTxUpdates返回结果是Future类型。具体实现在附件的JDBCEngine.scala中。

客户端构建一个MemberRow流，然后经过stub.insertRows发送给服务端：

  1.   val p1 = MemberRow( "Peter Chan",Some(JDBCDate(1967,5,17)),"new member1",None,_root_.com.google.protobuf.ByteString.EMPTY)
  2.   val p2 = MemberRow( "Alanda Wong",Some(JDBCDate(1980,11,10)),"new member2",None,_root_.com.google.protobuf.ByteString.EMPTY)
  3.   val p3 = MemberRow( "Kate Zhang",Some(JDBCDate(1969,8,13)),"new member3",None,_root_.com.google.protobuf.ByteString.EMPTY)
  4.   val p4 = MemberRow( "Tiger Chan",Some(JDBCDate(1962,5,1)),"new member4",None,_root_.com.google.protobuf.ByteString.EMPTY)
  6.   def insertRows: Source[JDBCResult,NotUsed] = {
  7.     logger.info(s"running insertRows ...")
  8.     Source
  9.       .fromIterator(() => List(p1,p2,p3,p4).toIterator)
  10.       .via(stub.insertRows)
  11.   }

最后，我们再示范jdbcBatchUpdate函数的使用。我们从服务端读取MemberRow再传回服务端进行更新操作。DDL如下：

  1. message MemberRow {
  2.   string name = 1;
  3.   JDBCDate birthday = 2;
  4.   string description = 3;
  5.   JDBCDateTime created_at = 4;
  6.   bytes picture = 5;
  7. }
  9. service JDBCServices {
  10.   rpc runQuery(JDBCQuery) returns (stream JDBCDataRow) {}
  11.   rpc batQuery(stream JDBCQuery) returns (stream JDBCDataRow) {}
  12.   rpc runDDL(JDBCUpdate) returns (JDBCResult) {}
  13.   rpc insertRows(stream MemberRow) returns(JDBCResult) {}
  14.   rpc updateRows(stream MemberRow) returns(JDBCResult) {}
  15.   rpc getMembers(JDBCQuery) returns (stream MemberRow) {}
  16. }

服务端函数定义如下：

  1.  val toMemberRow = (rs: WrappedResultSet) => MemberRow(
  2.     name = rs.string("name"),
  3.     description = rs.string("description"),
  4.     birthday = None,
  5.     createdAt = None,
  6.     picture = _root_.com.google.protobuf.ByteString.EMPTY
  7.   )
  8.   override def getMembers: Flow[JDBCQuery, MemberRow, NotUsed] = {
  9.     logger.info("**** getMembers called on service side ***")
  10.     Flow[JDBCQuery]
  11.       .flatMapConcat { q =>
  12.         //unpack JDBCQuery and construct the context
  13.         var params: Seq[Any] =  Nil
  14.         if (q.parameters != _root_.com.google.protobuf.ByteString.EMPTY)
  15.            params = unmarshal[Seq[Any]](q.parameters)
  16.         logger.info(s"**** query parameters: ${params} ****")
  17.         val ctx = JDBCQueryContext[MemberRow](
  18.           dbName = Symbol(q.dbName),
  19.           statement = q.statement,
  20.           parameters = params,
  21.           fetchSize = q.fetchSize.getOrElse(100),
  22.           autoCommit = q.autoCommit.getOrElse(false),
  23.           queryTimeout = q.queryTimeout
  24.         )
  25.         jdbcAkkaStream(ctx, toMemberRow)
  26.       }
  27.   }
  28. override def updateRows: Flow[MemberRow, JDBCResult, NotUsed] = {
  29.     logger.info("**** updateRows called on service side ***")
  30.     val updateSQL = "update members set description = ?, created_at = ? where name = ?"
  32.     Flow[MemberRow]
  33.       .flatMapConcat { row =>
  34.         val ctx = JDBCContext('h2)
  35.             .setBatchCommand(updateSQL)
  36.             .appendBatchParameters(
  37.               row.name + " updated.",
  38.               jdbcSetNow,
  39.               row.name
  40.             ).setBatchReturnGeneratedKeyOption(true)
  42.         logger.info(s"**** JDBCContext => ${ctx} ***")
  44.         Source
  45.           .fromFuture(jdbcBatchUpdate[Vector](ctx))
  46.           .map { r => JDBCResult(marshal(r)) }
  47.       }
  48.   }

jdbcBatchUpdate函数的源代码在附件JDBCEngine.scala中。客户端代码如下：

  1.   val queryMember = JDBCQuery (
  2.     dbName = "h2",
  3.     statement = "select * from members"
  4.   )
  5.   def updateRows: Source[JDBCResult,NotUsed] = {
  6.     logger.info(s"running updateRows ...")
  7.     Source
  8.       .single(queryMember)
  9.       .via(stub.getMembers)
  10.       .via(stub.updateRows)
  11.   }

下面的例子示范了如何利用JDBCActionStream来批量处理数据。服务端的源代码如下：

  1.   val params: JDBCDataRow => Seq[Any] = row => {
  2.     Seq((row.value.toInt * 2), row.state, row.county, row.year) }
  3.   val sql = "update AQMRPT set total = ? where statename = ? and countyname = ? and reportyear = ?"
  5.   val jdbcActionStream = JDBCActionStream('h2,sql ,params)
  6.     .setParallelism(4).setProcessOrder(false)
  7.   val jdbcActionFlow = jdbcActionStream.performOnRow
  9.   override def updateBat: Flow[JDBCDataRow, JDBCDataRow, NotUsed] = {
  10.     logger.info("**** updateBat called on service side ***")
  11.     Flow[JDBCDataRow]
  12.          .via(jdbcActionFlow)
  13.   }

jdbcActionFlow是一个Flow[R,R,_]，所以我们直接用via把它连接到上一个Flow。下面是JDBCActionStream的定义代码：

  1.   case class JDBCActionStream[R](dbName: Symbol, parallelism: Int = 1, processInOrder: Boolean = true,
  2.                                  statement: String, prepareParams: R => Seq[Any]) {
  3.     jas =>
  4.     def setDBName(db: Symbol): JDBCActionStream[R] = jas.copy(dbName = db)
  5.     def setParallelism(parLevel: Int): JDBCActionStream[R] = jas.copy(parallelism = parLevel)
  6.     def setProcessOrder(ordered: Boolean): JDBCActionStream[R] = jas.copy(processInOrder = ordered)
  8.     private def perform(r: R)(implicit ec: ExecutionContextExecutor) = {
  9.       import scala.concurrent._
  10.       val params = prepareParams(r)
  11.       Future {
  12.         NamedDB(dbName) autoCommit { session =>
  13.           session.execute(statement, params: _*)
  14.         }
  15.         r
  16.       }
  17.     }
  18.     def performOnRow(implicit ec: ExecutionContextExecutor): Flow[R, R, NotUsed] =
  19.       if (processInOrder)
  20.         Flow[R].mapAsync(parallelism)(perform)
  21.       else
  22.         Flow[R].mapAsyncUnordered(parallelism)(perform)
  24.   }
  25.   object JDBCActionStream {
  26.     def apply[R](_dbName: Symbol, _statement: String, params: R => Seq[Any]): JDBCActionStream[R] =
  27.       new JDBCActionStream[R](dbName = _dbName, statement=_statement, prepareParams = params)
  28.   }

函数performOnRow是个passthrough处理过程，使用了mapAsync来支持多线程运算。客户端调用方式如下：

  1.   def updateBatches: Source[JDBCDataRow,NotUsed] = {
  2.     logger.info(s"running updateBatches ...")
  3.       Source
  4.         .fromIterator(() => List(query,query2,query3).toIterator)
  5.         .via(stub.batQuery)
  6.         .via(stub.updateBat)
  8.   }

下面是本次示范的完整源代码：

jdbc.proto

  1. syntax = "proto3";
  3. import "google/protobuf/wrappers.proto";
  4. import "google/protobuf/any.proto";
  5. import "scalapb/scalapb.proto";
  8. package grpc.jdbc.services;
  10. option (scalapb.options) = {
  11.   // use a custom Scala package name
  12.   // package_name: "io.ontherocks.introgrpc.demo"
  14.   // don't append file name to package
  15.   flat_package: true
  17.   // generate one Scala file for all messages (services still get their own file)
  18.   single_file: true
  20.   // add imports to generated file
  21.   // useful when extending traits or using custom types
  22.   // import: "io.ontherocks.hellogrpc.RockingMessage"
  24.   // code to put at the top of generated file
  25.   // works only with `single_file: true`
  26.   //preamble: "sealed trait SomeSealedTrait"
  27. };
  29. /*
  30.  * Demoes various customization options provided by ScalaPBs.
  31.  */
  33. message JDBCDataRow {
  34.  string year = 1;
  35.  string state = 2;
  36.  string county = 3;
  37.  string value = 4;
  38. }
  41. message JDBCQuery {
  42.   string dbName = 1;
  43.   string statement = 2;
  44.   bytes parameters = 3;
  45.   google.protobuf.Int32Value fetchSize= 4;
  46.   google.protobuf.BoolValue autoCommit = 5;
  47.   google.protobuf.Int32Value queryTimeout = 6;
  48. }
  50. message JDBCResult {
  51.   bytes result = 1;
  52. }
  54. message JDBCUpdate {
  55.   string dbName = 1;
  56.   repeated string statements = 2;
  57.   bytes parameters = 3;
  58.   google.protobuf.Int32Value fetchSize= 4;
  59.   google.protobuf.Int32Value queryTimeout = 5;
  60.   int32 sqlType = 6;
  61.   google.protobuf.Int32Value batch = 7;
  62.   bytes returnGeneratedKey = 8;
  63. }
  65. message JDBCDate {
  66.   int32 yyyy = 1;
  67.   int32 mm   = 2;
  68.   int32 dd   = 3;
  69. }
  71. message JDBCTime {
  72.   int32 hh   = 1;
  73.   int32 mm   = 2;
  74.   int32 ss   = 3;
  75.   int32 nnn  = 4;
  76. }
  78. message JDBCDateTime {
  79.    JDBCDate date = 1;
  80.    JDBCTime time = 2;
  81. }
  83. message MemberRow {
  84.   string name = 1;
  85.   JDBCDate birthday = 2;
  86.   string description = 3;
  87.   JDBCDateTime created_at = 4;
  88.   bytes picture = 5;
  89. }
  91. service JDBCServices {
  92.   rpc runQuery(JDBCQuery) returns (stream JDBCDataRow) {}
  93.   rpc batQuery(stream JDBCQuery) returns (stream JDBCDataRow) {}
  94.   rpc runDDL(JDBCUpdate) returns (JDBCResult) {}
  95.   rpc insertRows(stream MemberRow) returns(JDBCResult) {}
  96.   rpc updateRows(stream MemberRow) returns(JDBCResult) {}
  97.   rpc getMembers(JDBCQuery) returns (stream MemberRow) {}
  98. }

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.jdbc.FileStreaming._
  14. import scalikejdbc.TxBoundary.Try._
  16. import scala.concurrent.ExecutionContextExecutor
  17. import java.io.InputStream
  19. object JDBCContext {
  20.   type SQLTYPE = Int
  21.   val SQL_EXEDDL= 1
  22.   val SQL_UPDATE = 2
  23.   val RETURN_GENERATED_KEYVALUE = true
  24.   val RETURN_UPDATED_COUNT = false
  26. }
  28. case class JDBCQueryContext[M](
  29.                                 dbName: Symbol,
  30.                                 statement: String,
  31.                                 parameters: Seq[Any] = Nil,
  32.                                 fetchSize: Int = 100,
  33.                                 autoCommit: Boolean = false,
  34.                                 queryTimeout: Option[Int] = None)
  37. case class JDBCContext(
  38.                         dbName: Symbol,
  39.                         statements: Seq[String] = Nil,
  40.                         parameters: Seq[Seq[Any]] = Nil,
  41.                         fetchSize: Int = 100,
  42.                         queryTimeout: Option[Int] = None,
  43.                         queryTags: Seq[String] = Nil,
  44.                         sqlType: JDBCContext.SQLTYPE = JDBCContext.SQL_UPDATE,
  45.                         batch: Boolean = false,
  46.                         returnGeneratedKey: Seq[Option[Any]] = Nil,
  47.                         // no return: None, return by index: Some(1), by name: Some("id")
  48.                         preAction: Option[PreparedStatement => Unit] = None,
  49.                         postAction: Option[PreparedStatement => Unit] = None) {
  51.   ctx =>
  53.   //helper functions
  55.   def appendTag(tag: String): JDBCContext = ctx.copy(queryTags = ctx.queryTags :+ tag)
  57.   def appendTags(tags: Seq[String]): JDBCContext = ctx.copy(queryTags = ctx.queryTags ++ tags)
  59.   def setFetchSize(size: Int): JDBCContext = ctx.copy(fetchSize = size)
  61.   def setQueryTimeout(time: Option[Int]): JDBCContext = ctx.copy(queryTimeout = time)
  63.   def setPreAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
  64.     if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
  65.       !ctx.batch && ctx.statements.size == 1)
  66.       ctx.copy(preAction = action)
  67.     else
  68.       throw new IllegalStateException("JDBCContex setting error: preAction not supported!")
  69.   }
  71.   def setPostAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
  72.     if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
  73.       !ctx.batch && ctx.statements.size == 1)
  74.       ctx.copy(postAction = action)
  75.     else
  76.       throw new IllegalStateException("JDBCContex setting error: preAction not supported!")
  77.   }
  79.   def appendDDLCommand(_statement: String, _parameters: Any*): JDBCContext = {
  80.     if (ctx.sqlType == JDBCContext.SQL_EXEDDL) {
  81.       ctx.copy(
  82.         statements = ctx.statements ++ Seq(_statement),
  83.         parameters = ctx.parameters ++ Seq(Seq(_parameters))
  84.       )
  85.     } else
  86.       throw new IllegalStateException("JDBCContex setting error: option not supported!")
  87.   }
  89.   def appendUpdateCommand(_returnGeneratedKey: Boolean, _statement: String,_parameters: Any*): JDBCContext = {
  90.     if (ctx.sqlType == JDBCContext.SQL_UPDATE && !ctx.batch) {
  91.       ctx.copy(
  92.         statements = ctx.statements ++ Seq(_statement),
  93.         parameters = ctx.parameters ++ Seq(_parameters),
  94.         returnGeneratedKey = ctx.returnGeneratedKey ++ (if (_returnGeneratedKey) Seq(Some(1)) else Seq(None))
  95.       )
  96.     } else
  97.       throw new IllegalStateException("JDBCContex setting error: option not supported!")
  98.   }
  100.   def appendBatchParameters(_parameters: Any*): JDBCContext = {
  101.     if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch)
  102.       throw new IllegalStateException("JDBCContex setting error: batch parameters only supported for SQL_UPDATE and batch = true!")
  104.     var matchParams = true
  105.     if (ctx.parameters != Nil)
  106.       if (ctx.parameters.head.size != _parameters.size)
  107.         matchParams = false
  108.     if (matchParams) {
  109.       ctx.copy(
  110.         parameters = ctx.parameters ++ Seq(_parameters)
  111.       )
  112.     } else
  113.       throw new IllegalStateException("JDBCContex setting error: batch command parameters not match!")
  114.   }
  117.   def setBatchReturnGeneratedKeyOption(returnKey: Boolean): JDBCContext = {
  118.     if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch)
  119.       throw new IllegalStateException("JDBCContex setting error: only supported in batch update commands!")
  120.     ctx.copy(
  121.       returnGeneratedKey = if (returnKey) Seq(Some(1)) else Nil
  122.     )
  123.   }
  125.   def setDDLCommand(_statement: String, _parameters: Any*): JDBCContext = {
  126.     ctx.copy(
  127.       statements = Seq(_statement),
  128.       parameters = Seq(_parameters),
  129.       sqlType = JDBCContext.SQL_EXEDDL,
  130.       batch = false
  131.     )
  132.   }
  134.   def setUpdateCommand(_returnGeneratedKey: Boolean, _statement: String,_parameters: Any*): JDBCContext = {
  135.     ctx.copy(
  136.       statements = Seq(_statement),
  137.       parameters = Seq(_parameters),
  138.       returnGeneratedKey = if (_returnGeneratedKey) Seq(Some(1)) else Seq(None),
  139.       sqlType = JDBCContext.SQL_UPDATE,
  140.       batch = false
  141.     )
  142.   }
  143.   def setBatchCommand(_statement: String): JDBCContext = {
  144.     ctx.copy (
  145.       statements = Seq(_statement),
  146.       sqlType = JDBCContext.SQL_UPDATE,
  147.       batch = true
  148.     )
  149.   }
  151. }
  153. object JDBCEngine {
  155.   import JDBCContext._
  157.   type JDBCDate = LocalDate
  158.   type JDBCDateTime = LocalDateTime
  159.   type JDBCTime = LocalTime
  161.   def jdbcSetDate(yyyy: Int, mm: Int, dd: Int) = LocalDate.of(yyyy,mm,dd)
  162.   def jdbcSetTime(hh: Int, mm: Int, ss: Int, nn: Int) = LocalTime.of(hh,mm,ss,nn)
  163.   def jdbcSetDateTime(date: JDBCDate, time: JDBCTime) =  LocalDateTime.of(date,time)
  164.   def jdbcSetNow = LocalDateTime.now()
  167.   type JDBCBlob = InputStream
  169.   def fileToJDBCBlob(fileName: String, timeOut: FiniteDuration = 60 seconds)(
  170.     implicit mat: Materializer) = FileToInputStream(fileName,timeOut)
  172.   def jdbcBlobToFile(blob: JDBCBlob, fileName: String)(
  173.     implicit mat: Materializer) =  InputStreamToFile(blob,fileName)
  177.   private def noExtractor(message: String): WrappedResultSet => Nothing = { (rs: WrappedResultSet) =>
  178.     throw new IllegalStateException(message)
  179.   }
  181.   def jdbcAkkaStream[A](ctx: JDBCQueryContext[A],extractor: WrappedResultSet => A)
  182.                        (implicit ec: ExecutionContextExecutor): Source[A,NotUsed] = {
  183.       val publisher: DatabasePublisher[A] = NamedDB(ctx.dbName) readOnlyStream {
  184.       val rawSql = new SQLToCollectionImpl[A, NoExtractor](ctx.statement, ctx.parameters)(noExtractor(""))
  185.       ctx.queryTimeout.foreach(rawSql.queryTimeout(_))
  186.       val sql: SQL[A, HasExtractor] = rawSql.map(extractor)
  188.       sql.iterator
  189.         .withDBSessionForceAdjuster(session => {
  190.           session.connection.setAutoCommit(ctx.autoCommit)
  191.           session.fetchSize(ctx.fetchSize)
  192.         })
  193.     }
  194.     Source.fromPublisher[A](publisher)
  195.   }
  198.   def jdbcQueryResult[C[_] <: TraversableOnce[_], A](ctx: JDBCQueryContext[A],
  199.                                                      extractor: WrappedResultSet => A)(
  200.                                                       implicit cbf: CanBuildFrom[Nothing, A, C[A]]): C[A] = {
  202.     val rawSql = new SQLToCollectionImpl[A, NoExtractor](ctx.statement, ctx.parameters)(noExtractor(""))
  203.     ctx.queryTimeout.foreach(rawSql.queryTimeout(_))
  204.     rawSql.fetchSize(ctx.fetchSize)
  205.     implicit val session = NamedAutoSession(ctx.dbName)
  206.     val sql: SQL[A, HasExtractor] = rawSql.map(extractor)
  207.     sql.collection.apply[C]()
  209.   }
  211.   def jdbcExecuteDDL(ctx: JDBCContext)(implicit ec: ExecutionContextExecutor): Future[String] = {
  212.     if (ctx.sqlType != SQL_EXEDDL) {
  213.       Future.failed(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_EXEDDL'!"))
  214.     }
  215.     else {
  216.       Future {
  217.         NamedDB(ctx.dbName) localTx { implicit session =>
  218.           ctx.statements.foreach { stm =>
  219.             val ddl = new SQLExecution(statement = stm, parameters = Nil)(
  220.               before = WrappedResultSet => {})(
  221.               after = WrappedResultSet => {})
  223.             ddl.apply()
  224.           }
  225.           "SQL_EXEDDL executed succesfully."
  226.         }
  227.       }
  228.     }
  229.   }
  231.   def jdbcBatchUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  232.     implicit ec: ExecutionContextExecutor,
  233.              cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  234.     if (ctx.statements == Nil)
  235.      Future.failed ( new IllegalStateException("JDBCContex setting error: statements empty!"))
  236.     if (ctx.sqlType != SQL_UPDATE) {
  237.       Future.failed(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!"))
  238.     }
  239.     else {
  240.       if (ctx.batch) {
  241.         if (noReturnKey(ctx)) {
  242.           val usql = SQL(ctx.statements.head)
  243.             .tags(ctx.queryTags: _*)
  244.             .batch(ctx.parameters: _*)
  245.           Future {
  246.             NamedDB(ctx.dbName) localTx { implicit session =>
  247.               ctx.queryTimeout.foreach(session.queryTimeout(_))
  248.               usql.apply[Seq]()
  249.               Seq.empty[Long].to[C]
  250.             }
  251.           }
  252.         } else {
  253.           val usql = new SQLBatchWithGeneratedKey(ctx.statements.head, ctx.parameters, ctx.queryTags)(None)
  254.           Future {
  255.             NamedDB(ctx.dbName) localTx { implicit session =>
  256.               ctx.queryTimeout.foreach(session.queryTimeout(_))
  257.               usql.apply[C]()
  258.             }
  259.           }
  260.         }
  262.       } else {
  263.         Future.failed(new IllegalStateException("JDBCContex setting error: must set batch = true !"))
  264.       }
  265.     }
  266.   }
  267.   private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  268.           implicit ec: ExecutionContextExecutor,
  269.                     cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  270.     val Some(key) :: xs = ctx.returnGeneratedKey
  271.     val params: Seq[Any] = ctx.parameters match {
  272.       case Nil => Nil
  273.       case p@_ => p.head
  274.     }
  275.     val usql = new SQLUpdateWithGeneratedKey(ctx.statements.head, params, ctx.queryTags)(key)
  276.     Future {
  277.       NamedDB(ctx.dbName) localTx { implicit session =>
  278.         session.fetchSize(ctx.fetchSize)
  279.         ctx.queryTimeout.foreach(session.queryTimeout(_))
  280.         val result = usql.apply()
  281.         Seq(result).to[C]
  282.       }
  283.     }
  284.   }
  286.   private def singleTxUpdateNoReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  287.           implicit ec: ExecutionContextExecutor,
  288.                    cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  289.     val params: Seq[Any] = ctx.parameters match {
  290.       case Nil => Nil
  291.       case p@_ => p.head
  292.     }
  293.     val before = ctx.preAction match {
  294.       case None => pstm: PreparedStatement => {}
  295.       case Some(f) => f
  296.     }
  297.     val after = ctx.postAction match {
  298.       case None => pstm: PreparedStatement => {}
  299.       case Some(f) => f
  300.     }
  301.     val usql = new SQLUpdate(ctx.statements.head,params,ctx.queryTags)(before)(after)
  302.     Future {
  303.       NamedDB(ctx.dbName) localTx {implicit session =>
  304.         session.fetchSize(ctx.fetchSize)
  305.         ctx.queryTimeout.foreach(session.queryTimeout(_))
  306.         val result = usql.apply()
  307.         Seq(result.toLong).to[C]
  308.       }
  309.     }
  311.   }
  313.   private def singleTxUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  314.     implicit ec: ExecutionContextExecutor,
  315.              cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  316.     if (noReturnKey(ctx))
  317.       singleTxUpdateNoReturnKey(ctx)
  318.     else
  319.       singleTxUpdateWithReturnKey(ctx)
  320.   }
  322.   private def noReturnKey(ctx: JDBCContext): Boolean = {
  323.     if (ctx.returnGeneratedKey != Nil) {
  324.       val k :: xs = ctx.returnGeneratedKey
  325.       k match {
  326.         case None => true
  327.         case Some(k) => false
  328.       }
  329.     } else true
  330.   }
  332.   def noActon: PreparedStatement=>Unit = pstm => {}
  334.   def multiTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  335.     implicit ec: ExecutionContextExecutor,
  336.              cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  337.     Future {
  338.       NamedDB(ctx.dbName) localTx { implicit session =>
  339.         session.fetchSize(ctx.fetchSize)
  340.         ctx.queryTimeout.foreach(session.queryTimeout(_))
  341.         val keys: Seq[Option[Any]] = ctx.returnGeneratedKey match {
  342.           case Nil => Seq.fill(ctx.statements.size)(None)
  343.           case k@_ => k
  344.         }
  345.         val sqlcmd = ctx.statements zip ctx.parameters zip keys
  346.         val results = sqlcmd.map { case ((stm, param), key) =>
  347.           key match {
  348.             case None =>
  349.               new SQLUpdate(stm, param, Nil)(noActon)(noActon).apply().toLong
  350.             case Some(k) =>
  351.               new SQLUpdateWithGeneratedKey(stm, param, Nil)(k).apply().toLong
  352.           }
  353.         }
  354.         results.to[C]
  355.       }
  356.     }
  357.   }
  360.   def jdbcTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  361.     implicit ec: ExecutionContextExecutor,
  362.              cbf: CanBuildFrom[Nothing, Long, C[Long]]): Future[C[Long]] = {
  363.     if (ctx.statements == Nil)
  364.       Future.failed( new IllegalStateException("JDBCContex setting error: statements empty!"))
  365.     if (ctx.sqlType != SQL_UPDATE) {
  366.       Future.failed(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!"))
  367.     }
  368.     else {
  369.       if (!ctx.batch) {
  370.         if (ctx.statements.size == 1)
  371.           singleTxUpdate(ctx)
  372.         else
  373.           multiTxUpdates(ctx)
  374.       } else
  375.         Future.failed(new IllegalStateException("JDBCContex setting error: must set batch = false !"))
  377.     }
  378.   }
  380.   case class JDBCActionStream[R](dbName: Symbol, parallelism: Int = 1, processInOrder: Boolean = true,
  381.                                  statement: String, prepareParams: R => Seq[Any]) {
  382.     jas =>
  383.     def setDBName(db: Symbol): JDBCActionStream[R] = jas.copy(dbName = db)
  384.     def setParallelism(parLevel: Int): JDBCActionStream[R] = jas.copy(parallelism = parLevel)
  385.     def setProcessOrder(ordered: Boolean): JDBCActionStream[R] = jas.copy(processInOrder = ordered)
  387.     private def perform(r: R)(implicit ec: ExecutionContextExecutor) = {
  388.       import scala.concurrent._
  389.       val params = prepareParams(r)
  390.       Future {
  391.         NamedDB(dbName) autoCommit { session =>
  392.           session.execute(statement, params: _*)
  393.         }
  394.         r
  395.       }
  396.      // Future.successful(r)
  397.     }
  398.     def performOnRow(implicit ec: ExecutionContextExecutor): Flow[R, R, NotUsed] =
  399.       if (processInOrder)
  400.         Flow[R].mapAsync(parallelism)(perform)
  401.       else
  402.         Flow[R].mapAsyncUnordered(parallelism)(perform)
  404.   }
  405.   object JDBCActionStream {
  406.     def apply[R](_dbName: Symbol, _statement: String, params: R => Seq[Any]): JDBCActionStream[R] =
  407.       new JDBCActionStream[R](dbName = _dbName, statement=_statement, prepareParams = params)
  408.   }
  409. }

JDBCService.scala

  1. package demo.grpc.jdbc.services
  3. import akka.NotUsed
  4. import akka.stream.scaladsl.{Source,Flow}
  5. import grpc.jdbc.services._
  6. import java.util.logging.Logger
  7. import protobuf.bytes.Converter._
  8. import sdp.jdbc.engine._
  9. import JDBCEngine._
  10. import scalikejdbc.WrappedResultSet
  11. import scala.concurrent.ExecutionContextExecutor
  13. class JDBCStreamingServices(implicit ec: ExecutionContextExecutor) extends JdbcGrpcAkkaStream.JDBCServices {
  14.   val logger = Logger.getLogger(classOf[JDBCStreamingServices].getName)
  16.   val toRow = (rs: WrappedResultSet) => JDBCDataRow(
  17.     year = rs.string("REPORTYEAR"),
  18.     state = rs.string("STATENAME"),
  19.     county = rs.string("COUNTYNAME"),
  20.     value = rs.string("VALUE")
  21.   )
  22.   override def runQuery: Flow[JDBCQuery, JDBCDataRow, NotUsed] = {
  23.     logger.info("**** runQuery called on service side ***")
  24.     Flow[JDBCQuery]
  25.       .flatMapConcat { q =>
  26.         //unpack JDBCQuery and construct the context
  27.         var params: Seq[Any] =  Nil
  28.         if (q.parameters != _root_.com.google.protobuf.ByteString.EMPTY)
  29.           params = unmarshal[Seq[Any]](q.parameters)
  30.         logger.info(s"**** query parameters: ${params} ****")
  31.         val ctx = JDBCQueryContext[JDBCDataRow](
  32.           dbName = Symbol(q.dbName),
  33.           statement = q.statement,
  34.           parameters = params,
  35.           fetchSize = q.fetchSize.getOrElse(100),
  36.           autoCommit = q.autoCommit.getOrElse(false),
  37.           queryTimeout = q.queryTimeout
  38.         )
  39.         jdbcAkkaStream(ctx, toRow)
  40.       }
  41.   }
  43.   override def batQuery: Flow[JDBCQuery, JDBCDataRow, NotUsed] = runQuery
  45.   override def runDDL: Flow[JDBCUpdate, JDBCResult, NotUsed] = {
  46.     logger.info("**** runDDL called on service side ***")
  47.     Flow[JDBCUpdate]
  48.       .flatMapConcat { context =>
  49.         //unpack JDBCUpdate and construct the context
  51.         val ctx = JDBCContext(
  52.           dbName = Symbol(context.dbName),
  53.           statements = context.statements,
  54.           sqlType = JDBCContext.SQL_EXEDDL,
  55.           queryTimeout = context.queryTimeout
  56.         )
  58.         logger.info(s"**** JDBCContext => ${ctx} ***")
  60.         Source
  61.           .fromFuture(jdbcExecuteDDL(ctx))
  62.           .map { r => JDBCResult(marshal(r)) }
  64.       }
  66.   }
  68.   override def insertRows: Flow[MemberRow, JDBCResult, NotUsed] = {
  69.     logger.info("**** insertRows called on service side ***")
  70.     val insertSQL = """
  71.       insert into members(
  72.         name,
  73.         birthday,
  74.         description,
  75.         created_at
  76.       ) values (?, ?, ?, ?)
  77.     """
  78.     Flow[MemberRow]
  79.       .flatMapConcat { row =>
  80.         val ctx = JDBCContext('h2)
  81.           .setUpdateCommand(true,insertSQL,
  82.              row.name,
  83.              jdbcSetDate(row.birthday.get.yyyy,row.birthday.get.mm,row.birthday.get.dd),
  84.              row.description,
  85.              jdbcSetNow
  86.           )
  88.         logger.info(s"**** JDBCContext => ${ctx} ***")
  90.         Source
  91.           .fromFuture(jdbcTxUpdates[Vector](ctx))
  92.           .map { r => JDBCResult(marshal(r)) }
  93.       }
  94.   }
  96.   override def updateRows: Flow[MemberRow, JDBCResult, NotUsed] = {
  97.     logger.info("**** updateRows called on service side ***")
  98.     val updateSQL = "update members set description = ?, created_at = ? where name = ?"
  100.     Flow[MemberRow]
  101.       .flatMapConcat { row =>
  102.         val ctx = JDBCContext('h2)
  103.             .setBatchCommand(updateSQL)
  104.             .appendBatchParameters(
  105.               row.name + " updated.",
  106.               jdbcSetNow,
  107.               row.name
  108.             ).setBatchReturnGeneratedKeyOption(true)
  110.         logger.info(s"**** JDBCContext => ${ctx} ***")
  112.         Source
  113.           .fromFuture(jdbcBatchUpdate[Vector](ctx))
  114.           .map { r => JDBCResult(marshal(r)) }
  115.       }
  116.   }
  119.   val toMemberRow = (rs: WrappedResultSet) => MemberRow(
  120.     name = rs.string("name"),
  121.     description = rs.string("description"),
  122.     birthday = None,
  123.     createdAt = None,
  124.     picture = _root_.com.google.protobuf.ByteString.EMPTY
  125.   )
  126.   override def getMembers: Flow[JDBCQuery, MemberRow, NotUsed] = {
  127.     logger.info("**** getMembers called on service side ***")
  128.     Flow[JDBCQuery]
  129.       .flatMapConcat { q =>
  130.         //unpack JDBCQuery and construct the context
  131.         var params: Seq[Any] =  Nil
  132.         if (q.parameters != _root_.com.google.protobuf.ByteString.EMPTY)
  133.            params = unmarshal[Seq[Any]](q.parameters)
  134.         logger.info(s"**** query parameters: ${params} ****")
  135.         val ctx = JDBCQueryContext[MemberRow](
  136.           dbName = Symbol(q.dbName),
  137.           statement = q.statement,
  138.           parameters = params,
  139.           fetchSize = q.fetchSize.getOrElse(100),
  140.           autoCommit = q.autoCommit.getOrElse(false),
  141.           queryTimeout = q.queryTimeout
  142.         )
  143.         jdbcAkkaStream(ctx, toMemberRow)
  144.       }
  146.   }
  147. }

JDBCServer.scala

  1. package demo.grpc.jdbc.server
  3. import java.util.logging.Logger
  5. import akka.actor.ActorSystem
  6. import akka.stream.ActorMaterializer
  7. import io.grpc.Server
  8. import demo.grpc.jdbc.services._
  9. import io.grpc.ServerBuilder
  10. import grpc.jdbc.services._
  12. class gRPCServer(server: Server) {
  14.   val logger: Logger = Logger.getLogger(classOf[gRPCServer].getName)
  16.   def start(): Unit = {
  17.     server.start()
  18.     logger.info(s"Server started, listening on ${server.getPort}")
  19.     sys.addShutdownHook {
  20.       // Use stderr here since the logger may has been reset by its JVM shutdown hook.
  21.       System.err.println("*** shutting down gRPC server since JVM is shutting down")
  22.       stop()
  23.       System.err.println("*** server shut down")
  24.     }
  25.     ()
  26.   }
  28.   def stop(): Unit = {
  29.     server.shutdown()
  30.   }
  32.   /**
  33.     * Await termination on the main thread since the grpc library uses daemon threads.
  34.     */
  35.   def blockUntilShutdown(): Unit = {
  36.     server.awaitTermination()
  37.   }
  38. }
  40. object JDBCServer extends App {
  41.   import sdp.jdbc.config._
  43.   implicit val system = ActorSystem("JDBCServer")
  44.   implicit val mat = ActorMaterializer.create(system)
  45.   implicit val ec = system.dispatcher
  47.   ConfigDBsWithEnv("dev").setup('h2)
  48.   ConfigDBsWithEnv("dev").loadGlobalSettings()
  50.   val server = new gRPCServer(
  51.     ServerBuilder
  52.       .forPort(50051)
  53.       .addService(
  54.         JdbcGrpcAkkaStream.bindService(
  55.           new JDBCStreamingServices
  56.         )
  57.       ).build()
  58.   )
  59.   server.start()
  60.   //  server.blockUntilShutdown()
  61.   scala.io.StdIn.readLine()
  62.   ConfigDBsWithEnv("dev").close('h2)
  63.   mat.shutdown()
  64.   system.terminate()
  65. }

JDBCClient.scala

  1. package demo.grpc.jdbc.client
  2. import grpc.jdbc.services._
  3. import java.util.logging.Logger
  5. import protobuf.bytes.Converter._
  6. import akka.stream.scaladsl._
  7. import akka.NotUsed
  8. import akka.actor.ActorSystem
  9. import akka.stream.{ActorMaterializer, ThrottleMode}
  10. import io.grpc._
  11. import sdp.jdbc.engine._
  13. class JDBCStreamClient(host: String, port: Int) {
  14.   val logger: Logger = Logger.getLogger(classOf[JDBCStreamClient].getName)
  16.   val channel = ManagedChannelBuilder
  17.     .forAddress(host,port)
  18.     .usePlaintext(true)
  19.     .build()
  22.   val stub = JdbcGrpcAkkaStream.stub(channel)
  23.   val query = JDBCQuery (
  24.     dbName = "h2",
  25.     statement = "select * from AQMRPT where STATENAME = ? and VALUE = ?",
  26.     parameters = marshal(Seq("Arizona", 2))
  27.   )
  28.   val query2 = JDBCQuery (
  29.     dbName = "h2",
  30.     statement = "select * from AQMRPT where STATENAME = ? and VALUE = ?",
  31.     parameters = marshal(Seq("Colorado", 3))
  32.   )
  33.   val query3= JDBCQuery (
  34.     dbName = "h2",
  35.     statement = "select * from AQMRPT where STATENAME = ? and VALUE = ?",
  36.     parameters = marshal(Seq("Arkansas", 8))
  37.   )
  39.   def queryRows: Source[JDBCDataRow,NotUsed] = {
  40.     logger.info(s"running queryRows ...")
  41.     Source
  42.         .single(query)
  43.         .via(stub.runQuery)
  44.   }
  46.   def batQueryRows: Source[JDBCDataRow,NotUsed] = {
  47.     logger.info(s"running batQueryRows ...")
  48.     Source
  49.       .fromIterator(() => List(query,query2,query3).toIterator)
  50.       .via(stub.batQuery)
  51.   }
  53.   val dropSQL: String ="""
  54.       drop table members
  55.     """
  57.   val createSQL: String ="""
  58.     create table members (
  59.       id serial not null primary key,
  60.       name varchar(30) not null,
  61.       description varchar(1000),
  62.       birthday date,
  63.       created_at timestamp not null,
  64.       picture blob
  65.     )"""
  66.   val ctx = JDBCUpdate (
  67.     dbName = "h2",
  68.     sqlType = JDBCContext.SQL_EXEDDL,
  69.     statements = Seq(dropSQL,createSQL)
  70.   )
  72.   def createTbl: Source[JDBCResult,NotUsed] = {
  73.     logger.info(s"running createTbl ...")
  74.     Source
  75.       .single(ctx)
  76.       .via(stub.runDDL)
  77.   }
  79.   val p1 = MemberRow( "Peter Chan",Some(JDBCDate(1967,5,17)),"new member1",None,_root_.com.google.protobuf.ByteString.EMPTY)
  80.   val p2 = MemberRow( "Alanda Wong",Some(JDBCDate(1980,11,10)),"new member2",None,_root_.com.google.protobuf.ByteString.EMPTY)
  81.   val p3 = MemberRow( "Kate Zhang",Some(JDBCDate(1969,8,13)),"new member3",None,_root_.com.google.protobuf.ByteString.EMPTY)
  82.   val p4 = MemberRow( "Tiger Chan",Some(JDBCDate(1962,5,1)),"new member4",None,_root_.com.google.protobuf.ByteString.EMPTY)
  84.   def insertRows: Source[JDBCResult,NotUsed] = {
  85.     logger.info(s"running insertRows ...")
  86.     Source
  87.       .fromIterator(() => List(p1,p2,p3,p4).toIterator)
  88.       .via(stub.insertRows)
  89.   }
  91.   val queryMember = JDBCQuery (
  92.     dbName = "h2",
  93.     statement = "select * from members"
  94.   )
  95.   def updateRows: Source[JDBCResult,NotUsed] = {
  96.     logger.info(s"running updateRows ...")
  97.     Source
  98.       .single(queryMember)
  99.       .via(stub.getMembers)
  100.       .via(stub.updateRows)
  101.   }
  102.   def updateBatches: Source[JDBCDataRow,NotUsed] = {
  103.     logger.info(s"running updateBatches ...")
  104.       Source
  105.         .fromIterator(() => List(query,query2,query3).toIterator)
  106.         .via(stub.batQuery)
  107.         .via(stub.updateBat)
  109.   }
  111. }
  114. object TestConversion extends App {
  116.   val orgval: Seq[Option[Any]] = Seq(Some(1),Some("id"),None,Some(2))
  117.   println(s"original value: ${orgval}")
  119.   val marval = marshal(orgval)
  121.   println(s"marshal value: ${marval}")
  123.   val unmval = unmarshal[Seq[Option[Any]]](marval)
  125.   println(s"marshal value: ${unmval}")
  128.   val m1 = MemberRow(name = "Peter")
  129.   val m2 = m1.update(
  130.     _.birthday.yyyy := 1989,
  131.     _.birthday.mm := 10,
  132.     _.birthday.dd := 3,
  133.     _.description := "a new member"
  134.   )
  137. }
  140. object QueryRows extends App {
  141.   implicit val system = ActorSystem("QueryRows")
  142.   implicit val mat = ActorMaterializer.create(system)
  144.   val client = new JDBCStreamClient("localhost", 50051)
  146.   client.queryRows.runForeach { r => println(r) }
  149.   scala.io.StdIn.readLine()
  150.   mat.shutdown()
  151.   system.terminate()
  152. }
  154. object BatQueryRows extends App {
  155.   implicit val system = ActorSystem("BatQueryRows")
  156.   implicit val mat = ActorMaterializer.create(system)
  158.   val client = new JDBCStreamClient("localhost", 50051)
  160.   client.batQueryRows.runForeach(println)
  162.   scala.io.StdIn.readLine()
  163.   mat.shutdown()
  164.   system.terminate()
  165. }
  167. object RunDDL extends App {
  168.   implicit val system = ActorSystem("RunDDL")
  169.   implicit val mat = ActorMaterializer.create(system)
  171.   val client = new JDBCStreamClient("localhost", 50051)
  173.   client.createTbl.runForeach{r => println(unmarshal[Seq[Any]](r.result))}
  175.   scala.io.StdIn.readLine()
  176.   mat.shutdown()
  177.   system.terminate()
  179. }
  181. object InsertRows extends App {
  182.   implicit val system = ActorSystem("InsertRows")
  183.   implicit val mat = ActorMaterializer.create(system)
  185.   val client = new JDBCStreamClient("localhost", 50051)
  187.   client.insertRows.runForeach { r => println(unmarshal[Vector[Long]](r.result)) }
  189.   scala.io.StdIn.readLine()
  190.   mat.shutdown()
  191.   system.terminate()
  192. }
  194. object UpdateRows extends App {
  195.   implicit val system = ActorSystem("UpdateRows")
  196.   implicit val mat = ActorMaterializer.create(system)
  198.   val client = new JDBCStreamClient("localhost", 50051)
  200.   client.updateRows.runForeach{ r => println(unmarshal[Vector[Long]](r.result)) }
  202.   scala.io.StdIn.readLine()
  203.   mat.shutdown()
  204.   system.terminate()
  205. }
  207. object BatUpdates extends App {
  208.   implicit val system = ActorSystem("BatUpdates")
  209.   implicit val mat = ActorMaterializer.create(system)
  211.   val client = new JDBCStreamClient("localhost", 50051)
  213.   client.updateBatches.runForeach(println)
  215.   scala.io.StdIn.readLine()
  216.   mat.shutdown()
  217.   system.terminate()
  218. }

ByteConverter.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. }

其它部分的源代码和系统设置可以从上次的讨论稿中获取。

 

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

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号