我想在apache Flink的AsyncFunction中设置doobie,但它需要一个隐式的读取输出实例。另一方面,我不能发送Read[Out]作为一个隐式参数,因为AsyncFunction需要是可序列化的,这个隐式值违反了这个规则:
package FlinkHelpers
import Configs.JdbConfig
import doobie.hikari._
import doobie._
import doobie.implicits._
import doobie.util.ExecutionContexts
import cats._
import cats.data._
import cats.effect.IO.asyncForIO
import cats.effect._
import cats.implicits._
import fs2.Stream
import cats.effect.unsafe.implicits.global
import org.apache.flink.runtime.concurrent.Executors
import org.apache.flink.streaming.api.scala.async.{AsyncFunction, ResultFuture}
import scala.concurrent.{ExecutionContext, Future}
import scala.util.Try
class AsyncDatabaseRequestOne[In,Out](configs:JdbConfig)(implicit r:Read[Out]) extends AsyncFunction[In,(In, Out)] {
lazy val transactor: Resource[IO, HikariTransactor[IO]] = {
for {
ce <- ExecutionContexts.fixedThreadPool[IO](32)
xa <- HikariTransactor.newHikariTransactor[IO](
"org.postgresql.Driver",
"jdbc:postgresql://localhost:5433/doobie",
"psql",
"110271",
ce
)
} yield xa
}
/** The context used for the future callbacks */
implicit lazy val executor: ExecutionContext = ExecutionContext.fromExecutor(Executors.directExecutor())
override def asyncInvoke(input: In, resultFuture: ResultFuture[(In, Out)]): Unit = {
val v2: IO[Out] = transactor.use[Out] { xa2: HikariTransactor[IO] => {
sql"select code,name,population,gnp from Out limit 1"
.query[Out] // Query0[String]
.unique // ConnectionIO[List[String]]
.transact(xa2) // IO[List[String]]
}
}
val future = v2.unsafeToFuture()
future.onSuccess({
case result: Out => resultFuture.complete(List((input, result)))
})
}
}
另一方面,scala不允许我使r
字段过渡有人知道怎么解这个吗?
class AsyncDatabaseRequestOne[In,Out](configs:JdbConfig)(r:Read[Out])...
中的r
是一个构造函数参数,但是Scala也可以为它生成private[this]
字段:
消除与父类的类字段同名的构造函数参数的歧义(answer)
尝试添加注释@transient
class AsyncDatabaseRequestOne[In,Out](configs:JdbConfig)(@transient private[this] implicit val r:Read[Out])...
或
class AsyncDatabaseRequestOne[In,Out](configs:JdbConfig)(@(transient @field) private[this] implicit val r:Read[Out])...
因为@transient
已经被@field
定义了
@field
final class transient extends scala.annotation.StaticAnnotation
我想在我们的情况下,@field
元注释是不必要的。