-
Notifications
You must be signed in to change notification settings - Fork 347
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
16 changed files
with
709 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
85 changes: 85 additions & 0 deletions
85
quill-cassandra-ce/src/main/scala-2.12/io/getquill/CassandraCeContext.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| package io.getquill | ||
|
|
||
| import cats._ | ||
| import cats.effect._ | ||
| import com.datastax.oss.driver.api.core.cql.Row | ||
| import cats.syntax.all._ | ||
| import com.datastax.oss.driver.api.core.CqlSession | ||
| import com.datastax.oss.driver.api.core.cql.AsyncResultSet | ||
| import fs2.{ Chunk, Stream } | ||
| import com.typesafe.config.Config | ||
| import io.getquill.context.cassandra.CqlIdiom | ||
| import io.getquill.util.{ ContextLogger, LoadConfig } | ||
| import io.getquill.context.ExecutionInfo | ||
| import io.getquill.context.ce.CeContext | ||
|
|
||
| import scala.jdk.CollectionConverters._ | ||
| import scala.language.higherKinds | ||
|
|
||
| class CassandraCeContext[N <: NamingStrategy, F[_]]( | ||
| naming: N, | ||
| session: CqlSession, | ||
| preparedStatementCacheSize: Long | ||
| )(implicit val af: Async[F]) | ||
| extends CassandraCqlSessionContext[N](naming, session, preparedStatementCacheSize) | ||
| with CeContext[CqlIdiom, N, F] { | ||
|
|
||
| private val logger = ContextLogger(classOf[CassandraCeContext[_, F]]) | ||
|
|
||
| private[getquill] def prepareRowAndLog(cql: String, prepare: Prepare = identityPrepare): F[PrepareRow] = for { | ||
| ec <- Async[F].executionContext | ||
| futureStatement = Sync[F].delay(prepareAsync(cql)(ec)) | ||
| prepStatement <- Async[F].fromFuture(futureStatement) | ||
| (params, bs) = prepare(prepStatement, this) | ||
| _ <- Sync[F].delay(logger.logQuery(cql, params)) | ||
| } yield bs | ||
|
|
||
| protected def page(rs: AsyncResultSet): Stream[F, Row] = | ||
| Stream.unfoldChunkEval(rs.remaining())(rem => | ||
| if (rem > 0) | ||
| af.delay[Option[(Chunk[Row], Int)]] { | ||
| val chunk: Chunk[Row] = Chunk.iterable(rs.currentPage().asScala) | ||
| Some((chunk, rs.remaining())) | ||
| } | ||
| else | ||
| af.pure[Option[(Chunk[Row], Int)]](None)) | ||
|
|
||
| def streamQuery[T](cql: String, prepare: Prepare = identityPrepare, extractor: Extractor[T] = identityExtractor)(info: ExecutionInfo, dc: Runner): StreamResult[T] = { | ||
| Stream | ||
| .eval(prepareRowAndLog(cql, prepare)) | ||
| .evalMap(p => af.fromCompletableFuture(af.delay(session.executeAsync(p).toCompletableFuture))) | ||
| .flatMap(page) | ||
| .map(it => extractor(it, this)) | ||
| } | ||
|
|
||
| def executeQuery[T](cql: String, prepare: Prepare = identityPrepare, extractor: Extractor[T] = identityExtractor)(info: ExecutionInfo, dc: Runner): Result[RunQueryResult[T]] = | ||
| streamQuery[T](cql, prepare, extractor)(info, dc).compile.toList | ||
|
|
||
| def executeQuerySingle[T](cql: String, prepare: Prepare = identityPrepare, extractor: Extractor[T] = identityExtractor)(info: ExecutionInfo, dc: Runner): Result[RunQuerySingleResult[T]] = | ||
| Functor[F].map(executeQuery(cql, prepare, extractor)(info, dc))(handleSingleResult) | ||
|
|
||
| def executeAction(cql: String, prepare: Prepare = identityPrepare)(info: ExecutionInfo, dc: Runner): Result[RunActionResult] = { | ||
| prepareRowAndLog(cql, prepare) | ||
| .flatMap(r => af.fromCompletableFuture(af.delay(session.executeAsync(r).toCompletableFuture))) | ||
| .map(_ => ()) | ||
| } | ||
|
|
||
| def executeBatchAction(groups: List[BatchGroup])(info: ExecutionInfo, dc: Runner): Result[RunBatchActionResult] = | ||
| groups.traverse_ { | ||
| case BatchGroup(cql, prepare) => | ||
| prepare.traverse_(executeAction(cql, _)(info, dc)) | ||
| } | ||
| } | ||
|
|
||
| object CassandraCeContext { | ||
|
|
||
| def apply[N <: NamingStrategy, F[_]: Async: FlatMap](naming: N, config: CassandraContextConfig): CassandraCeContext[N, F] = | ||
| new CassandraCeContext(naming, config.session, config.preparedStatementCacheSize) | ||
|
|
||
| def apply[N <: NamingStrategy, F[_]: Async: FlatMap](naming: N, config: Config): CassandraCeContext[N, F] = | ||
| CassandraCeContext(naming, CassandraContextConfig(config)) | ||
|
|
||
| def apply[N <: NamingStrategy, F[_]: Async: FlatMap](naming: N, configPrefix: String): CassandraCeContext[N, F] = | ||
| CassandraCeContext(naming, LoadConfig(configPrefix)) | ||
|
|
||
| } |
83 changes: 83 additions & 0 deletions
83
quill-cassandra-ce/src/main/scala-2.13/io/getquill/CassandraCeContext.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| package io.getquill | ||
|
|
||
| import cats._ | ||
| import cats.effect._ | ||
| import cats.syntax.all._ | ||
| import fs2.{Chunk, Stream} | ||
| import com.typesafe.config.Config | ||
| import com.datastax.oss.driver.api.core.cql.Row | ||
| import com.datastax.oss.driver.api.core.CqlSession | ||
| import com.datastax.oss.driver.api.core.cql.AsyncResultSet | ||
| import io.getquill.context.cassandra.CqlIdiom | ||
| import io.getquill.util.{ContextLogger, LoadConfig} | ||
| import io.getquill.context.ExecutionInfo | ||
| import io.getquill.context.ce.CeContext | ||
|
|
||
| import scala.jdk.CollectionConverters._ | ||
| import scala.language.higherKinds | ||
|
|
||
| class CassandraCeContext[N <: NamingStrategy, F[_]]( | ||
| naming: N, | ||
| session: CqlSession, | ||
| preparedStatementCacheSize: Long | ||
| )(implicit val af: Async[F]) | ||
| extends CassandraCqlSessionContext[N](naming, session, preparedStatementCacheSize) | ||
| with CeContext[CqlIdiom, N, F] { | ||
|
|
||
| private val logger = ContextLogger(classOf[CassandraCeContext[_, F]]) | ||
|
|
||
| private[getquill] def prepareRowAndLog(cql: String, prepare: Prepare = identityPrepare): F[PrepareRow] = for { | ||
| ec <- Async[F].executionContext | ||
| futureStatement = Sync[F].delay(prepareAsync(cql)(ec)) | ||
| prepStatement <- Async[F].fromFuture(futureStatement) | ||
| (params, bs) = prepare(prepStatement, this) | ||
| _ <- Sync[F].delay(logger.logQuery(cql, params)) | ||
| } yield bs | ||
|
|
||
| protected def page(rs: AsyncResultSet): Stream[F, Row] = | ||
| Stream.unfoldChunkEval(rs.remaining())(rem => | ||
| if (rem > 0) | ||
| af.delay(Some(Chunk.iterable(rs.currentPage().asScala), rs.remaining())) | ||
| else | ||
| af.pure(None) | ||
| ) | ||
|
|
||
| def streamQuery[T](cql: String, prepare: Prepare = identityPrepare, extractor: Extractor[T] = identityExtractor)(info: ExecutionInfo, dc: Runner): StreamResult[T] = { | ||
| Stream | ||
| .eval(prepareRowAndLog(cql, prepare)) | ||
| .evalMap(p => af.fromCompletableFuture(af.delay(session.executeAsync(p).toCompletableFuture))) | ||
| .flatMap(page) | ||
| .map(it => extractor(it, this)) | ||
| } | ||
|
|
||
| def executeQuery[T](cql: String, prepare: Prepare = identityPrepare, extractor: Extractor[T] = identityExtractor)(info: ExecutionInfo, dc: Runner): Result[RunQueryResult[T]] = | ||
| streamQuery[T](cql, prepare, extractor)(info, dc).compile.toList | ||
|
|
||
| def executeQuerySingle[T](cql: String, prepare: Prepare = identityPrepare, extractor: Extractor[T] = identityExtractor)(info: ExecutionInfo, dc: Runner): Result[RunQuerySingleResult[T]] = | ||
| Functor[F].map(executeQuery(cql, prepare, extractor)(info, dc))(handleSingleResult) | ||
|
|
||
| def executeAction(cql: String, prepare: Prepare = identityPrepare)(info: ExecutionInfo, dc: Runner): Result[RunActionResult] = { | ||
| prepareRowAndLog(cql, prepare) | ||
| .flatMap(r => af.fromCompletableFuture(af.delay(session.executeAsync(r).toCompletableFuture))) | ||
| .map(_ => ()) | ||
| } | ||
|
|
||
| def executeBatchAction(groups: List[BatchGroup])(info: ExecutionInfo, dc: Runner): Result[RunBatchActionResult] = | ||
| groups.traverse_ { | ||
| case BatchGroup(cql, prepare) => | ||
| prepare.traverse_(executeAction(cql, _)(info, dc)) | ||
| } | ||
| } | ||
|
|
||
| object CassandraCeContext { | ||
|
|
||
| def apply[N <: NamingStrategy, F[_] : Async : FlatMap](naming: N, config: CassandraContextConfig): CassandraCeContext[N, F] = | ||
| new CassandraCeContext(naming, config.session, config.preparedStatementCacheSize) | ||
|
|
||
| def apply[N <: NamingStrategy, F[_] : Async : FlatMap](naming: N, config: Config): CassandraCeContext[N, F] = | ||
| CassandraCeContext(naming, CassandraContextConfig(config)) | ||
|
|
||
| def apply[N <: NamingStrategy, F[_] : Async : FlatMap](naming: N, configPrefix: String): CassandraCeContext[N, F] = | ||
| CassandraCeContext(naming, LoadConfig(configPrefix)) | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| testStreamDB { | ||
| preparedStatementCacheSize=1 | ||
| keyspace=quill_test | ||
|
|
||
| session { | ||
| basic.contact-points = [ ${?CASSANDRA_CONTACT_POINT_0}, ${?CASSANDRA_CONTACT_POINT_1} ] | ||
| basic.load-balancing-policy.local-datacenter = ${?CASSANDRA_DC} | ||
| basic.request.consistency = LOCAL_QUORUM | ||
| basic.request.page-size = 999 | ||
| } | ||
|
|
||
| } |
36 changes: 36 additions & 0 deletions
36
...andra-ce/src/test/scala-2.12/io/getquill/context/cassandra/catEffect/DecodeNullSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| package io.getquill.context.cassandra.catEffect | ||
|
|
||
| import io.getquill._ | ||
|
|
||
| class DecodeNullSpec extends Spec { | ||
|
|
||
| "no default values when reading null" - { | ||
| "stream" in { | ||
| import io.getquill.context.cassandra.catsEffect.testCeDB._ | ||
| import io.getquill.context.cassandra.catsEffect.testCeDB | ||
| import cats.effect.unsafe.implicits.global | ||
|
|
||
| val writeEntities = quote(querySchema[DecodeNullTestWriteEntity]("DecodeNullTestEntity")) | ||
|
|
||
| val result = | ||
| for { | ||
| _ <- testCeDB.run(writeEntities.delete) | ||
| _ <- testCeDB.run(writeEntities.insert(lift(insertValue))) | ||
| result <- testCeDB.run(query[DecodeNullTestEntity]) | ||
| } yield { | ||
| result | ||
| } | ||
| intercept[IllegalStateException] { | ||
| await { | ||
| result.unsafeToFuture() | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| case class DecodeNullTestEntity(id: Int, value: Int) | ||
|
|
||
| case class DecodeNullTestWriteEntity(id: Int, value: Option[Int]) | ||
|
|
||
| val insertValue = DecodeNullTestWriteEntity(0, None) | ||
| } |
44 changes: 44 additions & 0 deletions
44
...ssandra-ce/src/test/scala-2.12/io/getquill/context/cassandra/catEffect/EncodingSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| package io.getquill.context.cassandra.catEffect | ||
|
|
||
| import io.getquill.Query | ||
| import io.getquill.context.cassandra.EncodingSpecHelper | ||
| import io.getquill.context.cassandra.catsEffect.testCeDB._ | ||
| import io.getquill.context.cassandra.catsEffect.testCeDB | ||
| import cats.effect.unsafe.implicits.global | ||
|
|
||
| class EncodingSpec extends EncodingSpecHelper { | ||
| "encodes and decodes types" - { | ||
| "stream" in { | ||
| val result = | ||
| for { | ||
| _ <- testCeDB.run(query[EncodingTestEntity].delete) | ||
| _ <- testCeDB.run(liftQuery(insertValues).foreach(e => query[EncodingTestEntity].insert(e))) | ||
| result <- testCeDB.run(query[EncodingTestEntity]) | ||
| } yield { | ||
| result | ||
| } | ||
| val f = result.unsafeToFuture() | ||
| val r = await(f) | ||
| verify(r) | ||
| } | ||
| } | ||
|
|
||
| "encodes collections" - { | ||
| "stream" in { | ||
| val q = quote { | ||
| (list: Query[Int]) => | ||
| query[EncodingTestEntity].filter(t => list.contains(t.id)) | ||
| } | ||
| val result = | ||
| for { | ||
| _ <- testCeDB.run(query[EncodingTestEntity].delete) | ||
| _ <- testCeDB.run(liftQuery(insertValues).foreach(e => query[EncodingTestEntity].insert(e))) | ||
| result <- testCeDB.run(q(liftQuery(insertValues.map(_.id)))) | ||
| } yield { | ||
| result | ||
| } | ||
| val f = result.unsafeToFuture() | ||
| verify(await(f)) | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.