Skip to content

Commit 698075d

Browse files
johanandrenjohanandren
authored
google pub sub and cloud storage: Scala 3 (#3350)
1 parent d9565ac commit 698075d

File tree

15 files changed

+79
-50
lines changed

15 files changed

+79
-50
lines changed

‎.github/workflows/check-build-test.yml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,8 +29,8 @@ jobs:
2929
# v4.1.1
3030
uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11
3131
with: # https://github.com/olafurpg/setup-scala#faster-checkout-of-big-repos
32-
fetch-depth: 100
33-
fetch-tags: true
32+
# we don't know what commit the last tag was it's safer to get entire repo so previousStableVersion resolves
33+
fetch-depth: 0
3434

3535
- name: Cache Coursier cache
3636
# https://github.com/coursier/cache-action/releases

‎build.sbt‎

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -209,7 +209,11 @@ lazy val googleCommon = alpakkaProject(
209209
"google.common",
210210
Dependencies.GoogleCommon,
211211
Test / fork := true,
212-
headerSources / excludeFilter := HiddenFileFilter || "JwtSprayJsonParser.scala"
212+
headerSources / excludeFilter := HiddenFileFilter || "JwtSprayJsonParser.scala",
213+
Scala3.settings,
214+
mimaPreviousArtifacts :=
215+
(if (scalaBinaryVersion.value == "3") Set.empty // No previous Scala 3 artifacts, drop once there are
216+
else mimaPreviousArtifacts.value)
213217
)
214218

215219
lazy val googleCloudBigQuery = alpakkaProject(
@@ -246,7 +250,11 @@ lazy val googleCloudPubSub = alpakkaProject(
246250
Dependencies.GooglePubSub,
247251
Test / fork := true,
248252
// See docker-compose.yml gcloud-pubsub-emulator_prep
249-
Test / envVars := Map("PUBSUB_EMULATOR_HOST" -> "localhost", "PUBSUB_EMULATOR_PORT" -> "8538")
253+
Test / envVars := Map("PUBSUB_EMULATOR_HOST" -> "localhost", "PUBSUB_EMULATOR_PORT" -> "8538"),
254+
Scala3.settings,
255+
mimaPreviousArtifacts :=
256+
(if (scalaBinaryVersion.value == "3") Set.empty // No previous Scala 3 artifacts, drop once there are
257+
else mimaPreviousArtifacts.value)
250258
).dependsOn(googleCommon)
251259

252260
lazy val googleCloudPubSubGrpc = alpakkaProject(
@@ -262,14 +270,22 @@ lazy val googleCloudPubSubGrpc = alpakkaProject(
262270
"-Wconf:src=.+/akka-grpc/main/.+:s",
263271
"-Wconf:src=.+/akka-grpc/test/.+:s"
264272
),
265-
compile / javacOptions := (compile / javacOptions).value.filterNot(_ == "-Xlint:deprecation")
273+
compile / javacOptions := (compile / javacOptions).value.filterNot(_ == "-Xlint:deprecation"),
274+
Scala3.settings,
275+
mimaPreviousArtifacts :=
276+
(if (scalaBinaryVersion.value == "3") Set.empty // No previous Scala 3 artifacts, drop once there are
277+
else mimaPreviousArtifacts.value)
266278
).enablePlugins(AkkaGrpcPlugin).dependsOn(googleCommon)
267279

268280
lazy val googleCloudStorage = alpakkaProject(
269281
"google-cloud-storage",
270282
"google.cloud.storage",
271283
Test / fork := true,
272-
Dependencies.GoogleStorage
284+
Dependencies.GoogleStorage,
285+
Scala3.settings,
286+
mimaPreviousArtifacts :=
287+
(if (scalaBinaryVersion.value == "3") Set.empty // No previous Scala 3 artifacts, drop once there are
288+
else mimaPreviousArtifacts.value)
273289
).dependsOn(googleCommon)
274290

275291
lazy val googleFcm = alpakkaProject("google-fcm", "google.firebase.fcm", Dependencies.GoogleFcm, Test / fork := true)

‎google-cloud-pub-sub/src/main/scala/akka/stream/alpakka/googlecloud/pubsub/impl/PubSubApi.scala‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -169,7 +169,7 @@ private[pubsub] trait PubSubApi {
169169
.mapMaterializedValue(_ => NotUsed)
170170

171171
private implicit val pullResponseUnmarshaller: FromResponseUnmarshaller[PullResponse] =
172-
Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
172+
Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
173173
response.status match {
174174
case StatusCodes.Success(_) if response.entity.contentType == ContentTypes.`application/json` =>
175175
Unmarshal(response.entity).to[PullResponse]
@@ -208,7 +208,7 @@ private[pubsub] trait PubSubApi {
208208
.mapMaterializedValue(_ => NotUsed)
209209

210210
private implicit val acknowledgeResponseUnmarshaller: FromResponseUnmarshaller[Done] =
211-
Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
211+
Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
212212
response.status match {
213213
case StatusCodes.Success(_) =>
214214
response.discardEntityBytes().future
@@ -261,7 +261,7 @@ private[pubsub] trait PubSubApi {
261261
publish(topic, parallelism, None)
262262

263263
private implicit val publishResponseUnmarshaller: FromResponseUnmarshaller[PublishResponse] =
264-
Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
264+
Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
265265
response.status match {
266266
case StatusCodes.Success(_) if response.entity.contentType == ContentTypes.`application/json` =>
267267
Unmarshal(response.entity).to[PublishResponse]

‎google-cloud-pub-sub/src/main/scala/akka/stream/alpakka/googlecloud/pubsub/model.scala‎

Lines changed: 18 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -60,11 +60,15 @@ object PubSubConfig {
6060
pullReturnImmediately = true,
6161
pullMaxMessagesPerInternalBatch = 1000,
6262
Some(
63-
GoogleSettings().copy(
64-
projectId = projectId,
65-
credentials =
66-
ServiceAccountCredentials(projectId, clientEmail, privateKey, Seq("https://www.googleapis.com/auth/pubsub"))
67-
)
63+
GoogleSettings
64+
.apply()
65+
.copy(
66+
projectId = projectId,
67+
credentials = ServiceAccountCredentials(projectId,
68+
clientEmail,
69+
privateKey,
70+
Seq("https://www.googleapis.com/auth/pubsub"))
71+
)
6872
)
6973
)
7074

@@ -85,11 +89,15 @@ object PubSubConfig {
8589
pullReturnImmediately = pullReturnImmediately,
8690
pullMaxMessagesPerInternalBatch = pullMaxMessagesPerInternalBatch,
8791
Some(
88-
GoogleSettings().copy(
89-
projectId = projectId,
90-
credentials =
91-
ServiceAccountCredentials(projectId, clientEmail, privateKey, Seq("https://www.googleapis.com/auth/pubsub"))
92-
)
92+
GoogleSettings
93+
.apply()
94+
.copy(
95+
projectId = projectId,
96+
credentials = ServiceAccountCredentials(projectId,
97+
clientEmail,
98+
privateKey,
99+
Seq("https://www.googleapis.com/auth/pubsub"))
100+
)
93101
)
94102
)
95103

‎google-cloud-storage/src/main/scala/akka/stream/alpakka/googlecloud/storage/impl/Formats.scala‎

Lines changed: 13 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -11,20 +11,21 @@ import akka.stream.alpakka.googlecloud.storage._
1111
import spray.json.{DefaultJsonProtocol, JsObject, JsValue, RootJsonFormat, RootJsonReader}
1212

1313
import scala.util.Try
14+
import spray.json.enrichAny
1415

1516
@akka.annotation.InternalApi
1617
object Formats extends DefaultJsonProtocol {
1718

1819
private final case class CustomerEncryption(encryptionAlgorithm: String, keySha256: String)
1920
private implicit val customerEncryptionJsonFormat: RootJsonFormat[CustomerEncryption] = jsonFormat2(
20-
CustomerEncryption
21+
CustomerEncryption.apply
2122
)
2223

2324
private final case class Owner(entity: String, entityId: Option[String])
24-
private implicit val OwnerJsonFormat: RootJsonFormat[Owner] = jsonFormat2(Owner)
25+
private implicit val OwnerJsonFormat: RootJsonFormat[Owner] = jsonFormat2(Owner.apply)
2526

2627
private final case class ProjectTeam(projectNumber: String, team: String)
27-
private implicit val ProjectTeamJsonFormat: RootJsonFormat[ProjectTeam] = jsonFormat2(ProjectTeam)
28+
private implicit val ProjectTeamJsonFormat: RootJsonFormat[ProjectTeam] = jsonFormat2(ProjectTeam.apply)
2829

2930
private final case class ObjectAccessControls(kind: String,
3031
id: String,
@@ -40,7 +41,7 @@ object Formats extends DefaultJsonProtocol {
4041
projectTeam: ProjectTeam,
4142
etag: String)
4243
private implicit val ObjectAccessControlsJsonFormat: RootJsonFormat[ObjectAccessControls] = jsonFormat13(
43-
ObjectAccessControls
44+
ObjectAccessControls.apply
4445
)
4546

4647
/**
@@ -73,7 +74,7 @@ object Formats extends DefaultJsonProtocol {
7374
)
7475

7576
private implicit val storageObjectReadOnlyJson: RootJsonFormat[StorageObjectReadOnlyJson] = jsonFormat18(
76-
StorageObjectReadOnlyJson
77+
StorageObjectReadOnlyJson.apply
7778
)
7879

7980
// private sub class of StorageObjectJson used to workaround 22 field jsonFormat issue
@@ -95,7 +96,7 @@ object Formats extends DefaultJsonProtocol {
9596
)
9697

9798
private implicit val storageObjectWritableJson: RootJsonFormat[StorageObjectWriteableJson] = jsonFormat14(
98-
StorageObjectWriteableJson
99+
StorageObjectWriteableJson.apply
99100
)
100101

101102
private implicit object StorageObjectJsonFormat extends RootJsonFormat[StorageObjectJson] {
@@ -130,7 +131,7 @@ object Formats extends DefaultJsonProtocol {
130131
items: Option[List[StorageObjectJson]]
131132
)
132133

133-
private implicit val bucketInfoJsonFormat: RootJsonFormat[BucketInfoJson] = jsonFormat6(BucketInfoJson)
134+
private implicit val bucketInfoJsonFormat: RootJsonFormat[BucketInfoJson] = jsonFormat6(BucketInfoJson.apply)
134135

135136
/**
136137
* Google API rewrite response object
@@ -146,7 +147,9 @@ object Formats extends DefaultJsonProtocol {
146147
resource: Option[StorageObjectJson]
147148
)
148149

149-
private implicit val rewriteResponseFormat: RootJsonFormat[RewriteResponseJson] = jsonFormat6(RewriteResponseJson)
150+
private implicit val rewriteResponseFormat: RootJsonFormat[RewriteResponseJson] = jsonFormat6(
151+
RewriteResponseJson.apply
152+
)
150153

151154
/**
152155
* Google API bucket response object
@@ -162,7 +165,7 @@ object Formats extends DefaultJsonProtocol {
162165
etag: String
163166
)
164167

165-
implicit val bucketInfoFormat: RootJsonFormat[BucketInfo] = jsonFormat2(BucketInfo)
168+
implicit val bucketInfoFormat: RootJsonFormat[BucketInfo] = jsonFormat2(BucketInfo.apply)
166169

167170
implicit object BucketListResultReads extends RootJsonReader[BucketListResult] {
168171
override def read(json: JsValue): BucketListResult = {
@@ -177,7 +180,7 @@ object Formats extends DefaultJsonProtocol {
177180
}
178181

179182
private implicit val bucketListResultJsonReads: RootJsonFormat[BucketListResultJson] = jsonFormat4(
180-
BucketListResultJson
183+
BucketListResultJson.apply
181184
)
182185

183186
implicit object RewriteResponseReads extends RootJsonReader[RewriteResponse] {

‎google-cloud-storage/src/main/scala/akka/stream/alpakka/googlecloud/storage/impl/GCStorageStream.scala‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -131,7 +131,7 @@ import scala.concurrent.Future
131131
metadata: Option[Map[String, String]] = None): Sink[ByteString, Future[StorageObject]] =
132132
Sink
133133
.fromMaterializer { (mat, attr) =>
134-
implicit val settings = {
134+
implicit val settings: GoogleSettings = {
135135
val s = resolveSettings(mat, attr)
136136
s.copy(requestSettings = s.requestSettings.copy(uploadChunkSize = chunkSize))
137137
}
@@ -158,7 +158,7 @@ import scala.concurrent.Future
158158
}: PartialFunction[HttpResponse, Future[StorageObject]]
159159
}.withDefaultRetry
160160

161-
ResumableUpload[StorageObject](request).addAttributes(GoogleAttributes.settings(settings))
161+
ResumableUpload[StorageObject](request)(um).addAttributes(GoogleAttributes.settings(settings))
162162
}
163163
.mapMaterializedValue(_.flatten)
164164

@@ -236,7 +236,7 @@ import scala.concurrent.Future
236236
getBucketPath(bucket) / "o" / objectName
237237

238238
implicit def unmarshaller[T: FromEntityUnmarshaller]: Unmarshaller[HttpResponse, T] =
239-
Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
239+
Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
240240
response match {
241241
case HttpResponse(status, _, entity, _) if status.isSuccess() && !status.isRedirection() =>
242242
Unmarshal(entity).to[T]
@@ -248,7 +248,7 @@ import scala.concurrent.Future
248248
}.withDefaultRetry
249249

250250
implicit def optionUnmarshaller[T: FromEntityUnmarshaller]: Unmarshaller[HttpResponse, Option[T]] =
251-
Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
251+
Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
252252
response match {
253253
case HttpResponse(status, _, entity, _) if status.isSuccess() && !status.isRedirection() =>
254254
Unmarshal(entity).to[T].map(Some(_))
@@ -324,7 +324,7 @@ import scala.concurrent.Future
324324
)
325325
}
326326

327-
GoogleSettings(
327+
GoogleSettings.apply(
328328
legacySettings.projectId,
329329
credentials,
330330
settings.requestSettings

‎google-cloud-storage/src/test/scala/akka/stream/alpakka/googlecloud/storage/GCSExtSpec.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ class GCSExtSpec extends AnyFlatSpec with Matchers with LogCapturing {
2424
).asJava
2525
)
2626

27-
implicit val system = ActorSystem.create("gcs", config)
27+
implicit val system: ActorSystem = ActorSystem.create("gcs", config)
2828
val ext = GCSExt(system)
2929

3030
ext.settings.endpointUrl shouldBe endpointUrl

‎google-cloud-storage/src/test/scala/akka/stream/alpakka/googlecloud/storage/GCStorageExtSpec.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ class GCStorageExtSpec extends AnyFlatSpec with Matchers with LogCapturing {
3434
"alpakka.google.cloud.storage.token-scope" -> tokenScope
3535
).asJava
3636
)
37-
implicit val system = ActorSystem.create("gcStorage", config)
37+
implicit val system: ActorSystem = ActorSystem.create("gcStorage", config)
3838
@nowarn("msg=deprecated")
3939
val ext = GCStorageExt(system)
4040

‎google-common/src/main/scala/akka/stream/alpakka/google/GoogleSettings.scala‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -80,9 +80,9 @@ object GoogleSettings {
8080

8181
}
8282

83-
final case class GoogleSettings @InternalApi private (projectId: String,
84-
credentials: Credentials,
85-
requestSettings: RequestSettings) {
83+
final case class GoogleSettings @InternalApi private[akka] (projectId: String,
84+
credentials: Credentials,
85+
requestSettings: RequestSettings) {
8686
def getProjectId = projectId
8787
def getCredentials = credentials
8888
def getRequestSettings = requestSettings
@@ -125,7 +125,7 @@ object RequestSettings {
125125
apply(userIp.toScala, quotaUser.toScala, prettyPrint, chunkSize, retrySettings, forwardProxy.toScala)
126126
}
127127

128-
final case class RequestSettings @InternalApi private (
128+
final case class RequestSettings @InternalApi private[akka] (
129129
userIp: Option[String],
130130
quotaUser: Option[String],
131131
prettyPrint: Boolean,
@@ -247,7 +247,7 @@ object ForwardProxy {
247247
credentials: Option[BasicHttpCredentials],
248248
trustPem: Option[String])(implicit system: ClassicActorSystemProvider): ForwardProxy = {
249249
ForwardProxy(
250-
trustPem.fold(Http(system).defaultClientHttpsContext)(ForwardProxyHttpsContext(_)),
250+
trustPem.fold(Http(system.classicSystem).defaultClientHttpsContext)(ForwardProxyHttpsContext(_)),
251251
ForwardProxyPoolSettings(scheme, host, port, credentials)(system.classicSystem)
252252
)
253253
}

‎google-common/src/main/scala/akka/stream/alpakka/google/ResumableUpload.scala‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,7 @@ private[alpakka] object ResumableUpload {
9191
import implicits._
9292

9393
implicit val um: FromResponseUnmarshaller[Uri] = Unmarshaller.withMaterializer {
94-
implicit ec => implicit mat => response: HttpResponse =>
94+
implicit ec => implicit mat => (response: HttpResponse) =>
9595
response.discardEntityBytes().future.map { _ =>
9696
response.header[Location].fold(throw InvalidResponseException(ErrorInfo("No Location header")))(_.uri)
9797
}
@@ -107,7 +107,7 @@ private[alpakka] object ResumableUpload {
107107
)(implicit mat: Materializer): Flow[Either[T, MaybeLast[Chunk]], Try[Option[T]], NotUsed] = {
108108
implicit val system: ActorSystem = mat.system
109109

110-
val um = Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
110+
val um = Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
111111
response.status match {
112112
case PermanentRedirect =>
113113
response.discardEntityBytes().future.map(_ => None)
@@ -145,7 +145,7 @@ private[alpakka] object ResumableUpload {
145145
import implicits._
146146

147147
implicit val um: FromResponseUnmarshaller[Either[T, Long]] = Unmarshaller.withMaterializer {
148-
implicit ec => implicit mat => response: HttpResponse =>
148+
implicit ec => implicit mat => (response: HttpResponse) =>
149149
response.status match {
150150
case OK | Created => Unmarshal(response).to[T].map(Left(_))
151151
case PermanentRedirect =>

0 commit comments

Comments
 (0)