PICE（1）：Programming In Clustered Environment – 集群环境内编程模式详解编程语言

        val ctx = JDBCQueryContext[JDBCDataRow]( 
          dbName = Symbol(q.dbName), 
          statement = q.statement, 
          parameters = params, 
          fetchSize = q.fetchSize.getOrElse(100), 
          autoCommit = q.autoCommit.getOrElse(false), 
          queryTimeout = q.queryTimeout 
        ) 
        jdbcAkkaStream(ctx, toRow)

message JDBCDataRow { 
 string year = 1; 
 string state = 2; 
 string county = 3; 
 string value = 4; 
} 
 
message JDBCQuery { 
  string dbName = 1; 
  string statement = 2; 
  bytes parameters = 3; 
  google.protobuf.Int32Value fetchSize= 4; 
  google.protobuf.BoolValue autoCommit = 5; 
  google.protobuf.Int32Value queryTimeout = 6; 
} 
 
service JDBCServices { 
  rpc runQuery(JDBCQuery) returns (stream JDBCDataRow) {} 
}

    parameters: _root_.com.google.protobuf.ByteString = _root_.com.google.protobuf.ByteString.EMPTY,

package protobuf.bytes 
import java.io.{ByteArrayInputStream,ByteArrayOutputStream,ObjectInputStream,ObjectOutputStream} 
import com.google.protobuf.ByteString 
object Converter { 
  def marshal(value: Any): ByteString = { 
    val stream: ByteArrayOutputStream = new ByteArrayOutputStream() 
    val oos = new ObjectOutputStream(stream) 
    oos.writeObject(value) 
    oos.close() 
    ByteString.copyFrom(stream.toByteArray()) 
  } 
 
  def unmarshal[A](bytes: ByteString): A = { 
    val ois = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray)) 
    val value = ois.readObject() 
    ois.close() 
    value.asInstanceOf[A] 
  } 
}

class JDBCStreamingServices(implicit ec: ExecutionContextExecutor) extends JdbcGrpcAkkaStream.JDBCServices { 
  val logger = Logger.getLogger(classOf[JDBCStreamingServices].getName) 
 
  val toRow = (rs: WrappedResultSet) => JDBCDataRow( 
    year = rs.string("REPORTYEAR"), 
    state = rs.string("STATENAME"), 
    county = rs.string("COUNTYNAME"), 
    value = rs.string("VALUE") 
  ) 
  override def runQuery: Flow[JDBCQuery, JDBCDataRow, NotUsed] = { 
    logger.info("**** runQuery called on service side ***") 
    Flow[JDBCQuery] 
      .flatMapConcat { q => 
        //unpack JDBCQuery and construct the context 
        val params: Seq[Any] = unmarshal[Seq[Any]](q.parameters) 
        logger.info(s"**** query parameters: ${params} ****") 
        val ctx = JDBCQueryContext[JDBCDataRow]( 
          dbName = Symbol(q.dbName), 
          statement = q.statement, 
          parameters = params, 
          fetchSize = q.fetchSize.getOrElse(100), 
          autoCommit = q.autoCommit.getOrElse(false), 
          queryTimeout = q.queryTimeout 
        ) 
        jdbcAkkaStream(ctx, toRow) 
      } 
  } 
}

  val query = JDBCQuery ( 
    dbName = "h2", 
    statement = "select * from AQMRPT where STATENAME = ? and VALUE = ?", 
    parameters = marshal(Seq("Arizona", 5)) 
  )

  def queryRows: Source[JDBCDataRow,NotUsed] = { 
    logger.info(s"running queryRows ...") 
    Source 
      .single(query) 
      .via(stub.runQuery) 
  }

object QueryRows extends App { 
  implicit val system = ActorSystem("QueryRows") 
  implicit val mat = ActorMaterializer.create(system) 
 
  val client = new JDBCStreamClient("localhost", 50051) 
 
  client.queryRows.runForeach(println) 
 
  scala.io.StdIn.readLine() 
  mat.shutdown() 
  system.terminate() 
}

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

import scalapb.compiler.Version.scalapbVersion 
import scalapb.compiler.Version.grpcJavaVersion 
name := "gRPCJDBCStreaming" 
version := "0.11" 
scalaVersion := "2.12.6" 
resolvers += Resolver.bintrayRepo("beyondthelines", "maven") 
libraryDependencies ++= Seq( 
"com.thesamet.scalapb" %% "scalapb-runtime" % scalapbVersion % "protobuf", 
"io.grpc" % "grpc-netty" % grpcJavaVersion, 
"com.thesamet.scalapb" %% "scalapb-runtime-grpc" % scalapbVersion, 
"io.monix" %% "monix" % "2.3.0", 
// for GRPC Akkastream 
"beyondthelines"         %% "grpcakkastreamruntime" % "0.0.5", 
// for scalikejdbc 
"org.scalikejdbc" %% "scalikejdbc"       % "3.2.1", 
"org.scalikejdbc" %% "scalikejdbc-test"   % "3.2.1"   % "test", 
"org.scalikejdbc" %% "scalikejdbc-config"  % "3.2.1", 
"org.scalikejdbc" %% "scalikejdbc-streams" % "3.2.1", 
"org.scalikejdbc" %% "scalikejdbc-joda-time" % "3.2.1", 
"com.h2database"  %  "h2"                % "1.4.196", 
"mysql" % "mysql-connector-java" % "6.0.6", 
"org.postgresql" % "postgresql" % "42.2.0", 
"commons-dbcp" % "commons-dbcp" % "1.4", 
"org.apache.tomcat" % "tomcat-jdbc" % "9.0.2", 
"com.zaxxer" % "HikariCP" % "2.7.4", 
"com.jolbox" % "bonecp" % "0.8.0.RELEASE", 
"com.typesafe.slick" %% "slick" % "3.2.1", 
"ch.qos.logback"  %  "logback-classic"   % "1.2.3", 
"com.typesafe.akka" %% "akka-actor" % "2.5.4", 
"com.typesafe.akka" %% "akka-stream" % "2.5.4" 
) 
PB.targets in Compile := Seq( 
scalapb.gen() -> (sourceManaged in Compile).value, 
// generate the akka stream files 
grpc.akkastreams.generators.GrpcAkkaStreamGenerator() -> (sourceManaged in Compile).value 
)

syntax = "proto3"; 
import "google/protobuf/wrappers.proto"; 
import "google/protobuf/any.proto"; 
import "scalapb/scalapb.proto"; 
package grpc.jdbc.services; 
option (scalapb.options) = { 
// use a custom Scala package name 
// package_name: "io.ontherocks.introgrpc.demo" 
// don't append file name to package 
flat_package: true 
// generate one Scala file for all messages (services still get their own file) 
single_file: true 
// add imports to generated file 
// useful when extending traits or using custom types 
// import: "io.ontherocks.hellogrpc.RockingMessage" 
// code to put at the top of generated file 
// works only with `single_file: true` 
//preamble: "sealed trait SomeSealedTrait" 
}; 
/* 
* Demoes various customization options provided by ScalaPBs. 
*/ 
message JDBCDataRow { 
string year = 1; 
string state = 2; 
string county = 3; 
string value = 4; 
} 
message JDBCQuery { 
string dbName = 1; 
string statement = 2; 
bytes parameters = 3; 
google.protobuf.Int32Value fetchSize= 4; 
google.protobuf.BoolValue autoCommit = 5; 
google.protobuf.Int32Value queryTimeout = 6; 
} 
service JDBCServices { 
rpc runQuery(JDBCQuery) returns (stream JDBCDataRow) {} 
}

# JDBC settings 
test { 
db { 
h2 { 
driver = "org.h2.Driver" 
url = "jdbc:h2:tcp://localhost/~/slickdemo" 
user = "" 
password = "" 
poolInitialSize = 5 
poolMaxSize = 7 
poolConnectionTimeoutMillis = 1000 
poolValidationQuery = "select 1 as one" 
poolFactoryName = "commons-dbcp2" 
} 
} 
db.mysql.driver = "com.mysql.cj.jdbc.Driver" 
db.mysql.url = "jdbc:mysql://localhost:3306/testdb" 
db.mysql.user = "root" 
db.mysql.password = "123" 
db.mysql.poolInitialSize = 5 
db.mysql.poolMaxSize = 7 
db.mysql.poolConnectionTimeoutMillis = 1000 
db.mysql.poolValidationQuery = "select 1 as one" 
db.mysql.poolFactoryName = "bonecp" 
# scallikejdbc Global settings 
scalikejdbc.global.loggingSQLAndTime.enabled = true 
scalikejdbc.global.loggingSQLAndTime.logLevel = info 
scalikejdbc.global.loggingSQLAndTime.warningEnabled = true 
scalikejdbc.global.loggingSQLAndTime.warningThresholdMillis = 1000 
scalikejdbc.global.loggingSQLAndTime.warningLogLevel = warn 
scalikejdbc.global.loggingSQLAndTime.singleLineMode = false 
scalikejdbc.global.loggingSQLAndTime.printUnprocessedStackTrace = false 
scalikejdbc.global.loggingSQLAndTime.stackTraceDepth = 10 
} 
dev { 
db { 
h2 { 
driver = "org.h2.Driver" 
url = "jdbc:h2:tcp://localhost/~/slickdemo" 
user = "" 
password = "" 
poolFactoryName = "hikaricp" 
numThreads = 10 
maxConnections = 12 
minConnections = 4 
keepAliveConnection = true 
} 
mysql { 
driver = "com.mysql.cj.jdbc.Driver" 
url = "jdbc:mysql://localhost:3306/testdb" 
user = "root" 
password = "123" 
poolInitialSize = 5 
poolMaxSize = 7 
poolConnectionTimeoutMillis = 1000 
poolValidationQuery = "select 1 as one" 
poolFactoryName = "bonecp" 
} 
postgres { 
driver = "org.postgresql.Driver" 
url = "jdbc:postgresql://localhost:5432/testdb" 
user = "root" 
password = "123" 
poolFactoryName = "hikaricp" 
numThreads = 10 
maxConnections = 12 
minConnections = 4 
keepAliveConnection = true 
} 
} 
# scallikejdbc Global settings 
scalikejdbc.global.loggingSQLAndTime.enabled = true 
scalikejdbc.global.loggingSQLAndTime.logLevel = info 
scalikejdbc.global.loggingSQLAndTime.warningEnabled = true 
scalikejdbc.global.loggingSQLAndTime.warningThresholdMillis = 1000 
scalikejdbc.global.loggingSQLAndTime.warningLogLevel = warn 
scalikejdbc.global.loggingSQLAndTime.singleLineMode = false 
scalikejdbc.global.loggingSQLAndTime.printUnprocessedStackTrace = false 
scalikejdbc.global.loggingSQLAndTime.stackTraceDepth = 10 
}

package sdp.jdbc.config 
import scala.collection.mutable 
import scala.concurrent.duration.Duration 
import scala.language.implicitConversions 
import com.typesafe.config._ 
import java.util.concurrent.TimeUnit 
import java.util.Properties 
import scalikejdbc.config._ 
import com.typesafe.config.Config 
import com.zaxxer.hikari._ 
import scalikejdbc.ConnectionPoolFactoryRepository 
/** Extension methods to make Typesafe Config easier to use */ 
class ConfigExtensionMethods(val c: Config) extends AnyVal { 
import scala.collection.JavaConverters._ 
def getBooleanOr(path: String, default: => Boolean = false) = if(c.hasPath(path)) c.getBoolean(path) else default 
def getIntOr(path: String, default: => Int = 0) = if(c.hasPath(path)) c.getInt(path) else default 
def getStringOr(path: String, default: => String = null) = if(c.hasPath(path)) c.getString(path) else default 
def getConfigOr(path: String, default: => Config = ConfigFactory.empty()) = if(c.hasPath(path)) c.getConfig(path) else default 
def getMillisecondsOr(path: String, default: => Long = 0L) = if(c.hasPath(path)) c.getDuration(path, TimeUnit.MILLISECONDS) else default 
def getDurationOr(path: String, default: => Duration = Duration.Zero) = 
if(c.hasPath(path)) Duration(c.getDuration(path, TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS) else default 
def getPropertiesOr(path: String, default: => Properties = null): Properties = 
if(c.hasPath(path)) new ConfigExtensionMethods(c.getConfig(path)).toProperties else default 
def toProperties: Properties = { 
def toProps(m: mutable.Map[String, ConfigValue]): Properties = { 
val props = new Properties(null) 
m.foreach { case (k, cv) => 
val v = 
if(cv.valueType() == ConfigValueType.OBJECT) toProps(cv.asInstanceOf[ConfigObject].asScala) 
else if(cv.unwrapped eq null) null 
else cv.unwrapped.toString 
if(v ne null) props.put(k, v) 
} 
props 
} 
toProps(c.root.asScala) 
} 
def getBooleanOpt(path: String): Option[Boolean] = if(c.hasPath(path)) Some(c.getBoolean(path)) else None 
def getIntOpt(path: String): Option[Int] = if(c.hasPath(path)) Some(c.getInt(path)) else None 
def getStringOpt(path: String) = Option(getStringOr(path)) 
def getPropertiesOpt(path: String) = Option(getPropertiesOr(path)) 
} 
object ConfigExtensionMethods { 
@inline implicit def configExtensionMethods(c: Config): ConfigExtensionMethods = new ConfigExtensionMethods(c) 
} 
trait HikariConfigReader extends TypesafeConfigReader { 
self: TypesafeConfig =>      // with TypesafeConfigReader => //NoEnvPrefix => 
 
import ConfigExtensionMethods.configExtensionMethods 
def getFactoryName(dbName: Symbol): String = { 
val c: Config = config.getConfig(envPrefix + "db." + dbName.name) 
c.getStringOr("poolFactoryName", ConnectionPoolFactoryRepository.COMMONS_DBCP) 
} 
def hikariCPConfig(dbName: Symbol): HikariConfig = { 
val hconf = new HikariConfig() 
val c: Config = config.getConfig(envPrefix + "db." + dbName.name) 
// Connection settings 
if (c.hasPath("dataSourceClass")) { 
hconf.setDataSourceClassName(c.getString("dataSourceClass")) 
} else { 
Option(c.getStringOr("driverClassName", c.getStringOr("driver"))).map(hconf.setDriverClassName _) 
} 
hconf.setJdbcUrl(c.getStringOr("url", null)) 
c.getStringOpt("user").foreach(hconf.setUsername) 
c.getStringOpt("password").foreach(hconf.setPassword) 
c.getPropertiesOpt("properties").foreach(hconf.setDataSourceProperties) 
// Pool configuration 
hconf.setConnectionTimeout(c.getMillisecondsOr("connectionTimeout", 1000)) 
hconf.setValidationTimeout(c.getMillisecondsOr("validationTimeout", 1000)) 
hconf.setIdleTimeout(c.getMillisecondsOr("idleTimeout", 600000)) 
hconf.setMaxLifetime(c.getMillisecondsOr("maxLifetime", 1800000)) 
hconf.setLeakDetectionThreshold(c.getMillisecondsOr("leakDetectionThreshold", 0)) 
hconf.setInitializationFailFast(c.getBooleanOr("initializationFailFast", false)) 
c.getStringOpt("connectionTestQuery").foreach(hconf.setConnectionTestQuery) 
c.getStringOpt("connectionInitSql").foreach(hconf.setConnectionInitSql) 
val numThreads = c.getIntOr("numThreads", 20) 
hconf.setMaximumPoolSize(c.getIntOr("maxConnections", numThreads * 5)) 
hconf.setMinimumIdle(c.getIntOr("minConnections", numThreads)) 
hconf.setPoolName(c.getStringOr("poolName", dbName.name)) 
hconf.setRegisterMbeans(c.getBooleanOr("registerMbeans", false)) 
// Equivalent of ConnectionPreparer 
hconf.setReadOnly(c.getBooleanOr("readOnly", false)) 
c.getStringOpt("isolation").map("TRANSACTION_" + _).foreach(hconf.setTransactionIsolation) 
hconf.setCatalog(c.getStringOr("catalog", null)) 
hconf 
} 
} 
import scalikejdbc._ 
trait ConfigDBs { 
self: TypesafeConfigReader with TypesafeConfig with HikariConfigReader => 
def setup(dbName: Symbol = ConnectionPool.DEFAULT_NAME): Unit = { 
getFactoryName(dbName) match { 
case "hikaricp" => { 
val hconf = hikariCPConfig(dbName) 
val hikariCPSource = new HikariDataSource(hconf) 
case class HikariDataSourceCloser(src: HikariDataSource) extends DataSourceCloser { 
var closed = false 
override def close(): Unit = src.close() 
} 
if (hconf.getDriverClassName != null && hconf.getDriverClassName.trim.nonEmpty) { 
Class.forName(hconf.getDriverClassName) 
} 
ConnectionPool.add(dbName, new DataSourceConnectionPool(dataSource = hikariCPSource,settings = DataSourceConnectionPoolSettings(), 
closer = HikariDataSourceCloser(hikariCPSource))) 
} 
case _ => { 
val JDBCSettings(url, user, password, driver) = readJDBCSettings(dbName) 
val cpSettings = readConnectionPoolSettings(dbName) 
if (driver != null && driver.trim.nonEmpty) { 
Class.forName(driver) 
} 
ConnectionPool.add(dbName, url, user, password, cpSettings) 
} 
} 
} 
def setupAll(): Unit = { 
loadGlobalSettings() 
dbNames.foreach { dbName => setup(Symbol(dbName)) } 
} 
def close(dbName: Symbol = ConnectionPool.DEFAULT_NAME): Unit = { 
ConnectionPool.close(dbName) 
} 
def closeAll(): Unit = { 
ConnectionPool.closeAll 
} 
} 
object ConfigDBs extends ConfigDBs 
with TypesafeConfigReader 
with StandardTypesafeConfig 
with HikariConfigReader 
case class ConfigDBsWithEnv(envValue: String) extends ConfigDBs 
with TypesafeConfigReader 
with StandardTypesafeConfig 
with HikariConfigReader 
with EnvPrefix { 
override val env = Option(envValue) 
}

package sdp.jdbc.engine 
import java.sql.PreparedStatement 
import scala.collection.generic.CanBuildFrom 
import akka.stream.scaladsl._ 
import scalikejdbc._ 
import scalikejdbc.streams._ 
import akka.NotUsed 
import akka.stream._ 
import scala.util._ 
import java.time._ 
import scala.concurrent.duration._ 
import sdp.jdbc.FileStreaming._ 
import scalikejdbc.TxBoundary.Try._ 
import scala.concurrent.ExecutionContextExecutor 
import java.io.InputStream 
object JDBCContext { 
type SQLTYPE = Int 
val SQL_EXEDDL= 1 
val SQL_UPDATE = 2 
val RETURN_GENERATED_KEYVALUE = true 
val RETURN_UPDATED_COUNT = false 
} 
case class JDBCQueryContext[M]( 
dbName: Symbol, 
statement: String, 
parameters: Seq[Any] = Nil, 
fetchSize: Int = 100, 
autoCommit: Boolean = false, 
queryTimeout: Option[Int] = None) 
// extractor: WrappedResultSet => M) 
case class JDBCContext( 
dbName: Symbol, 
statements: Seq[String] = Nil, 
parameters: Seq[Seq[Any]] = Nil, 
fetchSize: Int = 100, 
queryTimeout: Option[Int] = None, 
queryTags: Seq[String] = Nil, 
sqlType: JDBCContext.SQLTYPE = JDBCContext.SQL_UPDATE, 
batch: Boolean = false, 
returnGeneratedKey: Seq[Option[Any]] = Nil, 
// no return: None, return by index: Some(1), by name: Some("id") 
preAction: Option[PreparedStatement => Unit] = None, 
postAction: Option[PreparedStatement => Unit] = None) { 
ctx => 
//helper functions 
 
def appendTag(tag: String): JDBCContext = ctx.copy(queryTags = ctx.queryTags :+ tag) 
def appendTags(tags: Seq[String]): JDBCContext = ctx.copy(queryTags = ctx.queryTags ++ tags) 
def setFetchSize(size: Int): JDBCContext = ctx.copy(fetchSize = size) 
def setQueryTimeout(time: Option[Int]): JDBCContext = ctx.copy(queryTimeout = time) 
def setPreAction(action: Option[PreparedStatement => Unit]): JDBCContext = { 
if (ctx.sqlType == JDBCContext.SQL_UPDATE && 
!ctx.batch && ctx.statements.size == 1) 
ctx.copy(preAction = action) 
else 
throw new IllegalStateException("JDBCContex setting error: preAction not supported!") 
} 
def setPostAction(action: Option[PreparedStatement => Unit]): JDBCContext = { 
if (ctx.sqlType == JDBCContext.SQL_UPDATE && 
!ctx.batch && ctx.statements.size == 1) 
ctx.copy(postAction = action) 
else 
throw new IllegalStateException("JDBCContex setting error: preAction not supported!") 
} 
def appendDDLCommand(_statement: String, _parameters: Any*): JDBCContext = { 
if (ctx.sqlType == JDBCContext.SQL_EXEDDL) { 
ctx.copy( 
statements = ctx.statements ++ Seq(_statement), 
parameters = ctx.parameters ++ Seq(Seq(_parameters)) 
) 
} else 
throw new IllegalStateException("JDBCContex setting error: option not supported!") 
} 
def appendUpdateCommand(_returnGeneratedKey: Boolean, _statement: String, _parameters: Any*): JDBCContext = { 
if (ctx.sqlType == JDBCContext.SQL_UPDATE && !ctx.batch) { 
ctx.copy( 
statements = ctx.statements ++ Seq(_statement), 
parameters = ctx.parameters ++ Seq(_parameters), 
returnGeneratedKey = ctx.returnGeneratedKey ++ (if (_returnGeneratedKey) Seq(Some(1)) else Seq(None)) 
) 
} else 
throw new IllegalStateException("JDBCContex setting error: option not supported!") 
} 
def appendBatchParameters(_parameters: Any*): JDBCContext = { 
if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch) 
throw new IllegalStateException("JDBCContex setting error: batch parameters only supported for SQL_UPDATE and batch = true!") 
var matchParams = true 
if (ctx.parameters != Nil) 
if (ctx.parameters.head.size != _parameters.size) 
matchParams = false 
if (matchParams) { 
ctx.copy( 
parameters = ctx.parameters ++ Seq(_parameters) 
) 
} else 
throw new IllegalStateException("JDBCContex setting error: batch command parameters not match!") 
} 
def setBatchReturnGeneratedKeyOption(returnKey: Boolean): JDBCContext = { 
if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch) 
throw new IllegalStateException("JDBCContex setting error: only supported in batch update commands!") 
ctx.copy( 
returnGeneratedKey = if (returnKey) Seq(Some(1)) else Nil 
) 
} 
def setDDLCommand(_statement: String, _parameters: Any*): JDBCContext = { 
ctx.copy( 
statements = Seq(_statement), 
parameters = Seq(_parameters), 
sqlType = JDBCContext.SQL_EXEDDL, 
batch = false 
) 
} 
def setUpdateCommand(_returnGeneratedKey: Boolean, _statement: String, _parameters: Any*): JDBCContext = { 
ctx.copy( 
statements = Seq(_statement), 
parameters = Seq(_parameters), 
returnGeneratedKey = if (_returnGeneratedKey) Seq(Some(1)) else Seq(None), 
sqlType = JDBCContext.SQL_UPDATE, 
batch = false 
) 
} 
def setBatchCommand(_statement: String): JDBCContext = { 
ctx.copy ( 
statements = Seq(_statement), 
sqlType = JDBCContext.SQL_UPDATE, 
batch = true 
) 
} 
type JDBCDate = LocalDate 
type JDBCDateTime = LocalDateTime 
def jdbcSetDate(yyyy: Int, mm: Int, dd: Int) = LocalDate.of(yyyy,mm,dd) 
def jdbcSetNow = LocalDateTime.now() 
type JDBCBlob = InputStream 
def fileToJDBCBlob(fileName: String, timeOut: FiniteDuration = 60 seconds)( 
implicit mat: Materializer) = FileToInputStream(fileName,timeOut) 
def jdbcBlobToFile(blob: JDBCBlob, fileName: String)( 
implicit mat: Materializer) =  InputStreamToFile(blob,fileName) 
} 
object JDBCEngine { 
import JDBCContext._ 
private def noExtractor(message: String): WrappedResultSet => Nothing = { (rs: WrappedResultSet) => 
throw new IllegalStateException(message) 
} 
def jdbcAkkaStream[A](ctx: JDBCQueryContext[A],extractor: WrappedResultSet => A) 
(implicit ec: ExecutionContextExecutor): Source[A,NotUsed] = { 
val publisher: DatabasePublisher[A] = NamedDB('h2) readOnlyStream { 
val rawSql = new SQLToCollectionImpl[A, NoExtractor](ctx.statement, ctx.parameters)(noExtractor("")) 
ctx.queryTimeout.foreach(rawSql.queryTimeout(_)) 
val sql: SQL[A, HasExtractor] = rawSql.map(extractor) 
sql.iterator 
.withDBSessionForceAdjuster(session => { 
session.connection.setAutoCommit(ctx.autoCommit) 
session.fetchSize(ctx.fetchSize) 
}) 
} 
Source.fromPublisher[A](publisher) 
} 
def jdbcQueryResult[C[_] <: TraversableOnce[_], A](ctx: JDBCQueryContext[A], 
extractor: WrappedResultSet => A)( 
implicit cbf: CanBuildFrom[Nothing, A, C[A]]): C[A] = { 
val rawSql = new SQLToCollectionImpl[A, NoExtractor](ctx.statement, ctx.parameters)(noExtractor("")) 
ctx.queryTimeout.foreach(rawSql.queryTimeout(_)) 
rawSql.fetchSize(ctx.fetchSize) 
implicit val session = NamedAutoSession(ctx.dbName) 
val sql: SQL[A, HasExtractor] = rawSql.map(extractor) 
sql.collection.apply[C]() 
} 
def jdbcExcuteDDL(ctx: JDBCContext): Try[String] = { 
if (ctx.sqlType != SQL_EXEDDL) { 
Failure(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_EXEDDL'!")) 
} 
else { 
NamedDB(ctx.dbName) localTx { implicit session => 
Try { 
ctx.statements.foreach { stm => 
val ddl = new SQLExecution(statement = stm, parameters = Nil)( 
before = WrappedResultSet => {})( 
after = WrappedResultSet => {}) 
ddl.apply() 
} 
"SQL_EXEDDL executed succesfully." 
} 
} 
} 
} 
def jdbcBatchUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)( 
implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = { 
if (ctx.statements == Nil) 
throw new IllegalStateException("JDBCContex setting error: statements empty!") 
if (ctx.sqlType != SQL_UPDATE) { 
Failure(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!")) 
} 
else { 
if (ctx.batch) { 
if (noReturnKey(ctx)) { 
val usql = SQL(ctx.statements.head) 
.tags(ctx.queryTags: _*) 
.batch(ctx.parameters: _*) 
Try { 
NamedDB(ctx.dbName) localTx { implicit session => 
ctx.queryTimeout.foreach(session.queryTimeout(_)) 
usql.apply[Seq]() 
Seq.empty[Long].to[C] 
} 
} 
} else { 
val usql = new SQLBatchWithGeneratedKey(ctx.statements.head, ctx.parameters, ctx.queryTags)(None) 
Try { 
NamedDB(ctx.dbName) localTx { implicit session => 
ctx.queryTimeout.foreach(session.queryTimeout(_)) 
usql.apply[C]() 
} 
} 
} 
} else { 
Failure(new IllegalStateException("JDBCContex setting error: must set batch = true !")) 
} 
} 
} 
private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)( 
implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = { 
val Some(key) :: xs = ctx.returnGeneratedKey 
val params: Seq[Any] = ctx.parameters match { 
case Nil => Nil 
case p@_ => p.head 
} 
val usql = new SQLUpdateWithGeneratedKey(ctx.statements.head, params, ctx.queryTags)(key) 
Try { 
NamedDB(ctx.dbName) localTx { implicit session => 
session.fetchSize(ctx.fetchSize) 
ctx.queryTimeout.foreach(session.queryTimeout(_)) 
val result = usql.apply() 
Seq(result).to[C] 
} 
} 
} 
private def singleTxUpdateNoReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)( 
implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = { 
val params: Seq[Any] = ctx.parameters match { 
case Nil => Nil 
case p@_ => p.head 
} 
val before = ctx.preAction match { 
case None => pstm: PreparedStatement => {} 
case Some(f) => f 
} 
val after = ctx.postAction match { 
case None => pstm: PreparedStatement => {} 
case Some(f) => f 
} 
val usql = new SQLUpdate(ctx.statements.head,params,ctx.queryTags)(before)(after) 
Try { 
NamedDB(ctx.dbName) localTx {implicit session => 
session.fetchSize(ctx.fetchSize) 
ctx.queryTimeout.foreach(session.queryTimeout(_)) 
val result = usql.apply() 
Seq(result.toLong).to[C] 
} 
} 
} 
private def singleTxUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)( 
implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = { 
if (noReturnKey(ctx)) 
singleTxUpdateNoReturnKey(ctx) 
else 
singleTxUpdateWithReturnKey(ctx) 
} 
private def noReturnKey(ctx: JDBCContext): Boolean = { 
if (ctx.returnGeneratedKey != Nil) { 
val k :: xs = ctx.returnGeneratedKey 
k match { 
case None => true 
case Some(k) => false 
} 
} else true 
} 
def noActon: PreparedStatement=>Unit = pstm => {} 
def multiTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)( 
implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = { 
Try { 
NamedDB(ctx.dbName) localTx { implicit session => 
session.fetchSize(ctx.fetchSize) 
ctx.queryTimeout.foreach(session.queryTimeout(_)) 
val keys: Seq[Option[Any]] = ctx.returnGeneratedKey match { 
case Nil => Seq.fill(ctx.statements.size)(None) 
case k@_ => k 
} 
val sqlcmd = ctx.statements zip ctx.parameters zip keys 
val results = sqlcmd.map { case ((stm, param), key) => 
key match { 
case None => 
new SQLUpdate(stm, param, Nil)(noActon)(noActon).apply().toLong 
case Some(k) => 
new SQLUpdateWithGeneratedKey(stm, param, Nil)(k).apply().toLong 
} 
} 
results.to[C] 
} 
} 
} 
def jdbcTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)( 
implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = { 
if (ctx.statements == Nil) 
throw new IllegalStateException("JDBCContex setting error: statements empty!") 
if (ctx.sqlType != SQL_UPDATE) { 
Failure(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!")) 
} 
else { 
if (!ctx.batch) { 
if (ctx.statements.size == 1) 
singleTxUpdate(ctx) 
else 
multiTxUpdates(ctx) 
} else 
Failure(new IllegalStateException("JDBCContex setting error: must set batch = false !")) 
} 
} 
case class JDBCActionStream[R](dbName: Symbol, parallelism: Int = 1, processInOrder: Boolean = true, 
statement: String, prepareParams: R => Seq[Any]) { 
jas => 
def setDBName(db: Symbol): JDBCActionStream[R] = jas.copy(dbName = db) 
def setParallelism(parLevel: Int): JDBCActionStream[R] = jas.copy(parallelism = parLevel) 
def setProcessOrder(ordered: Boolean): JDBCActionStream[R] = jas.copy(processInOrder = ordered) 
private def perform(r: R) = { 
import scala.concurrent._ 
val params = prepareParams(r) 
NamedDB(dbName) autoCommit { session => 
session.execute(statement,params: _*) 
} 
Future.successful(r) 
} 
def performOnRow(implicit session: DBSession): Flow[R, R, NotUsed] = 
if (processInOrder) 
Flow[R].mapAsync(parallelism)(perform) 
else 
Flow[R].mapAsyncUnordered(parallelism)(perform) 
} 
object JDBCActionStream { 
def apply[R](_dbName: Symbol, _statement: String, params: R => Seq[Any]): JDBCActionStream[R] = 
new JDBCActionStream[R](dbName = _dbName, statement=_statement, prepareParams = params) 
} 
}

package sdp.jdbc 
import java.io.{InputStream, ByteArrayInputStream} 
import java.nio.ByteBuffer 
import java.nio.file.Paths 
import akka.stream.{Materializer} 
import akka.stream.scaladsl.{FileIO, StreamConverters} 
import scala.concurrent.{Await} 
import akka.util._ 
import scala.concurrent.duration._ 
object FileStreaming { 
def FileToByteBuffer(fileName: String, timeOut: FiniteDuration = 60 seconds)( 
implicit mat: Materializer):ByteBuffer = { 
val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) => 
hd ++ bs 
} 
(Await.result(fut, timeOut)).toByteBuffer 
} 
def FileToByteArray(fileName: String, timeOut: FiniteDuration = 60 seconds)( 
implicit mat: Materializer): Array[Byte] = { 
val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) => 
hd ++ bs 
} 
(Await.result(fut, timeOut)).toArray 
} 
def FileToInputStream(fileName: String, timeOut: FiniteDuration = 60 seconds)( 
implicit mat: Materializer): InputStream = { 
val fut = FileIO.fromPath(Paths.get(fileName)).runFold(ByteString()) { case (hd, bs) => 
hd ++ bs 
} 
val buf = (Await.result(fut, timeOut)).toArray 
new ByteArrayInputStream(buf) 
} 
def ByteBufferToFile(byteBuf: ByteBuffer, fileName: String)( 
implicit mat: Materializer) = { 
val ba = new Array[Byte](byteBuf.remaining()) 
byteBuf.get(ba,0,ba.length) 
val baInput = new ByteArrayInputStream(ba) 
val source = StreamConverters.fromInputStream(() => baInput)  //ByteBufferInputStream(bytes)) 
source.runWith(FileIO.toPath(Paths.get(fileName))) 
} 
def ByteArrayToFile(bytes: Array[Byte], fileName: String)( 
implicit mat: Materializer) = { 
val bb = ByteBuffer.wrap(bytes) 
val baInput = new ByteArrayInputStream(bytes) 
val source = StreamConverters.fromInputStream(() => baInput) //ByteBufferInputStream(bytes)) 
source.runWith(FileIO.toPath(Paths.get(fileName))) 
} 
def InputStreamToFile(is: InputStream, fileName: String)( 
implicit mat: Materializer) = { 
val source = StreamConverters.fromInputStream(() => is) 
source.runWith(FileIO.toPath(Paths.get(fileName))) 
} 
}

package protobuf.bytes 
import java.io.{ByteArrayInputStream,ByteArrayOutputStream,ObjectInputStream,ObjectOutputStream} 
import com.google.protobuf.ByteString 
object Converter { 
  def marshal(value: Any): ByteString = { 
    val stream: ByteArrayOutputStream = new ByteArrayOutputStream() 
    val oos = new ObjectOutputStream(stream) 
    oos.writeObject(value) 
    oos.close() 
    ByteString.copyFrom(stream.toByteArray()) 
  } 
 
  def unmarshal[A](bytes: ByteString): A = { 
    val ois = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray)) 
    val value = ois.readObject() 
    ois.close() 
    value.asInstanceOf[A] 
  } 
}

package demo.grpc.jdbc.services 
import akka.NotUsed 
import akka.stream.scaladsl.Flow 
import grpc.jdbc.services._ 
import java.util.logging.Logger 
import protobuf.bytes.Converter._ 
import sdp.jdbc.engine._ 
import JDBCEngine._ 
import scalikejdbc.WrappedResultSet 
import scala.concurrent.ExecutionContextExecutor 
class JDBCStreamingServices(implicit ec: ExecutionContextExecutor) extends JdbcGrpcAkkaStream.JDBCServices { 
val logger = Logger.getLogger(classOf[JDBCStreamingServices].getName) 
val toRow = (rs: WrappedResultSet) => JDBCDataRow( 
year = rs.string("REPORTYEAR"), 
state = rs.string("STATENAME"), 
county = rs.string("COUNTYNAME"), 
value = rs.string("VALUE") 
) 
override def runQuery: Flow[JDBCQuery, JDBCDataRow, NotUsed] = { 
logger.info("**** runQuery called on service side ***") 
Flow[JDBCQuery] 
.flatMapConcat { q => 
//unpack JDBCQuery and construct the context 
val params: Seq[Any] = unmarshal[Seq[Any]](q.parameters) 
logger.info(s"**** query parameters: ${params} ****") 
val ctx = JDBCQueryContext[JDBCDataRow]( 
dbName = Symbol(q.dbName), 
statement = q.statement, 
parameters = params, 
fetchSize = q.fetchSize.getOrElse(100), 
autoCommit = q.autoCommit.getOrElse(false), 
queryTimeout = q.queryTimeout 
) 
jdbcAkkaStream(ctx, toRow) 
} 
} 
}

package demo.grpc.jdbc.server 
import java.util.logging.Logger 
import akka.actor.ActorSystem 
import akka.stream.ActorMaterializer 
import io.grpc.Server 
import demo.grpc.jdbc.services._ 
import io.grpc.ServerBuilder 
import grpc.jdbc.services._ 
class gRPCServer(server: Server) { 
val logger: Logger = Logger.getLogger(classOf[gRPCServer].getName) 
def start(): Unit = { 
server.start() 
logger.info(s"Server started, listening on ${server.getPort}") 
sys.addShutdownHook { 
// Use stderr here since the logger may has been reset by its JVM shutdown hook. 
System.err.println("*** shutting down gRPC server since JVM is shutting down") 
stop() 
System.err.println("*** server shut down") 
} 
() 
} 
def stop(): Unit = { 
server.shutdown() 
} 
/** 
* Await termination on the main thread since the grpc library uses daemon threads. 
*/ 
def blockUntilShutdown(): Unit = { 
server.awaitTermination() 
} 
} 
object JDBCServer extends App { 
import sdp.jdbc.config._ 
implicit val system = ActorSystem("JDBCServer") 
implicit val mat = ActorMaterializer.create(system) 
implicit val ec = system.dispatcher 
ConfigDBsWithEnv("dev").setup('h2) 
ConfigDBsWithEnv("dev").loadGlobalSettings() 
val server = new gRPCServer( 
ServerBuilder 
.forPort(50051) 
.addService( 
JdbcGrpcAkkaStream.bindService( 
new JDBCStreamingServices 
) 
).build() 
) 
server.start() 
//  server.blockUntilShutdown() 
  scala.io.StdIn.readLine() 
ConfigDBsWithEnv("dev").close('h2) 
  mat.shutdown() 
system.terminate() 
}

package demo.grpc.jdbc.client 
import grpc.jdbc.services._ 
import java.util.logging.Logger 
import protobuf.bytes.Converter._ 
import akka.stream.scaladsl._ 
import akka.NotUsed 
import akka.actor.ActorSystem 
import akka.stream.{ActorMaterializer, ThrottleMode} 
import io.grpc._ 
class JDBCStreamClient(host: String, port: Int) { 
val logger: Logger = Logger.getLogger(classOf[JDBCStreamClient].getName) 
val channel = ManagedChannelBuilder 
.forAddress(host,port) 
.usePlaintext(true) 
.build() 
val stub = JdbcGrpcAkkaStream.stub(channel) 
val query = JDBCQuery ( 
dbName = "h2", 
statement = "select * from AQMRPT where STATENAME = ? and VALUE = ?", 
parameters = marshal(Seq("Arizona", 5)) 
) 
def queryRows: Source[JDBCDataRow,NotUsed] = { 
logger.info(s"running queryRows ...") 
Source 
.single(query) 
.via(stub.runQuery) 
} 
} 
object QueryRows extends App { 
implicit val system = ActorSystem("QueryRows") 
implicit val mat = ActorMaterializer.create(system) 
val client = new JDBCStreamClient("localhost", 50051) 
client.queryRows.runForeach(println) 
scala.io.StdIn.readLine() 
mat.shutdown() 
system.terminate() 
}