//设置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 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 models

import java.sql.ResultSet

import scalikejdbc._
import skinny.orm.{Alias, SkinnyCRUDMapperWithId}

import scala.util.Random

case class Vote(
    id: VoteId,
    destId: DestId,
    sessionId: SessionId,
    ip: Long,
    created: Long
)

case class VoteId(value: Long) extends AnyVal {
  override def toString = value.toString
}

object VoteId {
  val typeBinder: TypeBinder[VoteId] = new TypeBinder[VoteId] {
    override def apply(rs: ResultSet, columnIndex: Int): VoteId = VoteId(rs.getLong(columnIndex))
    override def apply(rs: ResultSet, columnLabel: String): VoteId = VoteId(rs.getLong(columnLabel))
  }
}

object Vote extends SkinnyCRUDMapperWithId[VoteId, Vote] {
  implicit def voteIdTypeBinder = VoteId.typeBinder
  implicit def destIdTypeBinder = DestId.typeBinder
  implicit def sessionIdTypeBinder = SessionId.typeBinder
  override val useExternalIdGenerator = true

  override def idToRawValue(id: VoteId): Any = id.value
  override def rawValueToId(value: Any): VoteId = VoteId(value.toString.toLong)

  override def generateId(): VoteId = VoteId(Random.nextLong())

  override def defaultAlias: Alias[Vote] = createAlias("v")

  override def extract(rs: WrappedResultSet, n: ResultName[Vote]): Vote = autoConstruct(rs, n)

  def create(destId: DestId, sessionId: SessionId, ip: Long)(implicit db: DBSession) =
    createWithAttributes(
      'id -> generateId(),
      'destId -> destId.value,
      'sessionId -> sessionId.value,
      'ip -> ip,
      'created -> System.currentTimeMillis()
    )
} 

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

import java.sql.ResultSet

import scalikejdbc._
import skinny.orm.{Alias, SkinnyCRUDMapperWithId}

case class Session(id: SessionId, created: Long) {
  def save()(implicit session: DBSession) = Session.create(this)
}

case class SessionId(value: Long) extends AnyVal {
  override def toString: String = value.toString
}

object SessionId {
  val typeBinder: TypeBinder[SessionId] = new TypeBinder[SessionId] {
    override def apply(rs: ResultSet, columnIndex: Int): SessionId = SessionId(rs.getLong(columnIndex))
    override def apply(rs: ResultSet, columnLabel: String): SessionId = SessionId(rs.getLong(columnLabel))
  }
}

object Session extends SkinnyCRUDMapperWithId[SessionId, Session] {
  implicit def sessionIdTypeBinder = SessionId.typeBinder
  override val useExternalIdGenerator = true

  override def idToRawValue(id: SessionId): Any = id.value
  override def rawValueToId(value: Any): SessionId = SessionId(value.toString.toLong)

  override def defaultAlias: Alias[Session] = createAlias("s")

  override def extract(rs: WrappedResultSet, n: ResultName[Session]): Session = autoConstruct(rs, n)

  def create(session: Session)(implicit db: DBSession) =
    createWithAttributes('id -> session.id, 'created -> session.created)
} 

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

import java.sql.{DriverManager, ResultSet}
import org.apache.log4j.Logger
import com.datawizards.dbtable2class.model.ColumnMetadata
import scala.collection.mutable.ListBuffer

trait Dialect {
  protected val log: Logger = Logger.getLogger(getClass.getName)

  def mapColumnTypeToScalaType(column: ColumnMetadata): String

  def extractTableColumns(dbUrl: String, connectionProperties: java.util.Properties, database: String, schema: String, table: String): Iterable[ColumnMetadata] = {
    Class.forName(driverClassName)
    val connection = DriverManager.getConnection(dbUrl, connectionProperties)
    val query = extractTableColumnsQuery(database, schema, table)
    val rs = connection.createStatement().executeQuery(query)
    val buffer = new ListBuffer[ColumnMetadata]
    while(rs.next()) {
      buffer += extractColumnMetadata(rs)
    }
    connection.close()

    val extractedColumns = buffer.toList
    if(extractedColumns.size == 0){
      throw new RuntimeException(s"No columns found for specified table ${database}.${schema}.${table}. Please check if table is correct.")
    }
    extractedColumns
  }

  protected def extractColumnMetadata(rs: ResultSet): ColumnMetadata =
    ColumnMetadata(
      columnName = rs.getString(columnWithColumnName),
      typeName = rs.getString(columnWithTypeName)
    )

  protected def driverClassName: String
  protected def extractTableColumnsQuery(database: String, schema: String, table: String): String
  protected def columnWithColumnName: String
  protected def columnWithTypeName: String

} 

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

import java.sql.{ResultSet, SQLException}


  def objectOfResultSet[A] (resultSet : ResultSet, toObject : (ResultSet ? A)) : A =
  {
    if (resultSet.next())
      toObject(resultSet)
    else
    {
      throw new SQLException("objectOfResultSet: no result found.")
    }
  }
} 

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

import models.{Types, User}

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

object UserQuery {
  def create(
    connection : Connection,
    main_email : String,
    password : String,
    firstname : String,
    lastname : String,
    avatar : String
  ) : Types.userid = {
    val statement : PreparedStatement =
      connection.prepareStatement(
        "INSERT INTO users.user (" +
          "main_email, " +
          "password, " +
          "firstname, " +
          "lastname, " +
          "avatar) " +
          "VALUES(?, ?, ?, ?, ?)",
        Statement.RETURN_GENERATED_KEYS
      )

    statement.setString(0, main_email)
    statement.setString(1, password)
    statement.setString(2, firstname)
    statement.setString(3, lastname)
    statement.setString(4, avatar)

    val generatedKeys : ResultSet = statement.getGeneratedKeys()
    if (generatedKeys.next()) {
      generatedKeys.getLong(1)
    }
    else {
      throw new SQLException("Adding ad failed. No ID obtained.")
    }
  }
} 

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

import java.sql.Connection
import java.sql.ResultSet
import java.sql.SQLException
import java.sql.Statement
import com.boost.bigdata.utils.log.LogSupport

class Session(connection: Connection) extends LogSupport{

  private var conn: Connection = connection
  private var stmt: Statement = conn.createStatement(ResultSet.TYPE_SCROLL_SENSITIVE, ResultSet.CONCUR_UPDATABLE)

  def executeQuery(sql: String): ResultSet = {
    var rs: ResultSet = null
    try {
      if (stmt != null) {
        rs = stmt.executeQuery(sql)
      }
    }
    catch {
      case e: SQLException => {
        log.warn(e.getMessage)
      }
    }
    rs
  }

  def executeUpdate(sql: String): Boolean = {
    var ret: Boolean = false
    try {
      if (stmt != null) {
        stmt.executeUpdate(sql)
        ret = true
      }
    }
    catch {
      case e: SQLException => {
        log.warn(e.getMessage)
      }
    }
    ret
  }

  def disconnect {
    MySql.closeStatement(stmt)
    stmt = null
    MySql.closeConnection(conn)
    conn = null
  }
}

object Session{
  def apply(con: Connection) = new Session(con)
} 

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

import java.sql.ResultSet
import java.util.UUID

import org.skife.jdbi.v2.StatementContext
import org.skife.jdbi.v2.sqlobject.customizers.Mapper
import org.skife.jdbi.v2.sqlobject.{Bind, SqlQuery}
import org.skife.jdbi.v2.tweak.ResultSetMapper
import com.github.fcristovao.domain.{Customer, CustomerId}


trait CustomerDao {
  @SqlQuery("select * from customers where customer_id = :id limit 1")
  @Mapper(classOf[CustomerMapper])
  def findCustomerById(@Bind("id") id: CustomerId) : Customer
}


class CustomerMapper extends ResultSetMapper[Customer] {
  override def map(index: Int, r: ResultSet, ctx: StatementContext): Customer = {
    Customer(UUID.fromString(r.getString("customer_id")))
  }
} 

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

import java.sql.ResultSet

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import akka.util.ByteString
import org.scalatest.{AsyncFlatSpec, BeforeAndAfterAll, Matchers}
import ru.arigativa.akka.streams.ConnectionProvider._
import ru.arigativa.akka.streams.{PgCopySourceSettings, PgCopyStreamConverters}
import util.PostgresFixture


class CopySourceSpec extends AsyncFlatSpec with Matchers with PostgresFixture with BeforeAndAfterAll {

  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()
  override def afterAll(): Unit = {
    system.terminate()
  }

  def fetchPeople(rs: ResultSet): (Long, String, Int) = {
    rs.next()
    (rs.getLong(1), rs.getString(2), rs.getInt(3))
  }

  "PgCopySource" should "copy bytes as expected" in {
    withPostgres("people_filled") { conn =>
      PgCopyStreamConverters.bytesSource("COPY (SELECT id, name, age FROM people) TO STDOUT", PgCopySourceSettings(conn))
        .runWith(Sink.fold(ByteString.empty) {
          case (acc, next) => acc ++ next
        })
        .map { result =>
          result.decodeString("UTF-8") shouldBe (
            "1\tAlex\t26\n" +
            "2\tLisa\t22\n" +
            "3\tWith\\r\\n\\t special chars\\\\\t10\n" +
            "4\t\\N\t-1\n"
          )
        }
    }
  }

  "PgCopySource" should "copy lines as expected" in {
    withPostgres("people_filled") { conn =>
      PgCopyStreamConverters.source("COPY (SELECT id, name, age FROM people) TO STDOUT", PgCopySourceSettings(conn))
        .runWith(Sink.seq)
        .map { result =>
          result shouldBe Seq(
            Seq("1", "Alex", "26"),
            Seq("2", "Lisa", "22"),
            Seq("3", "With\r\n\t special chars\\", "10"),
            Seq("4", null, "-1")
          )
        }
    }
  }
} 

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

import scala.collection.mutable.ListBuffer
import java.sql.ResultSet

trait FromResultSet[A] {
  def toEntityList(rs: ResultSet): List[A] = {
    val buffer = new ListBuffer[A]
    while (rs.next) {
      buffer += convert(rs)
    }
    buffer.toList
  }

  def toEntity(rs: ResultSet): Option[A] =
    toEntityList(rs).headOption

  protected def convert(rs: ResultSet): A
}

object FromResultSet {

  def apply[A](implicit instance: FromResultSet[A]): FromResultSet[A] = instance
  private def build[A](f: ResultSet => A): FromResultSet[A] = new FromResultSet[A] {
    def convert(rs: ResultSet): A = f(rs)
  }

  implicit val petFromRs: FromResultSet[Pet] =
    build { rs =>
      Pet(
        Some(rs.getLong(PetRepo.Id)),
        rs.getString(PetRepo.Name),
        rs.getObject(PetRepo.BirthDate, classOf[java.time.LocalDate]),
        rs.getLong(PetRepo.TypeId),
        rs.getLong(PetRepo.OwnerId)
      )
    }

  implicit val petTypeFromRs: FromResultSet[PetType] =
    build { rs =>
      PetType(Some(rs.getLong(PetRepo.Id)), rs.getString(PetRepo.Name))
    }

  implicit val ownerFromRs: FromResultSet[Owner] =
    build { rs =>
      Owner(
        Some(rs.getLong(OwnerRepo.Id)),
        rs.getString(OwnerRepo.FirstName) + " " + rs.getString(OwnerRepo.LastName),
        rs.getString(OwnerRepo.FirstName),
        rs.getString(OwnerRepo.LastName),
        rs.getString(OwnerRepo.Address),
        rs.getString(OwnerRepo.City),
        rs.getString(OwnerRepo.Telephone)
      )
    }
} 

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

import java.sql.{Connection, ResultSet}

class RichConnection(val underlying:Connection) {

  
  def withQuery[T](query:String)(f:ResultSet => T):T = {
    val statement = underlying.prepareStatement(query)
    val results = statement.executeQuery
    try {
      f(results) // loan the ResultSet to the client
    }
    finally {
      // Ensure all the resources get freed.
      results.close
      statement.close
    }
  }
} 

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

import java.sql.ResultSet
import scala.collection.mutable.ListBuffer
import app.util.db.Postgre.Query

object Article {
  val NAME = "Article"
  val ID = "id"
  val TITLE = "title"
  val CONTENT = "content"

  def from(rs: ResultSet): Article =
    new Article(
      rs.getLong(ID),
      rs.getString(TITLE),
      rs.getString(CONTENT)
    )

  def create() =
    Query.execute(s"""
      CREATE TABLE IF NOT EXISTS $NAME(
        $ID BIGSERIAL PRIMARY KEY,
        $TITLE VARCHAR(1024),
        $CONTENT TEXT
      )
    """);

  def drop() = Query.execute(s"DROP TABLE $NAME");

  def select(): List[Article] = {
    val rs = Query.builder(s"SELECT * FROM $NAME").executeQuery()
    var list = ListBuffer.empty[Article]
    while (rs.next())
      list += Article.from(rs)
    list.toList
  }
}

class Article(var id: Long, var title: String, var content: String) {
  import Article._

  def insert() =
    Query.builder(s"INSERT INTO $NAME ($TITLE) VALUES (?)")
      .set(title)
      .executeUpdate()
} 

//设置package包名称以及导入依赖的类
package pl.scalare.impl.repo

import java.sql.ResultSet

import org.skife.jdbi.v2.StatementContext
import org.skife.jdbi.v2.sqlobject.customizers.RegisterMapper
import org.skife.jdbi.v2.sqlobject.{Bind, BindBean, SqlQuery, SqlUpdate}
import org.skife.jdbi.v2.tweak.ResultSetMapper
import pl.scalare.core.model.Event
import pl.scalare.core.repo.EventRepo

@RegisterMapper(Array(classOf[EventMapper]))
trait EventRepoImpl extends EventRepo {
  @SqlUpdate("create table events (id int primary key, uuid varchar, json varchar)")
  def createSomethingTable()

  @SqlUpdate("insert into events (id, uuid, json) values (:id, :uuid, :json)")
  def insert(@BindBean event: Event)

  @SqlUpdate("insert into events (id, uuid, json) values (:id, :uuid, :json)")
  def insert(@Bind("id") id: Int, @Bind("uuid") uuid: String, @Bind("json") json: String)

  @SqlQuery("select * from events")
  def getAll: List[Event]

  @SqlQuery("select * from events where id = :id")
  def findById(@Bind("id") id: Int): Event
}

class EventMapper extends ResultSetMapper[Event] {
  def map(index: Int, rs: ResultSet, ctx: StatementContext) = new Event(rs.getInt("id"), rs.getString("uuid"), rs.getString("json"))
} 

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

import org.apache.spark.Partition
import java.sql.ResultSet
import java.sql.Connection
import scala.reflect.ClassTag
import org.apache.spark.SparkContext
import java.sql.PreparedStatement
import org.apache.spark.rdd.RDD
import org.apache.spark.Logging
import org.apache.spark.TaskContext
import org.apache.spark.util

class OracleJdbcPartition(idx: Int, parameters: Map[String, Object]) extends Partition {
  override def index = idx
  val partitionParameters = parameters
}

abstract class OracleJdbcRdd[T:ClassTag] (
  sc:SparkContext,
  getConnection:() => Connection,
  sql:String,
  getOracleJdbcPatition :()=>Array[Partition],
  preparedStatement :(PreparedStatement,OracleJdbcPartition)=> PreparedStatement,
  mapRow :(ResultSet)=> T=OracleJdbcRdd.resultSetToObjectArray _)
  extends RDD[T](sc,Nil)with Logging{
  
  def getPartitions: Array[Partition] = {
    getOracleJdbcPatition();
  }
  
 
  

}

object OracleJdbcRdd {
  def resultSetToObjectArray(rs: ResultSet): Array[Object] = {
    Array.tabulate[Object](rs.getMetaData.getColumnCount)(i => rs.getObject(i + 1))
  }
  
   trait ConnectionFactory extends Serializable {
    @throws[Exception]
    def getConnection: Connection
  }
} 

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

import java.sql.{Connection, PreparedStatement, ResultSet}
import java.util.UUID
import dao.postgres.common.TaskTable
import model.{Task, TaskStatus}
import util.JdbcUtil._

object TaskMarshaller {

  def unmarshalTask(rs: ResultSet): Task = {
    import TaskTable._
    Task(
      id = rs.getObject(COL_ID).asInstanceOf[UUID],
      processId = rs.getObject(COL_PROCESS_ID).asInstanceOf[UUID],
      processDefinitionName = rs.getString(COL_PROC_DEF_NAME),
      taskDefinitionName = rs.getString(COL_TASK_DEF_NAME),
      executable = PostgresJsonMarshaller.toExecutable(rs.getString(COL_EXECUTABLE)),
      previousAttempts = rs.getInt(COL_ATTEMPTS),
      startedAt = javaDate(rs.getTimestamp(COL_STARTED)),
      status = rs.getString(COL_STATUS) match {
        case STATUS_SUCCEEDED => TaskStatus.Success(javaDate(rs.getTimestamp(COL_ENDED_AT)))
        case STATUS_FAILED => TaskStatus.Failure(javaDate(rs.getTimestamp(COL_ENDED_AT)), Option(rs.getString(COL_REASON)))
        case STATUS_RUNNING => TaskStatus.Running()
      }
    )
  }

  def marshalTask(task: Task, stmt: PreparedStatement, columns: Seq[String], startIndex: Int = 1)
                 (implicit conn: Connection) = {
    import TaskTable._
    var index = startIndex
    columns.foreach { col =>
      col match {
        case COL_ID => stmt.setObject(index, task.id)
        case COL_PROCESS_ID => stmt.setObject(index, task.processId)
        case COL_PROC_DEF_NAME => stmt.setString(index, task.processDefinitionName)
        case COL_TASK_DEF_NAME => stmt.setString(index, task.taskDefinitionName)
        case COL_EXECUTABLE => stmt.setString(index, PostgresJsonMarshaller.toJson(task.executable))
        case COL_ATTEMPTS => stmt.setInt(index, task.previousAttempts)
        case COL_STARTED => stmt.setTimestamp(index, task.startedAt)
        case COL_STATUS => stmt.setString(index, task.status match {
          case TaskStatus.Success(_) => STATUS_SUCCEEDED
          case TaskStatus.Failure(_, _) => STATUS_FAILED
          case TaskStatus.Running() => STATUS_RUNNING
        })
        case COL_REASON => stmt.setString(index, task.status match {
          case TaskStatus.Failure(_, reasons) => reasons.mkString(",")
          case _ => null
        })
        case COL_ENDED_AT => stmt.setTimestamp(index, task.endedAt.getOrElse(null))
      }
      index += 1
    }
  }

} 

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

import java.sql.{Connection, PreparedStatement, ResultSet}
import java.util.UUID

import dao.postgres.common.{ProcessTriggerRequestTable, TaskTriggerRequestTable}
import model.ProcessTriggerRequest
import util.JdbcUtil._

object ProcessTriggerRequestMarshaller {

  def marshal(request: ProcessTriggerRequest, stmt: PreparedStatement, columns: Seq[String], startIndex: Int = 1)
             (implicit conn: Connection) = {
    import ProcessTriggerRequestTable._
    var index = startIndex
    columns.foreach { col =>
      col match {
        case COL_REQUEST_ID => stmt.setObject(index, request.requestId)
        case COL_PROCESS_DEF_NAME => stmt.setString(index, request.processDefinitionName)
        case COL_REQUESTED_AT => stmt.setTimestamp(index, request.requestedAt)
        case COL_STARTED_PROCESS_ID => stmt.setObject(index, request.startedProcessId.orNull)
        case COL_TASK_FILTER => stmt.setArray(index, request.taskFilter.map(makeStringArray).orNull)
      }
      index += 1
    }
  }

  def unmarshal(rs: ResultSet): ProcessTriggerRequest = {
    import ProcessTriggerRequestTable._
    ProcessTriggerRequest(
      requestId = rs.getObject(COL_REQUEST_ID).asInstanceOf[UUID],
      processDefinitionName = rs.getString(COL_PROCESS_DEF_NAME),
      requestedAt = javaDate(rs.getTimestamp(COL_REQUESTED_AT)),
      startedProcessId = Option(rs.getObject(COL_STARTED_PROCESS_ID)).map(_.asInstanceOf[UUID]),
      taskFilter = getStringArray(rs, COL_TASK_FILTER)
    )
  }

} 

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

import java.sql.{Connection, PreparedStatement, ResultSet}
import java.util.UUID
import dao.postgres.common.TaskDefinitionTable
import model.{TaskBackoff, TaskDefinition, TaskDependencies, TaskLimits}
import util.JdbcUtil._

object TaskDefinitionMarshaller {

  def marshal(definition: TaskDefinition, stmt: PreparedStatement, columns: Seq[String], startIndex: Int = 1)
             (implicit conn: Connection) = {
    import TaskDefinitionTable._
    var index = startIndex
    columns.foreach { col =>
      col match {
        case COL_NAME => stmt.setString(index, definition.name)
        case COL_PROC_ID => stmt.setObject(index, definition.processId)
        case COL_EXECUTABLE => stmt.setString(index, PostgresJsonMarshaller.toJson(definition.executable))
        case COL_MAX_ATTEMPTS => stmt.setInt(index, definition.limits.maxAttempts)
        case COL_MAX_EXECUTION_TIME => stmt.setObject(index, definition.limits.maxExecutionTimeSeconds.orNull)
        case COL_BACKOFF_SECONDS => stmt.setInt(index, definition.backoff.seconds)
        case COL_BACKOFF_EXPONENT => stmt.setDouble(index, definition.backoff.exponent)
        case COL_REQUIRED_DEPS => stmt.setArray(index, makeStringArray(definition.dependencies.required))
        case COL_OPTIONAL_DEPS => stmt.setArray(index, makeStringArray(definition.dependencies.optional))
        case COL_REQUIRE_EXPLICIT_SUCCESS => stmt.setBoolean(index, definition.requireExplicitSuccess)
      }
      index += 1
    }
  }

  def unmarshal(rs: ResultSet): TaskDefinition = {
    import TaskDefinitionTable._
    TaskDefinition(
      name = rs.getString(COL_NAME),
      processId = rs.getObject(COL_PROC_ID).asInstanceOf[UUID],
      executable = PostgresJsonMarshaller.toExecutable(rs.getString(COL_EXECUTABLE)),
      limits = TaskLimits(
        maxAttempts = rs.getInt(COL_MAX_ATTEMPTS),
        maxExecutionTimeSeconds = getIntOption(rs, COL_MAX_EXECUTION_TIME)
      ),
      backoff = TaskBackoff(
        seconds = rs.getInt(COL_BACKOFF_SECONDS),
        exponent = rs.getDouble(COL_BACKOFF_EXPONENT)
      ),
      dependencies = TaskDependencies(
        required = getStringArray(rs, COL_REQUIRED_DEPS).getOrElse(Seq.empty),
        optional = getStringArray(rs, COL_OPTIONAL_DEPS).getOrElse(Seq.empty)
      ),
      requireExplicitSuccess = rs.getBoolean(COL_REQUIRE_EXPLICIT_SUCCESS)
    )
  }

} 

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

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

import dao.postgres.common.TaskDefinitionTemplateTable
import model._
import util.JdbcUtil._

object TaskDefinitionTemplateMarshaller {

  def marshal(definition: TaskDefinitionTemplate, stmt: PreparedStatement, columns: Seq[String], startIndex: Int = 1)
             (implicit conn: Connection) = {
    import TaskDefinitionTemplateTable._
    var index = startIndex
    columns.foreach { col =>
      col match {
        case COL_NAME => stmt.setString(index, definition.name)
        case COL_PROC_DEF_NAME => stmt.setString(index, definition.processDefinitionName)
        case COL_EXECUTABLE => stmt.setString(index, PostgresJsonMarshaller.toJson(definition.executable))
        case COL_MAX_ATTEMPTS => stmt.setInt(index, definition.limits.maxAttempts)
        case COL_MAX_EXECUTION_TIME => stmt.setObject(index, definition.limits.maxExecutionTimeSeconds.orNull)
        case COL_BACKOFF_SECONDS => stmt.setInt(index, definition.backoff.seconds)
        case COL_BACKOFF_EXPONENT => stmt.setDouble(index, definition.backoff.exponent)
        case COL_REQUIRED_DEPS => stmt.setArray(index, makeStringArray(definition.dependencies.required))
        case COL_OPTIONAL_DEPS => stmt.setArray(index, makeStringArray(definition.dependencies.optional))
        case COL_REQUIRE_EXPLICIT_SUCCESS => stmt.setBoolean(index, definition.requireExplicitSuccess)
      }
      index += 1
    }
  }

  def unmarshal(rs: ResultSet): TaskDefinitionTemplate = {
    import TaskDefinitionTemplateTable._
    TaskDefinitionTemplate(
      name = rs.getString(COL_NAME),
      processDefinitionName = rs.getString(COL_PROC_DEF_NAME),
      executable = PostgresJsonMarshaller.toExecutable(rs.getString(COL_EXECUTABLE)),
      limits = TaskLimits(
        maxAttempts = rs.getInt(COL_MAX_ATTEMPTS),
        maxExecutionTimeSeconds = getIntOption(rs, COL_MAX_EXECUTION_TIME)
      ),
      backoff = TaskBackoff(
        seconds = rs.getInt(COL_BACKOFF_SECONDS),
        exponent = rs.getDouble(COL_BACKOFF_EXPONENT)
      ),
      dependencies = TaskDependencies(
        required = getStringArray(rs, COL_REQUIRED_DEPS).getOrElse(Seq.empty),
        optional = getStringArray(rs, COL_OPTIONAL_DEPS).getOrElse(Seq.empty)
      ),
      requireExplicitSuccess = rs.getBoolean(COL_REQUIRE_EXPLICIT_SUCCESS)
    )
  }

}