[DRAFT] init http module

.scalafix.conf

@@ -1,13 +1,9 @@
 rules = [
   DisableSyntax
-  ExplicitResultTypes
   LeakingImplicitClassVal
   NoAutoTupling
   NoValInForComprehension
   OrganizeImports
-  ProcedureSyntax
-  RemoveUnused
-  MissingFinal
 ]
 
 OrganizeImports {
@@ -23,10 +19,6 @@ OrganizeImports {
   ]
 }
 
-RemoveUnused {
-  imports = false # handled by OrganizeImports
-}
-
 DisableSyntax.noThrows = true
 DisableSyntax.noNulls = true
 DisableSyntax.noReturns = true

.scalafix_test.conf

@@ -1,23 +1,11 @@
 rules = [
-  Disable
   DisableSyntax
-  ExplicitResultTypes
   LeakingImplicitClassVal
   NoAutoTupling
   NoValInForComprehension
   OrganizeImports
-  ProcedureSyntax
-  RemoveUnused
-  MissingFinal
 ]
 
-Disable {
-  ifSynthetic = [
-    "scala/Option.option2Iterable"
-    "scala/Predef.any2stringadd"
-  ]
-}
-
 OrganizeImports {
   removeUnused = true
   targetDialect = Scala3
@@ -31,10 +19,8 @@ OrganizeImports {
   ]
 }
 
-RemoveUnused {
-  imports = false # handled by OrganizeImports
-}
-
+DisableSyntax.noThrows = false
+DisableSyntax.noNulls = false
 DisableSyntax.noReturns = true
 DisableSyntax.noXml = true
 DisableSyntax.noFinalize = true

build.sbt

@@ -1,11 +1,15 @@
-import zio.sbt.githubactions.{Job, Step, Condition, ActionRef}
+import zio.sbt.githubactions.{ActionRef, Condition, Job, Step}
 import _root_.io.circe.Json
+
+import scala.annotation.tailrec
 enablePlugins(ZioSbtEcosystemPlugin, ZioSbtCiPlugin)
 
 lazy val _scala2 = "2.13.15"
 
 lazy val _scala3 = "3.3.4"
 
+lazy val zioGcpVersion = "0.0.3+9-e55f796a-SNAPSHOT"
+
 inThisBuild(
   List(
     name         := "ZIO Google Cloud Pub/Sub",
@@ -27,11 +31,11 @@ inThisBuild(
         url = url("https://github.com/qhquanghuy"),
       ),
     ),
-    zioVersion         := "2.1.11",
+    zioVersion         := "2.1.13",
     scala213           := _scala2,
     scala3             := _scala3,
-    scalaVersion       := _scala2,
-    crossScalaVersions := Seq(_scala2, _scala3),
+    scalaVersion       := _scala3,
+    crossScalaVersions := Seq(_scala3),
     versionScheme      := Some("early-semver"),
     ciEnabledBranches  := Seq("master"),
     ciJvmOptions ++= Seq("-Xms2G", "-Xmx2G", "-Xss4M", "-XX:+UseG1GC"),
@@ -116,9 +120,6 @@ inThisBuild(
     },
     scalafmt         := true,
     scalafmtSbtCheck := true,
-    scalafixDependencies ++= List(
-      "com.github.vovapolu" %% "scaluzzi" % "0.1.23"
-    ),
   )
 )
 
@@ -149,8 +150,12 @@ lazy val ciGenerateGithubWorkflowV2 = Def.task {
 lazy val commonSettings = List(
   libraryDependencies ++= {
     CrossVersion.partialVersion(scalaVersion.value) match {
-      case Some((2, _)) => Seq(compilerPlugin("com.olegpy" %% "better-monadic-for" % "0.3.1"))
-      case _            => Seq()
+      case Some((2, _)) =>
+        Seq(
+          compilerPlugin("com.olegpy"    %% "better-monadic-for" % "0.3.1"),
+          compilerPlugin("org.typelevel" %% "kind-projector"     % "0.13.3" cross CrossVersion.full),
+        )
+      case _ => Seq()
     }
   },
   javacOptions ++= Seq("-source", "17"),
@@ -163,7 +168,9 @@ lazy val commonSettings = List(
   Compile / scalacOptions --= sys.env.get("CI").fold(Seq("-Xfatal-warnings"))(_ => Nil),
   Test / scalafixConfig := Some(new File(".scalafix_test.conf")),
   Test / scalacOptions --= Seq("-Xfatal-warnings"),
-) ++ scalafixSettings
+  semanticdbEnabled := true,
+  semanticdbVersion := scalafixSemanticdb.revision, // use Scalafix compatible version
+)
 
 val noPublishSettings = List(
   publish         := {},
@@ -177,9 +184,12 @@ lazy val root =
     .aggregate(
       zioPubsub.jvm,
       zioPubsub.native,
+      zioPubsubHttp.jvm,
+      zioPubsubHttp.native,
       zioPubsubGoogle,
       zioPubsubGoogleTest,
-      zioPubsubTestkit,
+      zioPubsubTestkit.jvm,
+      zioPubsubTestkit.native,
       zioPubsubSerdeCirce.jvm,
       zioPubsubSerdeCirce.native,
       zioPubsubSerdeVulcan,
@@ -206,6 +216,28 @@ lazy val zioPubsub = crossProject(JVMPlatform, NativePlatform)
     )
   )
 
+lazy val zioPubsubHttp = crossProject(JVMPlatform, NativePlatform)
+  .in(file("zio-pubsub-http"))
+  .settings(moduleName := "zio-pubsub-http")
+  .dependsOn(zioPubsub)
+  .settings(commonSettings)
+  .settings(
+    libraryDependencies ++= Seq(
+      "com.anymindgroup" %%% "zio-gcp-auth"      % zioGcpVersion,
+      "com.anymindgroup" %%% "zio-gcp-pubsub-v1" % zioGcpVersion,
+    ),
+    // temprorary until zio-gcp is published
+    credentials += {
+      for {
+        username <- sys.env.get("ARTIFACT_REGISTRY_USERNAME")
+        apiKey   <- sys.env.get("ARTIFACT_REGISTRY_PASSWORD")
+      } yield Credentials("https://asia-maven.pkg.dev", "asia-maven.pkg.dev", username, apiKey)
+    }.getOrElse(Credentials(Path.userHome / ".ivy2" / ".credentials")),
+    resolvers ++= Seq(
+      "AnyChat Maven" at "https://asia-maven.pkg.dev/anychat-staging/maven"
+    ),
+  )
+
 val vulcanVersion = "1.11.1"
 lazy val zioPubsubSerdeVulcan = (project in file("zio-pubsub-serde-vulcan"))
   .settings(moduleName := "zio-pubsub-serde-vulcan")
@@ -247,7 +279,7 @@ lazy val zioPubsubGoogle = (project in file("zio-pubsub-google"))
 
 lazy val zioPubsubGoogleTest = project
   .in(file("zio-pubsub-google-test"))
-  .dependsOn(zioPubsub.jvm, zioPubsubGoogle, zioPubsubTestkit, zioPubsubSerdeCirce.jvm, zioPubsubSerdeVulcan)
+  .dependsOn(zioPubsub.jvm, zioPubsubGoogle, zioPubsubTestkit.jvm, zioPubsubSerdeCirce.jvm, zioPubsubSerdeVulcan)
   .settings(moduleName := "zio-pubsub-google-test")
   .settings(commonSettings)
   .settings(noPublishSettings)
@@ -258,10 +290,10 @@ lazy val zioPubsubGoogleTest = project
     (Test / fork)              := true,
   )
 
-// TODO remove dependency on zioPubsubGoogle
 lazy val zioPubsubTestkit =
-  (project in file("zio-pubsub-testkit"))
-    .dependsOn(zioPubsub.jvm, zioPubsubGoogle)
+  crossProject(JVMPlatform, NativePlatform)
+    .in(file("zio-pubsub-testkit"))
+    .dependsOn(zioPubsub, zioPubsubHttp)
     .settings(moduleName := "zio-pubsub-testkit")
     .settings(commonSettings)
     .settings(
@@ -274,26 +306,26 @@ lazy val zioPubsubTestkit =
 lazy val zioPubsubTest =
   crossProject(JVMPlatform, NativePlatform)
     .in(file("zio-pubsub-test"))
-    .dependsOn(zioPubsub, zioPubsubSerdeCirce)
+    .dependsOn(zioPubsub, zioPubsubSerdeCirce, zioPubsubHttp, zioPubsubTestkit)
     .settings(moduleName := "zio-pubsub-test")
     .settings(commonSettings)
     .settings(noPublishSettings)
     .settings(testDeps)
     .jvmSettings(coverageEnabled := true)
     .nativeSettings(coverageEnabled := false)
 
-lazy val examples = (project in file("examples"))
-  .dependsOn(zioPubsubGoogle)
-  .settings(noPublishSettings)
-  .settings(
-    scalaVersion       := _scala3,
-    crossScalaVersions := Seq(_scala3),
-    coverageEnabled    := false,
-    fork               := true,
-    libraryDependencies ++= Seq(
-      "dev.zio" %% "zio-json" % "0.7.1"
-    ),
-  )
+//lazy val examples = (project in file("examples"))
+//  .dependsOn(zioPubsubGoogle)
+//  .settings(noPublishSettings)
+//  .settings(
+//    scalaVersion       := _scala3,
+//    crossScalaVersions := Seq(_scala3),
+//    coverageEnabled    := false,
+//    fork               := true,
+//    libraryDependencies ++= Seq(
+//      "dev.zio" %% "zio-json" % "0.7.1"
+//    ),
+//  )
 
 lazy val testDeps = Seq(
   libraryDependencies ++= Seq(

docker-compose.yaml

@@ -1,7 +1,7 @@
 services:
   pubsub:
     # https://console.cloud.google.com/gcr/images/google.com:cloudsdktool/GLOBAL/cloud-sdk
-    image: gcr.io/google.com/cloudsdktool/cloud-sdk:480.0.0-emulators
+    image: gcr.io/google.com/cloudsdktool/cloud-sdk:498.0.0-emulators
     ports:
       - "8085:8085"
     command: gcloud beta emulators pubsub start --project=any --host-port=0.0.0.0:8085

examples/src/main/scala/BasicSubscription.scala

@@ -22,8 +22,8 @@ object BasicSubscription extends ZIOAppDefault:
 
     ZLayer.scoped(
       G.Subscriber.makeStreamingPullSubscriber(
-        connection = G.PubsubConnectionConfig.Emulator(
-          G.PubsubConnectionConfig.GcpProject("any"),
+        connection = PubsubConnectionConfig.Emulator(
+          PubsubConnectionConfig.GcpProject("any"),
           "localhost:8085",
         )
       )

examples/src/main/scala/ExamplesAdminSetup.scala

@@ -1,4 +1,4 @@
-import com.anymindgroup.pubsub.google.{PubsubAdmin, PubsubConnectionConfig}
+import com.anymindgroup.pubsub.google.PubsubAdmin
 import com.anymindgroup.pubsub.*
 import zio.*
 

examples/src/main/scala/PubAndSubAndAdminExample.scala

@@ -60,9 +60,9 @@ object PubAndSubAndAdminExample extends ZIOAppDefault:
         } yield ()
       }
 
-  val pubsubConnection: G.PubsubConnectionConfig =
-    G.PubsubConnectionConfig.Emulator(
-      project = G.PubsubConnectionConfig.GcpProject("any"),
+  val pubsubConnection: PubsubConnectionConfig =
+    PubsubConnectionConfig.Emulator(
+      project = PubsubConnectionConfig.GcpProject("any"),
       host = "localhost:8085",
     )
 

examples/src/main/scala/SamplesPublisher.scala

@@ -25,8 +25,8 @@ object SamplesPublisher extends ZIOAppDefault:
     ZLayer.scoped(
       G.Publisher.make(
         config = G.PublisherConfig(
-          connection = G.PubsubConnectionConfig.Emulator(
-            G.PubsubConnectionConfig.GcpProject("any"),
+          connection = PubsubConnectionConfig.Emulator(
+            PubsubConnectionConfig.GcpProject("any"),
             "localhost:8085",
           ),
           topicName = "basic_example",

project/build.properties

@@ -1 +1 @@
-sbt.version = 1.10.3
+sbt.version = 1.10.6

project/plugins.sbt

@@ -7,7 +7,7 @@ addSbtPlugin("dev.zio"        % "zio-sbt-ci"        % zioSbtVersion exclude ("or
 addSbtPlugin("org.xerial.sbt" % "sbt-sonatype"      % "3.12.2")
 addSbtPlugin("dev.zio"        % "zio-sbt-ecosystem" % zioSbtVersion)
 
-addSbtPlugin("org.scala-native" % "sbt-scala-native" % "0.5.5")
+addSbtPlugin("org.scala-native" % "sbt-scala-native" % "0.5.6")
 
 addSbtPlugin("org.typelevel" % "sbt-tpolecat" % "0.5.2")
 

zio-pubsub-google-test/zio-gc-pubsub-google-test/src/test/scala/com/anymindgroup/pubsub/google/AvroPublisherSpec.scala → zio-pubsub-google-test/src/test/scala/com/anymindgroup/pubsub/google/AvroPublisherSpec.scala

@@ -1,7 +1,7 @@
 package com.anymindgroup.pubsub.google
 
+import com.anymindgroup.pubsub.PubsubTestSupport.*
 import com.anymindgroup.pubsub.google
-import com.anymindgroup.pubsub.google.PubsubTestSupport.*
 import com.anymindgroup.pubsub.google.TestSupport.*
 import com.anymindgroup.pubsub.model.*
 import com.anymindgroup.pubsub.pub.*
@@ -39,22 +39,22 @@ object AvroPublisherSpec extends ZIOSpecDefault {
         SchemaSettings(
           schema = Some(
             SchemaRegistry(
-              id = s"${topicName.getTopic()}_v1",
+              id = s"${topicName.topic}_v1",
               schemaType = SchemaType.Avro,
               definition = ZIO.succeed(TestEvent.avroCodecSchema),
             )
           ),
           encoding = encoding,
         )
       topic = Topic[Any, TestEvent](
-                topicName.getTopic(),
+                topicName.topic,
                 schema,
                 VulcanSerde.fromAvroCodec(TestEvent.avroCodec, encoding),
               )
       publisherConfig = PublisherConfig.forTopic(conn, topic, enableOrdering = true)
       subscription = Subscription(
                        topicName = topic.name,
-                       name = subscriptionName.getSubscription(),
+                       name = subscriptionName.subscription,
                        filter = None,
                        enableOrdering = true,
                        expiration = None,
@@ -91,6 +91,7 @@ object AvroPublisherSpec extends ZIOSpecDefault {
             )
         publishedOrderingKeys = testMessages.map(_.orderingKey)
         consumedOrderingKeys  = consumed.map(_.meta.orderingKey)
+        _                    <- assert(publishedOrderingKeys)(hasSameElements(consumedOrderingKeys))
         _                    <- assert(consumedAttr)(hasSameElements(publishedAttrs))
       } yield assert(publishedOrderingKeys)(hasSameElements(consumedOrderingKeys))
     }) ::

zio-pubsub-google-test/zio-gc-pubsub-google-test/src/test/scala/com/anymindgroup/pubsub/google/PubsubAdminSpec.scala → zio-pubsub-google-test/src/test/scala/com/anymindgroup/pubsub/google/PubsubAdminSpec.scala

@@ -1,8 +1,9 @@
 package com.anymindgroup.pubsub.google
 
+import scala.concurrent.duration.Duration
 import scala.util.Try
 
-import com.anymindgroup.pubsub.google.PubsubTestSupport.*
+import com.anymindgroup.pubsub.PubsubTestSupport.*
 import com.anymindgroup.pubsub.google.TestSupport.*
 import com.anymindgroup.pubsub.model.{SchemaRegistry, SchemaType, *}
 import com.anymindgroup.pubsub.serde.{CirceSerde, VulcanSerde}
@@ -31,7 +32,7 @@ object PubsubAdminSpec extends ZIOSpecDefault {
               case SchemaSettings(Encoding.Json, None) => CirceSerde.fromCirceCodec(TestEvent.jsonCodec)
             }
     topicName <- topicNameGen
-  } yield Topic(topicName.getTopic(), schemaSetting, serde)
+  } yield Topic(topicName.topic, schemaSetting, serde)
 
   override def spec: Spec[TestEnvironment & Scope, Any] = suite("PubsubAdminSpec")(
     test("crating a subscriptions for non existing topic fails") {
@@ -49,16 +50,17 @@ object PubsubAdminSpec extends ZIOSpecDefault {
         _             <- PubsubAdmin.setup(connection, topics.map(_._1), subscriptions.flatten)
         _ <- ZIO.foreachDiscard(subscriptions.flatten) { subscription =>
                for {
-                 maybePubsubSub <- findSubscription(subscription.name)
+                 maybePubsubSub <- findSubscription(SubscriptionName(connection.project.name, subscription.name))
                  _              <- assert(maybePubsubSub)(isSome)
                  pubsubSub       = maybePubsubSub.get
-                 pubsubSubName   = pubsubSub.getName().split("/").last
+                 pubsubSubName   = pubsubSub.name.split("/").last
                  _              <- assert(pubsubSubName)(equalTo(subscription.name))
-                 _              <- assert(pubsubSub.getEnableMessageOrdering())(equalTo(subscription.enableOrdering))
-                 _              <- assert(pubsubSub.getFilter())(equalTo(subscription.filter.map(_.value).getOrElse("")))
-                 _ <- assert(pubsubSub.getExpirationPolicy().getTtl().getSeconds())(
-                        equalTo(subscription.expiration.map(_.getSeconds()).getOrElse(0L))
-                      )
+                 _              <- assert(pubsubSub.enableMessageOrdering.getOrElse(false))(equalTo(subscription.enableOrdering))
+                 _              <- assert(pubsubSub.filter.getOrElse(""))(equalTo(subscription.filter.map(_.value).getOrElse("")))
+                 _ <-
+                   assert(pubsubSub.expirationPolicy.flatMap(_.ttl.map(Duration(_).toSeconds)))(
+                     equalTo(subscription.expiration.map(_.toSeconds()))
+                   )
                } yield assertTrue(true)
              }
         exit <- PubsubAdmin.setup(connection, topics.map(_._1), subscriptions.flatten).exit
@@ -90,6 +92,6 @@ object PubsubAdminSpec extends ZIOSpecDefault {
     },
   ).provideSomeShared[Scope](
     emulatorConnectionConfigLayer(),
-    SubscriptionAdmin.layer,
+    emulatorBackendLayer,
   ) @@ TestAspect.nondeterministic
 }