From 5713e1463b62bbc1a05c940918826632b56c13dd Mon Sep 17 00:00:00 2001 From: Miles Sabin Date: Sat, 16 Nov 2024 11:04:12 +0000 Subject: [PATCH] Revert SkunkMapping to initialising with a session resource --- .../src/test/scala/SkunkDatabaseSuite.scala | 15 ++-- .../js-jvm/src/test/scala/SkunkSuites.scala | 78 +++++++++---------- .../subscription/SubscriptionMapping.scala | 9 ++- .../subscription/SubscriptionSuite.scala | 6 +- .../shared/src/main/scala/SkunkMapping.scala | 17 ++-- .../main/scala/SkunkMappingCompanion.scala | 6 +- 6 files changed, 66 insertions(+), 65 deletions(-) diff --git a/modules/skunk/js-jvm/src/test/scala/SkunkDatabaseSuite.scala b/modules/skunk/js-jvm/src/test/scala/SkunkDatabaseSuite.scala index de2267e7..ecbe37bc 100644 --- a/modules/skunk/js-jvm/src/test/scala/SkunkDatabaseSuite.scala +++ b/modules/skunk/js-jvm/src/test/scala/SkunkDatabaseSuite.scala @@ -33,27 +33,28 @@ import grackle.sqlpg.test._ trait SkunkDatabaseSuite extends SqlPgDatabaseSuite { - def sessionResource: Resource[IO, Session[IO]] = { + def poolResource: Resource[IO, Resource[IO, Session[IO]]] = { val connInfo = postgresConnectionInfo import connInfo._ - Session.single[IO]( + Session.pooled[IO]( host = host, port = port, user = username, password = Some(password), database = databaseName, + max = 3, //debug = true, ) } - val sessionFixture: IOFixture[Session[IO]] = ResourceSuiteLocalFixture("skunk", sessionResource) - override def munitFixtures: Seq[IOFixture[_]] = Seq(sessionFixture) + val poolFixture: IOFixture[Resource[IO, Session[IO]]] = ResourceSuiteLocalFixture("skunk", poolResource) + override def munitFixtures: Seq[IOFixture[_]] = Seq(poolFixture) - def session: Session[IO] = sessionFixture() + def pool: Resource[IO, Session[IO]] = poolFixture() - abstract class SkunkTestMapping[F[_]: Sync](session: Session[F], monitor: SkunkMonitor[F] = SkunkMonitor.noopMonitor[IO]) - extends SkunkMapping[F](session, monitor) with SqlTestMapping[F] { + abstract class SkunkTestMapping[F[_]: Sync](pool: Resource[F, Session[F]], monitor: SkunkMonitor[F] = SkunkMonitor.noopMonitor[IO]) + extends SkunkMapping[F](pool, monitor) with SqlTestMapping[F] { type TestCodec[T] = (SCodec[T], Boolean) diff --git a/modules/skunk/js-jvm/src/test/scala/SkunkSuites.scala b/modules/skunk/js-jvm/src/test/scala/SkunkSuites.scala index 6d749833..8e5a3a87 100644 --- a/modules/skunk/js-jvm/src/test/scala/SkunkSuites.scala +++ b/modules/skunk/js-jvm/src/test/scala/SkunkSuites.scala @@ -30,61 +30,61 @@ import grackle.Mapping import org.typelevel.twiddles._ final class ArrayJoinSuite extends SkunkDatabaseSuite with SqlArrayJoinSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlArrayJoinMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlArrayJoinMapping[IO] } final class CoalesceSuite extends SkunkDatabaseSuite with SqlCoalesceSuite { type Fragment = skunk.AppliedFragment def mapping: IO[(Mapping[IO], SqlStatsMonitor[IO,Fragment])] = - SkunkMonitor.statsMonitor[IO].map(mon => (new SkunkTestMapping(session, mon) with SqlCoalesceMapping[IO], mon)) + SkunkMonitor.statsMonitor[IO].map(mon => (new SkunkTestMapping(pool, mon) with SqlCoalesceMapping[IO], mon)) } final class ComposedWorldSuite extends SkunkDatabaseSuite with SqlComposedWorldSuite { def mapping: IO[(CurrencyMapping[IO], Mapping[IO])] = for { currencyMapping <- CurrencyMapping[IO] - } yield (currencyMapping, new SqlComposedMapping(new SkunkTestMapping(session) with SqlWorldMapping[IO], currencyMapping)) + } yield (currencyMapping, new SqlComposedMapping(new SkunkTestMapping(pool) with SqlWorldMapping[IO], currencyMapping)) } final class CompositeKeySuite extends SkunkDatabaseSuite with SqlCompositeKeySuite { - lazy val mapping = new SkunkTestMapping(session) with SqlCompositeKeyMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlCompositeKeyMapping[IO] } final class CursorJsonSuite extends SkunkDatabaseSuite with SqlCursorJsonSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlCursorJsonMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlCursorJsonMapping[IO] } final class EmbeddingSuite extends SkunkDatabaseSuite with SqlEmbeddingSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlEmbeddingMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlEmbeddingMapping[IO] } final class Embedding2Suite extends SkunkDatabaseSuite with SqlEmbedding2Suite { - lazy val mapping = new SkunkTestMapping(session) with SqlEmbedding2Mapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlEmbedding2Mapping[IO] } final class Embedding3Suite extends SkunkDatabaseSuite with SqlEmbedding3Suite { - lazy val mapping = new SkunkTestMapping(session) with SqlEmbedding3Mapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlEmbedding3Mapping[IO] } final class FilterJoinAliasSuite extends SkunkDatabaseSuite with SqlFilterJoinAliasSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlFilterJoinAliasMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlFilterJoinAliasMapping[IO] } final class FilterOrderOffsetLimitSuite extends SkunkDatabaseSuite with SqlFilterOrderOffsetLimitSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlFilterOrderOffsetLimitMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlFilterOrderOffsetLimitMapping[IO] } final class FilterOrderOffsetLimit2Suite extends SkunkDatabaseSuite with SqlFilterOrderOffsetLimit2Suite { - lazy val mapping = new SkunkTestMapping(session) with SqlFilterOrderOffsetLimit2Mapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlFilterOrderOffsetLimit2Mapping[IO] } final class GraphSuite extends SkunkDatabaseSuite with SqlGraphSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlGraphMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlGraphMapping[IO] } final class InterfacesSuite extends SkunkDatabaseSuite with SqlInterfacesSuite { lazy val mapping = - new SkunkTestMapping(session) with SqlInterfacesMapping[IO] { + new SkunkTestMapping(pool) with SqlInterfacesMapping[IO] { def entityType: TestCodec[EntityType] = (codec.int4.imap(EntityType.fromInt)(EntityType.toInt), false) } @@ -92,18 +92,18 @@ final class InterfacesSuite extends SkunkDatabaseSuite with SqlInterfacesSuite { final class InterfacesSuite2 extends SkunkDatabaseSuite with SqlInterfacesSuite2 { lazy val mapping = - new SkunkTestMapping(session) with SqlInterfacesMapping2[IO] { + new SkunkTestMapping(pool) with SqlInterfacesMapping2[IO] { def entityType: TestCodec[EntityType] = (codec.int4.imap(EntityType.fromInt)(EntityType.toInt), false) } } final class JsonbSuite extends SkunkDatabaseSuite with SqlJsonbSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlJsonbMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlJsonbMapping[IO] } final class LikeSuite extends SkunkDatabaseSuite with SqlLikeSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlLikeMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlLikeMapping[IO] } final class MappingValidatorValidSuite extends SkunkDatabaseSuite with SqlMappingValidatorValidSuite { @@ -123,12 +123,12 @@ final class MappingValidatorInvalidSuite extends SkunkDatabaseSuite with SqlMapp } final class MixedSuite extends SkunkDatabaseSuite with SqlMixedSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlMixedMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlMixedMapping[IO] } final class MovieSuite extends SkunkDatabaseSuite with SqlMovieSuite { lazy val mapping = - new SkunkTestMapping(session) with SqlMovieMapping[IO] { + new SkunkTestMapping(pool) with SqlMovieMapping[IO] { def genre: TestCodec[Genre] = (codec.int4.imap(Genre.fromInt)(Genre.toInt), false) def feature: TestCodec[Feature] = (codec.varchar.imap(Feature.fromString)(_.toString), false) def tagList: TestCodec[List[String]] = (codec.int4.imap(Tags.fromInt)(Tags.toInt), false) @@ -137,21 +137,21 @@ final class MovieSuite extends SkunkDatabaseSuite with SqlMovieSuite { final class MutationSuite extends SkunkDatabaseSuite with SqlMutationSuite { // A resource that copies and drops the table used in the tests. - def withDuplicatedTables(s: Session[IO]): Resource[IO, Session[IO]] = { - val alloc = s.execute(sql"CREATE TABLE city_copy AS SELECT * FROM city".command).as(s) - val free = s.execute(sql"DROP TABLE city_copy".command).void + def withDuplicatedTables(p: Resource[IO, Session[IO]]): Resource[IO, Resource[IO, Session[IO]]] = { + val alloc = p.use(_.execute(sql"CREATE TABLE city_copy AS SELECT * FROM city".command)).as(p) + val free = p.use(_.execute(sql"DROP TABLE city_copy".command)).void Resource.make(alloc)(_ => free) } - override def sessionResource: Resource[IO, Session[IO]] = - super.sessionResource.flatMap(withDuplicatedTables) + override def poolResource: Resource[IO, Resource[IO, Session[IO]]] = + super.poolResource.flatMap(withDuplicatedTables) lazy val mapping = - new SkunkTestMapping(session) with SqlMutationMapping[IO] { + new SkunkTestMapping(pool) with SqlMutationMapping[IO] { def updatePopulation(id: Int, population: Int): IO[Unit] = - session.prepareR(sql"UPDATE city_copy SET population=${codec.int4} WHERE id=${codec.int4}".command).use { ps => + pool.use(_.prepareR(sql"UPDATE city_copy SET population=${codec.int4} WHERE id=${codec.int4}".command).use { ps => ps.execute(population *: id *: EmptyTuple).void - } + }) def createCity(name: String, countryCode: String, population: Int): IO[Int] = { val q = sql""" @@ -159,9 +159,9 @@ final class MutationSuite extends SkunkDatabaseSuite with SqlMutationSuite { VALUES (nextval('city_id'), ${codec.varchar}, ${codec.bpchar(3)}, 'ignored', ${codec.int4}) RETURNING id """.query(codec.int4) - session.prepareR(q).use { ps => + pool.use(_.prepareR(q).use { ps => ps.unique(name *: countryCode *: population *: EmptyTuple) - } + }) } } } @@ -172,7 +172,7 @@ final class NestedEffectsSuite extends SkunkDatabaseSuite with SqlNestedEffectsS currencyService0 <- CurrencyService[IO] } yield { val mapping = - new SkunkTestMapping(session) with SqlNestedEffectsMapping[IO] { + new SkunkTestMapping(pool) with SqlNestedEffectsMapping[IO] { lazy val currencyService = currencyService0 } (currencyService0, mapping) @@ -180,50 +180,50 @@ final class NestedEffectsSuite extends SkunkDatabaseSuite with SqlNestedEffectsS } final class Paging1Suite extends SkunkDatabaseSuite with SqlPaging1Suite { - lazy val mapping = new SkunkTestMapping(session) with SqlPaging1Mapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlPaging1Mapping[IO] } final class Paging2Suite extends SkunkDatabaseSuite with SqlPaging2Suite { - lazy val mapping = new SkunkTestMapping(session) with SqlPaging2Mapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlPaging2Mapping[IO] } final class Paging3Suite extends SkunkDatabaseSuite with SqlPaging3Suite { - lazy val mapping = new SkunkTestMapping(session) with SqlPaging3Mapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlPaging3Mapping[IO] } final class ProjectionSuite extends SkunkDatabaseSuite with SqlProjectionSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlProjectionMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlProjectionMapping[IO] } final class RecursiveInterfacesSuite extends SkunkDatabaseSuite with SqlRecursiveInterfacesSuite { lazy val mapping = - new SkunkTestMapping(session) with SqlRecursiveInterfacesMapping[IO] { + new SkunkTestMapping(pool) with SqlRecursiveInterfacesMapping[IO] { def itemType: TestCodec[ItemType] = (codec.int4.imap(ItemType.fromInt)(ItemType.toInt), false) } } final class SiblingListsSuite extends SkunkDatabaseSuite with SqlSiblingListsSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlSiblingListsData[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlSiblingListsData[IO] } final class TreeSuite extends SkunkDatabaseSuite with SqlTreeSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlTreeMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlTreeMapping[IO] } final class UnionsSuite extends SkunkDatabaseSuite with SqlUnionSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlUnionsMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlUnionsMapping[IO] } final class WorldSuite extends SkunkDatabaseSuite with SqlWorldSuite { - lazy val mapping = new SkunkTestMapping(session) with SqlWorldMapping[IO] + lazy val mapping = new SkunkTestMapping(pool) with SqlWorldMapping[IO] } final class WorldCompilerSuite extends SkunkDatabaseSuite with SqlWorldCompilerSuite { type Fragment = skunk.AppliedFragment def mapping: IO[(Mapping[IO], SqlStatsMonitor[IO,Fragment])] = - SkunkMonitor.statsMonitor[IO].map(mon => (new SkunkTestMapping(session, mon) with SqlWorldMapping[IO], mon)) + SkunkMonitor.statsMonitor[IO].map(mon => (new SkunkTestMapping(pool, mon) with SqlWorldMapping[IO], mon)) def simpleRestrictedQuerySql: String = "SELECT country.code, country.name FROM country WHERE ((country.code = $1))" diff --git a/modules/skunk/js-jvm/src/test/scala/subscription/SubscriptionMapping.scala b/modules/skunk/js-jvm/src/test/scala/subscription/SubscriptionMapping.scala index 27de6cae..e39fb25b 100644 --- a/modules/skunk/js-jvm/src/test/scala/subscription/SubscriptionMapping.scala +++ b/modules/skunk/js-jvm/src/test/scala/subscription/SubscriptionMapping.scala @@ -15,7 +15,7 @@ package grackle.skunk.test.subscription -import cats.effect.Sync +import cats.effect.{Resource, Sync} import skunk.Session import skunk.codec.all._ import skunk.implicits._ @@ -97,7 +97,8 @@ trait SubscriptionMapping[F[_]] extends SkunkMapping[F] { fieldMappings = List( RootStream.computeChild("channel")((child, _, _) => for { - id <- session.channel(id"city_channel").listen(256).map(_.value.toInt) + s <- fs2.Stream.resource(pool) + id <- s.channel(id"city_channel").listen(256).map(_.value.toInt) } yield Unique(Filter(Eql(CityType / "id", Const(id)), child)).success ) ) @@ -111,6 +112,6 @@ trait SubscriptionMapping[F[_]] extends SkunkMapping[F] { } object SubscriptionMapping extends SkunkMappingCompanion { - def mkMapping[F[_]: Sync](session: Session[F], monitor: SkunkMonitor[F]): Mapping[F] = - new SkunkMapping[F](session, monitor) with SubscriptionMapping[F] + def mkMapping[F[_]: Sync](pool: Resource[F, Session[F]], monitor: SkunkMonitor[F]): Mapping[F] = + new SkunkMapping[F](pool, monitor) with SubscriptionMapping[F] } diff --git a/modules/skunk/js-jvm/src/test/scala/subscription/SubscriptionSuite.scala b/modules/skunk/js-jvm/src/test/scala/subscription/SubscriptionSuite.scala index ad2c7b49..e13b04cf 100644 --- a/modules/skunk/js-jvm/src/test/scala/subscription/SubscriptionSuite.scala +++ b/modules/skunk/js-jvm/src/test/scala/subscription/SubscriptionSuite.scala @@ -27,7 +27,7 @@ import grackle.skunk.test.SkunkDatabaseSuite class SubscriptionSuite extends SkunkDatabaseSuite { - lazy val mapping = SubscriptionMapping.mkMapping(session) + lazy val mapping = SubscriptionMapping.mkMapping(pool) test("subscription driven by a Postgres channel") { @@ -79,8 +79,8 @@ class SubscriptionSuite extends SkunkDatabaseSuite { _ <- IO.sleep(1.second) // Send some notifications through Postgres, which will trigger queries on the subscription. - _ <- { - val ch = session.channel(id"city_channel").contramap[Int](_.toString) + _ <- pool.use { s => + val ch = s.channel(id"city_channel").contramap[Int](_.toString) List(101, 102, 103).traverse_(ch.notify) } diff --git a/modules/skunk/shared/src/main/scala/SkunkMapping.scala b/modules/skunk/shared/src/main/scala/SkunkMapping.scala index 767bddc1..848cea3b 100644 --- a/modules/skunk/shared/src/main/scala/SkunkMapping.scala +++ b/modules/skunk/shared/src/main/scala/SkunkMapping.scala @@ -19,7 +19,7 @@ package skunk import scala.util.control.NonFatal import cats.{Foldable, Reducible} -import cats.effect.Sync +import cats.effect.{ Resource, Sync } import cats.implicits._ import _root_.skunk.{ AppliedFragment, Decoder, Session, Fragment => SFragment } import _root_.skunk.data.Arr @@ -32,7 +32,7 @@ import grackle.sql._ import grackle.sqlpg._ abstract class SkunkMapping[F[_]]( - val session: Session[F], + val pool: Resource[F, Session[F]], val monitor: SkunkMonitor[F] )( implicit val M: Sync[F] @@ -41,7 +41,7 @@ abstract class SkunkMapping[F[_]]( trait SkunkMappingLike[F[_]] extends Mapping[F] with SqlPgMappingLike[F] { outer => implicit val M: Sync[F] - val session: Session[F] + val pool: Resource[F, Session[F]] val monitor: SkunkMonitor[F] // Grackle needs to know about codecs, encoders, and fragments. @@ -162,12 +162,11 @@ trait SkunkMappingLike[F[_]] extends Mapping[F] with SqlPgMappingLike[F] { outer } } - session.prepare(fragment.fragment.query(rowDecoder)).flatMap { pq => - pq.stream(fragment.argument, 1024) - .compile - .toVector - } - .onError { + pool.use { s => + Resource.eval(s.prepare(fragment.fragment.query(rowDecoder))).use { ps => + ps.stream(fragment.argument, 1024).compile.toVector + } + }.onError { case NonFatal(e) => Sync[F].delay(e.printStackTrace()) } } diff --git a/modules/skunk/shared/src/main/scala/SkunkMappingCompanion.scala b/modules/skunk/shared/src/main/scala/SkunkMappingCompanion.scala index 4f74882a..e389139d 100644 --- a/modules/skunk/shared/src/main/scala/SkunkMappingCompanion.scala +++ b/modules/skunk/shared/src/main/scala/SkunkMappingCompanion.scala @@ -17,13 +17,13 @@ package grackle package skunk import _root_.skunk.Session -import cats.effect.Sync +import cats.effect.{ Resource, Sync } trait SkunkMappingCompanion { - def mkMapping[F[_]: Sync](pool: Session[F], monitor: SkunkMonitor[F]): Mapping[F] + def mkMapping[F[_]: Sync](pool: Resource[F, Session[F]], monitor: SkunkMonitor[F]): Mapping[F] - final def mkMapping[F[_]: Sync](pool: Session[F]): Mapping[F] = + final def mkMapping[F[_]: Sync](pool: Resource[F, Session[F]]): Mapping[F] = mkMapping(pool, SkunkMonitor.noopMonitor) }