//设置package包名称以及导入依赖的类
package domain

import java.sql.{Connection, ResultSet, Statement}

import scala.collection.immutable.IndexedSeq

case class DBAccessWorker(connection: Connection, sql: String) {

  private def execute(execSql: String) = {
    val stmt: Statement = connection.createStatement
    val rs: ResultSet = stmt.executeQuery(execSql)
    val columnCnt: Int = rs.getMetaData.getColumnCount

    val columns: IndexedSeq[String] = 1 to columnCnt map rs.getMetaData.getColumnName
    val results: Iterator[IndexedSeq[String]] = Iterator.continually(rs).takeWhile(_.next()).map{ rs =>
      columns map rs.getString
    }

    (columns, results)
  }

  def execute(): SqlResults = {
    val (d1, d2) = execute(sql)
    SqlResults(d1, d2)
  }

  def executeStatistics(): Statistics = {
    val (d1, d2) = execute(s"EXPLAIN $sql")
    Statistics(d1, d2)
  }
}

case class SqlResults(columns: IndexedSeq[String], results: Iterator[IndexedSeq[String]])

case class Statistics(columns: IndexedSeq[String], results: Iterator[IndexedSeq[String]]) 

//设置package包名称以及导入依赖的类
package com.citypay.pan.search.db

import java.sql.{Connection, ResultSet}
import javax.sql.DataSource

import com.jolbox.bonecp.BoneCP

import scala.collection.mutable.ListBuffer

object DbUtils {

  def withConn[T](ds: BoneCP)(fn: Connection => T): T = {
    val conn = ds.getConnection
    try {
      fn(conn)
    } finally {
      conn.close()
    }
  }

  implicit class RsExtender(rs: ResultSet) {

    def toList[T](row: ResultSet => T): List[T] = {
      var buffer = ListBuffer[T]()
      try {
        while (rs.next()) {
          buffer += row(rs)
        }
      } finally {
        rs.close()
      }
      buffer.toList
    }

  }


} 

//设置package包名称以及导入依赖的类
package yumi.pipeline.sinks
import java.sql.{Connection, PreparedStatement}

import org.apache.spark.sql.jdbc.JdbcDialect
import org.apache.spark.sql.types._


class SqlServerSink extends DatabaseSink
  with Serializable {

  override def bulkInsertStatement (conn: Connection,
                                    table: String,
                                    rddSchema: StructType,
                                    dialect: JdbcDialect,
                                    rowCount: Int) : PreparedStatement = {
    val columns = rddSchema.fields.map(x => dialect.quoteIdentifier(x.name)).mkString(",")
    val parameters = rddSchema.fields.map(_ => "?").mkString("(", ",", ")")
    val sql = new StringBuilder(s"INSERT INTO $table($columns) VALUES ")

    (1 to rowCount).foldLeft(false) { (prependComma, index) =>

      if (prependComma) {
        sql.append(", ")
      }

      sql.append(parameters)

      true
    }

    conn.prepareStatement(sql.toString())
  }
} 

//设置package包名称以及导入依赖的类
package yumi.pipeline.sinks

import java.sql.{Connection, PreparedStatement}

import org.apache.spark.sql.jdbc.JdbcDialect
import org.apache.spark.sql.types._

class OracleDatabaseSink extends DatabaseSink with Serializable {

  override def bulkInsertStatement (conn: Connection,
                                    table: String,
                                    rddSchema: StructType,
                                    dialect: JdbcDialect,
                                    rowCount: Int) : PreparedStatement = {
    val columns = rddSchema.fields.map(x => dialect.quoteIdentifier(x.name)).mkString(",")
    val parameters = rddSchema.fields.map(_ => "?").mkString(s"INTO $table($columns) VALUES (", ",", ")")
    val sql = new StringBuilder("INSERT ALL ")

    (1 to rowCount).foldLeft(false) { (prependComma, index) =>

      if (prependComma) {
        sql.append(" ")
      }

      sql.append(parameters)

      true
    }
    sql.append("  SELECT 1 FROM DUAL")

    conn.prepareStatement(sql.toString())
  }
} 

//设置package包名称以及导入依赖的类
package yumi.pipeline.sinks

import java.sql.{Connection, PreparedStatement}

import org.apache.spark.sql.jdbc.JdbcDialect
import org.apache.spark.sql.types._

class MySqlSink extends DatabaseSink with Serializable {

  override def bulkInsertStatement (conn: Connection,
                                    table: String,
                                    rddSchema: StructType,
                                    dialect: JdbcDialect,
                                    rowCount: Int) : PreparedStatement = {
    val columns = rddSchema.fields.map(x => dialect.quoteIdentifier(x.name)).mkString(",")
    val parameters = rddSchema.fields.map(_ => "?").mkString("(", ",", ")")
    val sql = new StringBuilder(s"INSERT INTO $table($columns) VALUES ")

    (1 to rowCount).foldLeft(false) { (prependComma, index) =>

      if (prependComma) {
        sql.append(", ")
      }

      sql.append(parameters)

      true
    }

    conn.prepareStatement(sql.toString())
  }
} 

//设置package包名称以及导入依赖的类
package org.blinkmob

import java.sql.Connection
import javax.sql.DataSource

import resource._

import scala.util.control.ControlThrowable
import scala.util.{Failure, Success, Try}



trait CxProvider{ this:hasDataSource =>

  def cx[A](block: Connection => A): A = managed(ds.getConnection()).acquireAndGet(block(_))

  def tx[A](block: Connection => A): A ={
    cx { c =>
      c.setAutoCommit(false)
      Try({
        val r = block(c)
        c.commit()
        r
      }) match{
        case Success(r) => r
        case Failure(f) => {
          f match{
            case e: ControlThrowable => c.commit(); throw e
            case e => c.rollback(); throw e
          }
        }
      }


    }
  }

  def rbtx[A](block: Connection => A): A ={
    cx { c =>
      c.setAutoCommit(false)
      Try(block(c)) match{
        case Success(r) => c.rollback(); r
        case Failure(f) => c.rollback(); throw f
      }
    }
  }

} 

//设置package包名称以及导入依赖的类
package storage

import akka.actor.ActorSystem
import com.google.inject.AbstractModule
import java.sql.Connection
import javax.inject.{ Inject, Singleton }
import models.user.UserService
import models.user.Roles
import org.jooq.impl.DSL
import org.jooq.{ SQLDialect, DSLContext }
import play.api.db.Database
import scala.collection.JavaConversions._
import scala.concurrent.{ ExecutionContext, Future }
import scala.io.Source

object DB {
  
  val CURRENT_SQLDIALECTT = SQLDialect.POSTGRES_9_4

}


  private def initDB(connection: Connection) = {
    
    // Splitting by ; is not 100% robust - but should be sufficient for our own schema file
    val statement = connection.createStatement

    Source.fromFile("conf/schema.sql", "UTF-8")
      .getLines().map(_.trim)
      .filter(line => !(line.startsWith("--") || line.isEmpty))
      .mkString(" ").split(";")
      .foreach(s => {
        statement.addBatch(s + ";")
      })

    statement.executeBatch()
    statement.close()    
    
    val f = for {
      _ <- userService.insertUser("recogito", "[email protected]", "recogito")
      _ <- userService.insertUserRole("recogito", Roles.Admin)
    } yield()
    
    f.recover { case t: Throwable => t.printStackTrace() }
  } 
  
} 

//设置package包名称以及导入依赖的类
package repositories.filesystem

import java.sql.Connection
import java.util.UUID
import javax.inject.Inject

import anorm.SqlParser._
import anorm._
import models.{FsNode, Permission}
import play.api.db.DBApi

import scala.concurrent.ExecutionContext

class [email protected]()(
  dbApi: DBApi
)(
  implicit ec: ExecutionContext
) {

  import PermissionRepository._

  private[filesystem] def insertNonAtomic(fileSystemElement: FsNode, permission: Permission)(implicit c: Connection) = {
    insertPermission(fileSystemElement, permission).execute()
  }

}

object PermissionRepository {

  val table = "permission"

  val parser = {
    get[UUID]("account_id") ~
    get[Array[String]]("permissions")  map {
      case account_id ~ permissions
        => Permission(account_id, permissions)
    }
  }

  private def insertPermission(fileSystemElement: FsNode, permission: Permission) = SQL"""
     INSERT INTO #$table (
       account_id,
       directory_id,
       permissions)
     VALUES (
       ${permission.accountId}::uuid,
       ${fileSystemElement.id}::uuid,
       ${permission.permissions.toArray[String]}
     );
    """
} 

//设置package包名称以及导入依赖的类
import java.sql.Connection
import java.sql.ResultSet
import java.sql.DriverManager


object ScalaDatabase{

	def getData(conn:Connection) : String = {
	  
	  println("In get data")
		Class.forName("com.mysql.jdbc.Driver").newInstance();
    val st = conn.createStatement()
		val rs = st.executeQuery("select * from student where name = 'Shantanu'")
		var result = ""
		//println("COUNT")
		while(rs.next())
		{
			val id = rs.getInt("id")
			val name = rs.getString("name")
			val uni = rs.getString("university")
			val age = rs.getInt("age")
			result = "Student id is: " + id + "\n" + "Student name is: " + name + "\n" + "Student university is: " + uni + "\n" + "Student age is: " + age
		}
		
		rs.close() 
		//conn.close()
		return result
		
	}

	def time[R](block: => R): R = {  
    	val t0 = System.nanoTime()
    	val result = block    // call-by-name
    	val t1 = System.nanoTime()
    	println("Elapsed time: " + (t1 - t0) + "ns")
    	result
	}

	def main(args: Array[String]): Unit = {
	  
	  var conn:Connection = null

		conn = DriverManager.getConnection("jdbc:mysql://localhost/language_comparison", "guest", "admin123")

		val result = time {getData(conn)}
		println(result)

        val runtime = Runtime.getRuntime
        println("** Used Memory:  " + (runtime.totalMemory - runtime.freeMemory))
	}
} 

//设置package包名称以及导入依赖的类
package com.dao

import java.sql.{Connection, DriverManager, SQLException}


object OracleUtil {
  val driverClass = "oracle.jdbc.driver.OracleDriver"
  val jdbcUrl = "jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=172.30.12.26)(PORT=1521))(FAILOVER=ON)(LOAD_BALANCE=ON))(CONNECT_DATA= (SERVICE_NAME=msds)))"
  val username = "acctpaymertest1"
  val password = "acctpaymer*1"

  try {
    Class.forName(driverClass)
  } catch {
    case e: ClassNotFoundException => throw e
    case e: Exception => throw e
  }

  @throws(classOf[SQLException])
  def getConnection: Connection = {
    DriverManager.getConnection(jdbcUrl, username, password)
  }

  @throws(classOf[SQLException])
  def doTrancation(trancations: Set[String]): Unit = {
    val connection = getConnection
    connection.setAutoCommit(false)
    trancations.foreach(connection.createStatement().execute(_))
    connection.commit
    connection.close
  }
} 

//设置package包名称以及导入依赖的类
package models

import java.sql.{Connection, PreparedStatement, ResultSet}
import play.api.db._
import play.api.Play.current
import java.sql.Date
import java.sql.Time


case class ConsultaReserva(var id: Int, var fecha: String, var hora: String, var usuario_email: String, var mesa_numero: Int, var restaurante_nit: String, var estado: Int)


object ConsultaReservaRepository {
  val Insert = "INSERT INTO reserva(fecha,hora,usuario_email,mesa_numero,restaurante_nit) values(?,?,?,?,?)"

}


class ConsultaReservas {

  import ConsultaReservaRepository._

  def traerTodo1(date_ini : Long, date_end : Long): Seq[ConsultaReserva] = {
    val conn = DB.getConnection()
    try {
      val stmt = conn.createStatement

      val rs = stmt.executeQuery("SELECT id, fecha, hora, usuario_email, mesa_numero, restaurante_nit, estado FROM reserva WHERE fecha BETWEEN '"+date_ini+"' AND '"+date_end+"' ORDER BY id DESC")

      var ConsultaReservas: Seq[ConsultaReserva] = Seq.empty

      while (rs.next) {
        val consultaReserva = new ConsultaReserva(rs.getInt("id"),
          rs.getString("fecha"),
          rs.getString("hora"),
          rs.getString("usuario_email"),
          rs.getInt("mesa_numero"),
          rs.getString("restaurante_nit"),
          rs.getInt("estado")
        )
        ConsultaReservas = ConsultaReservas :+ consultaReserva
      }
      ConsultaReservas
    } finally {
      conn.close()
    }

  }

  private def crearPrepareStatementGuardar(conexion: Connection, consultaReserva: ConsultaReserva): PreparedStatement = {
    val preparedStatement = conexion.prepareStatement(Insert)
    var date: Date = Date.valueOf(consultaReserva.fecha)
    var time: Time = Time.valueOf(consultaReserva.hora)
    preparedStatement.setDate(1, date)
    preparedStatement.setTime(2, time)
    preparedStatement.setString(3, consultaReserva.usuario_email)
    preparedStatement.setInt(4, consultaReserva.mesa_numero)
    preparedStatement.setString(5, consultaReserva.restaurante_nit)
    preparedStatement
  }
} 

//设置package包名称以及导入依赖的类
package pl.gigiel.vocabulary.tool

import java.sql.DriverManager
import java.sql.Connection
import org.h2.tools.Server

object SimpleApp extends App {
  //val fileName = "E:\\_tgl\\workspace-tomaszacer\\vocabulary-tool-confidential\\resources\\jack-london-white-fang.txt"
  //val all = words(fileName)
  //println(all.length)
  //println(all)

  db
  println("ok")

  def words(fileName: String): List[String] = {
    val source = scala.io.Source.fromFile(fileName)
    val wordPattern = """([A-Za-z\-'])+""".r
    val words = try source.getLines.flatMap(wordPattern.findAllIn).map(_.toLowerCase).toList.distinct.sorted finally source.close()

    return words
  }

  def db: Unit = {
    val driver = "org.h2.Driver"
    val url = "jdbc:h2:mem:vocabulary-db;DB_CLOSE_DELAY=-1;DATABASE_TO_UPPER=false"
    val username = "vocabulary-username"
    val password = "vocabulary-password"

    var server: Server = null
    var connection: Connection = null

    try {
      Class.forName(driver) // unnecessary in jdbc 4+
      server = Server.createTcpServer("-tcpPort", "9123", "-tcpAllowOthers").start
      // jdbc:h2:tcp://localhost:9123/mem:vocabulary-db;DB_CLOSE_DELAY=-1;DATABASE_TO_UPPER=false
      connection = DriverManager.getConnection(url, username, password)

      val stmt = connection.createStatement()
      stmt.execute("create table infos(id int primary key, info varchar(255))")
      stmt.execute("insert into infos(id, info) values(1, 'abc')")
      val rs = stmt.executeQuery("select id, info from infos")
      while (rs.next()) {
        val id = rs.getString("id")
        val info = rs.getString("info")
        println("id, info = " + id + ", " + info)
      }
      io.StdIn.readInt
    } catch {
      case e: Throwable => e.printStackTrace
    } finally {
      connection.close
      server.stop
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.example.db


import java.sql.{Connection, DriverManager}

import com.twitter.util.Future
import org.postgresql.util.HostSpec


object DB {
  val url = "jdbc:postgresql://localhost:5432/postgres?user=Sergey&password=root"
  val connection = DriverManager.getConnection(url)

  def isValidToken(token: String) = {
    val statement = connection.prepareStatement("SELECT access_token FROM user_access WHERE access_token = ?")
    statement.setString(1, token)
    Future {
      statement.executeQuery()
    } map { resultSet =>
      resultSet.next()
    }
  }

  def createToken(userId: BigInt): Future[String] = {
    val token = java.util.UUID.randomUUID().toString
    val statement = connection.prepareStatement("INSERT INTO user_access VALUES (?, ?)")
    statement.setBigDecimal(2, BigDecimal(userId).bigDecimal)
    statement.setString(1, token)
    Future {
      statement.executeUpdate()
    } map {
      case 1 => token
      case _ => ""
    }
  }
} 

//设置package包名称以及导入依赖的类
package sql.migration

import java.sql.Connection

import com.typesafe.slick.testkit.util.DelegateConnection
import org.apache.commons.codec.digest.DigestUtils
import org.flywaydb.core.api.migration.jdbc.JdbcMigration
import slick.driver.PostgresDriver.api._
import slick.jdbc.JdbcDataSource

import scala.concurrent.Await
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.concurrent.forkjoin.ThreadLocalRandom

class V20150520123900__InitGroupsBots extends JdbcMigration {
  def migrate(connection: Connection): Unit = {
    val wrappedConn = new DelegateConnection(connection) {
      override def close(): Unit = ()
    }
    val db = Database.forSource(new JdbcDataSource {
      def createConnection(): Connection = wrappedConn
      def close(): Unit = ()
    })

    Await.ready(
      db.run {
        for {
          groupIds ? sql"select id from groups".as[Int]
          _ ? DBIO.sequence(groupIds.map { groupId ?
            val rnd = ThreadLocalRandom.current()
            val botId = nextIntId(rnd)
            val accessSalt = nextAccessSalt(rnd)
            val botToken = DigestUtils.sha256Hex(rnd.nextInt().toString)
            for {
              _ ? sqlu"""insert into users(id, access_salt, name, country_code, sex, state, deleted_at, is_bot)
                values($botId, $accessSalt, 'Bot', 'US', 1, 1, null, true)"""
              _ ? sqlu"insert into groups_bots(group_id, user_id, token) values($groupId, $botId, $botToken)"
            } yield ()
          })
        } yield ()
      }, 5.seconds
    )

  }

  private def nextIntId(rnd: ThreadLocalRandom): Int = rnd.nextInt(Int.MaxValue) + 1

  private def nextAccessSalt(rnd: ThreadLocalRandom): String = rnd.nextLong().toString
} 

//设置package包名称以及导入依赖的类
package im.actor.server.db

import java.sql.Connection
import javax.sql.DataSource

import org.flywaydb.core.Flyway
import org.flywaydb.core.api.MigrationInfo
import org.flywaydb.core.api.callback.FlywayCallback

trait FlywayInit {
  def initFlyway(ds: DataSource) = {
    val flyway = new Flyway()
    flyway.setDataSource(ds)
    flyway.setLocations("sql.migration")
    flyway.setCallbacks(new BeforeCleanCallback())
    flyway.setBaselineOnMigrate(true)
    //??????.sql?????????????.sql???
    //by Lining 2016/7/27
    flyway.setValidateOnMigrate(false)
    flyway
  }
}

class BeforeCleanCallback extends FlywayCallback {
  def afterBaseline(connection: Connection): Unit = {}
  def afterClean(connection: Connection): Unit = {}
  def afterEachMigrate(connection: Connection): Unit = {}
  def afterInfo(connection: Connection): Unit = {}
  def afterEachMigrate(connection: Connection, migrationInfo: MigrationInfo): Unit = {}
  def afterMigrate(connection: Connection): Unit = {}
  def afterRepair(connection: Connection): Unit = {}
  def afterValidate(connection: Connection): Unit = {}
  def beforeBaseline(connection: Connection): Unit = {}
  def beforeClean(connection: Connection): Unit = executeStmt(connection, """DROP EXTENSION IF EXISTS "ltree" CASCADE;""")
  def beforeEachMigrate(connection: Connection, migrationInfo: MigrationInfo): Unit = {}
  def beforeInfo(connection: Connection): Unit = {}
  def beforeInit(connection: Connection): Unit = {}
  def beforeMigrate(connection: Connection): Unit = {}
  def beforeRepair(connection: Connection): Unit = {}
  def beforeValidate(connection: Connection): Unit = {}
  def afterInit(connection: Connection): Unit = executeStmt(connection, """DROP EXTENSION IF EXISTS "ltree" CASCADE;""")

  def executeStmt(connection: Connection, statement: String): Unit = {
    if (connection.getMetaData().getDriverName().startsWith("PostgreSQL")) {
      val stmt = connection.prepareStatement(statement)
      try {
        stmt.execute()
      } finally {
        stmt.close()
      }
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.gabry.weixin.utils

import java.sql.{Connection, DriverManager}

import com.jolbox.bonecp.{BoneCP, BoneCPConfig}
import com.typesafe.config.ConfigFactory

import scala.collection.mutable


object ConnectionPool {
  private val connectionPool = {
    val connectionPoolList = mutable.Map[String,BoneCP]()
    val connectionPoolNameList=ConfigFactory.load("WeiXin.conf").getStringList("ConnectionPool.config-name")
    connectionPoolNameList.toArray.foreach{ configName=>{
      try{
        val boneConfig = new BoneCPConfig(configName.toString)
        connectionPoolList += (configName.toString -> new BoneCP(boneConfig))
      } catch {
        case exception:Exception=>
          exception.printStackTrace()
      }
    } }
    connectionPoolList
  }

  def getConnection(connectionId:String):Option[Connection] ={
    connectionPool(connectionId.toString) match {
      case connPool:BoneCP => Some(connPool.getConnection)
      case _ => None
    }
  }

  def closeConnection(connection:Connection): Unit = {
    if(connection!=null && !connection.isClosed) connection.close()
  }
} 

//设置package包名称以及导入依赖的类
package us.illyohs.bansdb.util

import javax.sql.DataSource
import java.sql.Connection
import java.sql.SQLException
import java.sql.Statement
import java.util.UUID

import org.spongepowered.api.Sponge
import org.spongepowered.api.service.sql.SqlService
import us.illyohs.bansdb.util.ConfigUtil._


object DBUtil {
  private var sqlServ: SqlService = null
  private var ds: DataSource = null

  def execute(execute: String): Unit = {
    sqlServ = Sponge.getServiceManager.provide(classOf[SqlService]).get()
    val URL: String = "jdbc:mysql://" + getURL + ":" + getPort + "/" + "?user=" + getUsername + "&password=" + getPass
    try {
      ds = sqlServ.getDataSource(URL)
      val connection: Connection = ds.getConnection
      val statement: Statement = connection.createStatement
      statement.execute(execute)
      statement.close
      connection.close
    }
    catch {
      case e: SQLException => e.printStackTrace
    }
  }

  def init: Unit = {
    execute("CREATE TABLE IF NOT EXISTS WARN_DB (warnee TEXT, sender TEXT, warnings SMALLINT, reason TEXT)")
    execute("CREATE TABLE IF NOT EXISTS BAN_DB (banee TEXT, sender TEXT, reason)")
  }


  def getWarning: Int = {
    null
  }

  def isPlayerBanned(uUID: UUID): Boolean = {
    return false
  }

  def isPlayerMuted(): Boolean = {
    return false
  }
} 

//设置package包名称以及导入依赖的类
package mon4all.db

import java.sql.{Connection, ResultSet, SQLException, Statement}

object Helpers {

  def getCurrentTimeStamp = {
		val today = new java.util.Date()
		new java.sql.Timestamp(today.getTime())
	}

  def printSQLException(e: SQLException) {
    val msg: String =
      "SQL State: " + e.getSQLState + "\n" +
      "Message: " + e.getMessage + "\n\n"
    println(msg)
  }

  def executeSQL(sql: String, stmt: Statement, ignore: Set[String] = Set()) {
    try stmt.execute(sql)
    catch {
      case e: SQLException =>
        if(ignore(e.getSQLState())) println("SQLException ignored")
        else printSQLException(e)
    }
  }

  
  def createTables(sql: String*)(implicit conn: Connection) {
    sql foreach(s => executeSQL(s, conn.createStatement, Set("X0Y32")))
  }

  def dropTables(tables: String*)(implicit conn: Connection) {
    tables foreach(table => executeSQL("DROP TABLE " + table, conn.createStatement))
  }


  trait ResultSetToList {
    def toList[T](retrieve: ResultSet => T): List[T]
  }
  implicit def wrapResultSet(rs: ResultSet) = new ResultSetToList {
    def toList[T](retrieve: ResultSet => T): List[T] =
      new Iterator[T] {
        def hasNext = rs.next()
        def next() = retrieve(rs)
      }
      .toList
  }

} 

//设置package包名称以及导入依赖的类
package com.example.lagompractice.impl

import java.sql.Connection

import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, ReadSideProcessor}
import com.lightbend.lagom.scaladsl.persistence.jdbc.{JdbcReadSide, JdbcSession}

import scala.concurrent.ExecutionContext

class MessageRepository(session: JdbcSession)(implicit ec: ExecutionContext) {

}

class MessageEventProcessor(readSide: JdbcReadSide)(implicit ec: ExecutionContext) extends ReadSideProcessor[MessageEvent] {

  import JdbcSession.tryWith

  override def aggregateTags = MessageEvent.Tag.allTags

  override def buildHandler() = {
    readSide.builder[MessageEvent]("messageEventOffset")
      .setGlobalPrepare(createTables)
      .setEventHandler[MessageCreated](processMessageChanged)
      .build()
  }

  private def createTables(connection: Connection): Unit = {
    tryWith(connection.prepareStatement("""
      CREATE TABLE IF NOT EXISTS messages (
        messageId varchar(36),
        message TEXT,
        PRIMARY KEY (messageId)
      )
    """))(_.execute())
  }

  private def processMessageChanged(connection: Connection, eventElement: EventStreamElement[MessageCreated]): Unit = {
    tryWith(connection.prepareStatement("INSERT INTO messages (messageId, message) VALUES (?, ?)")) { ps =>
      ps.setString(1, eventElement.event.message.id.toString)
      ps.setString(2, eventElement.event.message.message)
      ps.execute()
    }
  }

  /*
  private def createTables(connection: Connection): Unit = {
    tryWith(connection.prepareStatement("INSERT INTO messages (message) VALUES (?)")) { ps =>
      ps.execute()
    }
  }
  */
} 

//设置package包名称以及导入依赖的类
package yumi.pipeline.sinks

import java.sql.{Connection, PreparedStatement}

import org.apache.spark.sql.jdbc.JdbcDialect
import org.apache.spark.sql.types._

class PostgreSqlSink extends DatabaseSink with Serializable {

  override def bulkInsertStatement (conn: Connection,
                                    table: String,
                                    rddSchema: StructType,
                                    dialect: JdbcDialect,
                                    rowCount: Int) : PreparedStatement = {
    val columns = rddSchema.fields.map(x => dialect.quoteIdentifier(x.name)).mkString(",")
    val parameters = rddSchema.fields.map(_ => "?").mkString("(", ",", ")")
    val sql = new StringBuilder(s"INSERT INTO $table($columns) VALUES ")

    (1 to rowCount).foldLeft(false) { (prependComma, index) =>

      if (prependComma) {
        sql.append(", ")
      }

      sql.append(parameters)

      true
    }

    conn.prepareStatement(sql.toString())
  }
}