You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2023/06/08 23:00:59 UTC
[incubator-pekko-connectors] branch scala3 updated (585cacbca -> c54ee9320)
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a change to branch scala3
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors.git
omit 585cacbca Scala3 support for more google connectors (#143)
omit 1238ce6b2 support scala3 (google-common) (#141)
omit 41bd2a199 More scala3 work (cassandra / azure) (#130)
omit 732a73cd9 Scala3 couchbase support (#128)
omit d155fa68d try scala 3 (#123)
add d269db329 Use pekko-streams-circe for IronMQ
new efe1dd947 try scala 3 (#123)
new 40b4df526 Scala3 couchbase support (#128)
new 7af5577a0 More scala3 work (cassandra / azure) (#130)
new 99fd51ceb support scala3 (google-common) (#141)
new c54ee9320 Scala3 support for more google connectors (#143)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (585cacbca)
\
N -- N -- N refs/heads/scala3 (c54ee9320)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/pekko/stream/connectors/ironmq/impl/IronMqClient.scala | 2 +-
project/Dependencies.scala | 3 ++-
2 files changed, 3 insertions(+), 2 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
[incubator-pekko-connectors] 03/05: More scala3 work (cassandra / azure) (#130)
Posted by fa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch scala3
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors.git
commit 7af5577a04a8633181221ff6d4e7f083051579b5
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Wed May 31 08:23:09 2023 +0100
More scala3 work (cassandra / azure) (#130)
* scala3 support for cassandra connector
* try azure connector
---
.../connectors/azure/storagequeue/impl/AzureQueueSourceStage.scala | 2 +-
.../stream/connectors/cassandra/javadsl/CassandraSessionSpec.scala | 2 +-
.../connectors/cassandra/scaladsl/CassandraSessionPerformanceSpec.scala | 2 +-
project/Dependencies.scala | 2 --
4 files changed, 3 insertions(+), 5 deletions(-)
diff --git a/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/impl/AzureQueueSourceStage.scala b/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/impl/AzureQueueSourceStage.scala
index 6bfcc9a8e..712d2baac 100644
--- a/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/impl/AzureQueueSourceStage.scala
+++ b/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/impl/AzureQueueSourceStage.scala
@@ -68,7 +68,7 @@ import scala.collection.mutable.Queue
setHandler(
out,
new OutHandler {
- override def onPull: Unit =
+ override def onPull(): Unit =
if (!buffer.isEmpty) {
push(out, buffer.dequeue())
} else {
diff --git a/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/javadsl/CassandraSessionSpec.scala b/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/javadsl/CassandraSessionSpec.scala
index 6fa00ffa1..0c76ddb86 100644
--- a/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/javadsl/CassandraSessionSpec.scala
+++ b/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/javadsl/CassandraSessionSpec.scala
@@ -38,7 +38,7 @@ import scala.concurrent.duration._
final class CassandraSessionSpec extends CassandraSpecBase(ActorSystem("CassandraSessionSpec")) {
- val log = Logging(system, this.getClass)
+ val log = Logging(system, classOf[CassandraSessionSpec])
val javadslSessionRegistry = javadsl.CassandraSessionRegistry.get(system)
val data = 1 until 103
diff --git a/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraSessionPerformanceSpec.scala b/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraSessionPerformanceSpec.scala
index 843db7d2b..19ec7c6ac 100644
--- a/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraSessionPerformanceSpec.scala
+++ b/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraSessionPerformanceSpec.scala
@@ -25,7 +25,7 @@ import scala.concurrent.duration._
final class CassandraSessionPerformanceSpec extends CassandraSpecBase(ActorSystem("CassandraSessionPerformanceSpec")) {
- val log = Logging(system, this.getClass)
+ val log = Logging(system, classOf[CassandraSessionPerformanceSpec])
val data = 1 to 5 * 1000 * 1000
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index 89756b740..21f10c3f4 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -106,7 +106,6 @@ object Dependencies {
ExclusionRule("software.amazon.awssdk", "netty-nio-client"))) ++ Mockito)
val AzureStorageQueue = Seq(
- crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"com.microsoft.azure" % "azure-storage" % "8.0.0" // ApacheV2
))
@@ -116,7 +115,6 @@ object Dependencies {
val CassandraDriverVersionInDocs = "4.15"
val Cassandra = Seq(
- crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
("com.datastax.oss" % "java-driver-core" % CassandraDriverVersion)
.exclude("com.github.spotbugs", "spotbugs-annotations")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
[incubator-pekko-connectors] 02/05: Scala3 couchbase support (#128)
Posted by fa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch scala3
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors.git
commit 40b4df5268f88d196355fe4a20cc762758a2f1aa
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Tue May 30 20:55:16 2023 +0100
Scala3 couchbase support (#128)
* initial work on scala3 support for couchbase
* Update model.scala
* Update CouchbaseFlow.scala
* use jackson in test code instead of play-json
* Update Dependencies.scala
---
.../apache/pekko/stream/connectors/couchbase/model.scala | 9 +++++----
.../connectors/couchbase/scaladsl/CouchbaseFlow.scala | 16 +++++++++-------
.../connectors/couchbase/testing/CouchbaseSupport.scala | 15 +++++++++++----
project/Dependencies.scala | 6 +++---
4 files changed, 28 insertions(+), 18 deletions(-)
diff --git a/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/model.scala b/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/model.scala
index 0df164e88..1119810f2 100644
--- a/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/model.scala
+++ b/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/model.scala
@@ -270,7 +270,8 @@ sealed trait CouchbaseWriteResult[T <: Document[_]] {
/**
* Emitted for a successful Couchbase write operation.
*/
-final case class CouchbaseWriteSuccess[T <: Document[_]] private (override val doc: T) extends CouchbaseWriteResult[T] {
+final case class CouchbaseWriteSuccess[T <: Document[_]] private[couchbase] (
+ override val doc: T) extends CouchbaseWriteResult[T] {
val isSuccess: Boolean = true
val isFailure: Boolean = false
}
@@ -278,7 +279,7 @@ final case class CouchbaseWriteSuccess[T <: Document[_]] private (override val d
/**
* Emitted for a failed Couchbase write operation.
*/
-final case class CouchbaseWriteFailure[T <: Document[_]] private (override val doc: T, failure: Throwable)
+final case class CouchbaseWriteFailure[T <: Document[_]] private[couchbase] (override val doc: T, failure: Throwable)
extends CouchbaseWriteResult[T] {
val isSuccess: Boolean = false
val isFailure: Boolean = true
@@ -296,7 +297,7 @@ sealed trait CouchbaseDeleteResult {
/**
* Emitted for a successful Couchbase write operation.
*/
-final case class CouchbaseDeleteSuccess private (override val id: String) extends CouchbaseDeleteResult {
+final case class CouchbaseDeleteSuccess private[couchbase] (override val id: String) extends CouchbaseDeleteResult {
val isSuccess: Boolean = true
val isFailure: Boolean = false
}
@@ -304,7 +305,7 @@ final case class CouchbaseDeleteSuccess private (override val id: String) extend
/**
* Emitted for a failed Couchbase write operation.
*/
-final case class CouchbaseDeleteFailure private (override val id: String, failure: Throwable)
+final case class CouchbaseDeleteFailure private[couchbase] (override val id: String, failure: Throwable)
extends CouchbaseDeleteResult {
val isSuccess: Boolean = false
val isFailure: Boolean = true
diff --git a/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/CouchbaseFlow.scala b/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/CouchbaseFlow.scala
index 79039403d..faadb2e12 100644
--- a/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/CouchbaseFlow.scala
+++ b/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/CouchbaseFlow.scala
@@ -29,7 +29,7 @@ import pekko.stream.connectors.couchbase.{
import pekko.stream.scaladsl.Flow
import com.couchbase.client.java.document.{ Document, JsonDocument }
-import scala.concurrent.ExecutionContext
+import scala.concurrent.{ ExecutionContext, Future }
/**
* Scala API: Factory methods for Couchbase flows.
@@ -100,8 +100,8 @@ object CouchbaseFlow {
*/
def upsertDocWithResult[T <: Document[_]](sessionSettings: CouchbaseSessionSettings,
writeSettings: CouchbaseWriteSettings,
- bucketName: String): Flow[T, CouchbaseWriteResult[T], NotUsed] =
- Flow
+ bucketName: String): Flow[T, CouchbaseWriteResult[T], NotUsed] = {
+ val flow: Flow[T, CouchbaseWriteResult[T], Future[NotUsed]] = Flow
.fromMaterializer { (materializer, _) =>
val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
Flow[T]
@@ -115,7 +115,8 @@ object CouchbaseFlow {
}
})
}
- .mapMaterializedValue(_ => NotUsed)
+ flow.mapMaterializedValue(_ => NotUsed)
+ }
/**
* Create a flow to replace a Couchbase [[com.couchbase.client.java.document.JsonDocument JsonDocument]].
@@ -153,8 +154,8 @@ object CouchbaseFlow {
*/
def replaceDocWithResult[T <: Document[_]](sessionSettings: CouchbaseSessionSettings,
writeSettings: CouchbaseWriteSettings,
- bucketName: String): Flow[T, CouchbaseWriteResult[T], NotUsed] =
- Flow
+ bucketName: String): Flow[T, CouchbaseWriteResult[T], NotUsed] = {
+ val flow: Flow[T, CouchbaseWriteResult[T], Future[NotUsed]] = Flow
.fromMaterializer { (materializer, _) =>
val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
Flow[T]
@@ -168,7 +169,8 @@ object CouchbaseFlow {
}
})
}
- .mapMaterializedValue(_ => NotUsed)
+ flow.mapMaterializedValue(_ => NotUsed)
+ }
/**
* Create a flow to delete documents from Couchbase by `id`. Emits the same `id`.
diff --git a/couchbase/src/test/scala/org/apache/pekko/stream/connectors/couchbase/testing/CouchbaseSupport.scala b/couchbase/src/test/scala/org/apache/pekko/stream/connectors/couchbase/testing/CouchbaseSupport.scala
index a51fa055a..745f65478 100644
--- a/couchbase/src/test/scala/org/apache/pekko/stream/connectors/couchbase/testing/CouchbaseSupport.scala
+++ b/couchbase/src/test/scala/org/apache/pekko/stream/connectors/couchbase/testing/CouchbaseSupport.scala
@@ -25,8 +25,9 @@ import com.couchbase.client.deps.io.netty.util.CharsetUtil
import com.couchbase.client.java.ReplicateTo
import com.couchbase.client.java.document.json.JsonObject
import com.couchbase.client.java.document.{ BinaryDocument, JsonDocument, RawJsonDocument, StringDocument }
+import com.fasterxml.jackson.databind.json.JsonMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.slf4j.LoggerFactory
-import play.api.libs.json.Json
import scala.collection.immutable.Seq
import scala.concurrent.ExecutionContext.Implicits.global
@@ -35,6 +36,12 @@ import scala.concurrent.{ Await, Future }
case class TestObject(id: String, value: String)
+private[couchbase] object CouchbaseSupport {
+ val jacksonMapper = JsonMapper.builder()
+ .addModule(DefaultScalaModule)
+ .build()
+}
+
trait CouchbaseSupport {
private val log = LoggerFactory.getLogger(classOf[CouchbaseSupport])
@@ -64,7 +71,7 @@ trait CouchbaseSupport {
}
def toRawJsonDocument(testObject: TestObject): RawJsonDocument = {
- val json = Json.toJson(testObject)(Json.writes[TestObject]).toString()
+ val json = CouchbaseSupport.jacksonMapper.writeValueAsString(testObject)
RawJsonDocument.create(testObject.id, json)
}
@@ -72,12 +79,12 @@ trait CouchbaseSupport {
JsonDocument.create(testObject.id, JsonObject.create().put("id", testObject.id).put("value", testObject.value))
def toStringDocument(testObject: TestObject): StringDocument = {
- val json = Json.toJson(testObject)(Json.writes[TestObject]).toString()
+ val json = CouchbaseSupport.jacksonMapper.writeValueAsString(testObject)
StringDocument.create(testObject.id, json)
}
def toBinaryDocument(testObject: TestObject): BinaryDocument = {
- val json = Json.toJson(testObject)(Json.writes[TestObject]).toString()
+ val json = CouchbaseSupport.jacksonMapper.writeValueAsString(testObject)
val toWrite = Unpooled.copiedBuffer(json, CharsetUtil.UTF_8)
BinaryDocument.create(testObject.id, toWrite)
}
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index f8d605989..89756b740 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -125,13 +125,13 @@ object Dependencies {
"org.apache.pekko" %% "pekko-discovery" % PekkoVersion % Provided))
val Couchbase = Seq(
- crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"com.couchbase.client" % "java-client" % CouchbaseVersion, // ApacheV2
"io.reactivex" % "rxjava-reactive-streams" % "1.2.1", // ApacheV2
"org.apache.pekko" %% "pekko-discovery" % PekkoVersion % Provided, // Apache V2
- "com.typesafe.play" %% "play-json" % "2.9.2" % Test, // Apache V2
- "org.apache.pekko" %% "pekko-http" % PekkoHttpVersion % Test // Apache V2
+ "org.apache.pekko" %% "pekko-http" % PekkoHttpVersion % Test, // Apache V2
+ "com.fasterxml.jackson.core" % "jackson-databind" % JacksonDatabindVersion % Test, // Apache V2
+ "com.fasterxml.jackson.module" %% "jackson-module-scala" % JacksonDatabindVersion % Test // Apache V2
))
val `Doc-examples` = Seq(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
[incubator-pekko-connectors] 05/05: Scala3 support for more google connectors (#143)
Posted by fa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch scala3
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors.git
commit c54ee9320dd721135d880065b2efb402c92f2e4f
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Thu Jun 8 22:00:41 2023 +0100
Scala3 support for more google connectors (#143)
* support scala3 (google-common)
* support scala3 on more google connectors
* continue
* imports
* more implicit issues
* Update BigQueryCollectionFormats.scala
* further issues
* class access issues
* Update BigQueryRestBasicFormats.scala
* use compactPrint because toString causes tests to fail in Scala3
* review issue
---
.../elasticsearch/scaladsl/ElasticsearchFlow.scala | 2 +-
.../scala/docs/scaladsl/ElasticsearchV5Spec.scala | 6 ++---
.../scala/docs/scaladsl/ElasticsearchV7Spec.scala | 6 ++---
.../scala/docs/scaladsl/OpensearchV1Spec.scala | 6 ++---
.../bigquery/model/DatasetJsonProtocol.scala | 6 ++---
.../bigquery/model/ErrorProtoJsonProtocol.scala | 3 ++-
.../bigquery/model/JobJsonProtocol.scala | 21 +++++++++-------
.../bigquery/model/QueryJsonProtocol.scala | 9 ++++---
.../bigquery/model/TableDataJsonProtocol.scala | 13 +++++-----
.../bigquery/model/TableJsonProtocol.scala | 15 ++++++------
.../scaladsl/spray/BigQueryCollectionFormats.scala | 26 +++++++++++++-------
.../scaladsl/spray/BigQueryRestBasicFormats.scala | 24 +++++++++----------
.../src/test/scala/docs/scaladsl/BigQueryDoc.scala | 12 ++++++----
.../connectors/googlecloud/bigquery/e2e/A.scala | 4 ++--
.../e2e/scaladsl/BigQueryEndToEndSpec.scala | 18 +++++++-------
.../bigquery/scaladsl/BigQueryQueriesSpec.scala | 12 +++++-----
.../scaladsl/schema/BigQuerySchemasSpec.scala | 10 ++++----
.../scaladsl/spray/BigQueryJsonProtocolSpec.scala | 4 ++--
.../googlecloud/pubsub/impl/PubSubApi.scala | 28 ++++++++++++----------
.../googlecloud/pubsub/impl/PubSubApiSpec.scala | 2 +-
.../stream/connectors/google/GoogleSettings.scala | 6 +++--
.../stream/connectors/google/ResumableUpload.scala | 1 -
.../connectors/google/auth/NoCredentials.scala | 2 +-
project/Dependencies.scala | 2 --
24 files changed, 128 insertions(+), 110 deletions(-)
diff --git a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/scaladsl/ElasticsearchFlow.scala b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/scaladsl/ElasticsearchFlow.scala
index 4b9b6baa8..ce0744763 100644
--- a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/scaladsl/ElasticsearchFlow.scala
+++ b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/scaladsl/ElasticsearchFlow.scala
@@ -221,7 +221,7 @@ object ElasticsearchFlow {
}
private final class SprayJsonWriter[T](implicit writer: JsonWriter[T]) extends MessageWriter[T] {
- override def convert(message: T): String = message.toJson.toString()
+ override def convert(message: T): String = message.toJson.compactPrint
}
}
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala
index f8293cad9..bbba71dae 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala
@@ -158,9 +158,9 @@ class ElasticsearchV5Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
// #string
val write: Future[immutable.Seq[WriteResult[String, NotUsed]]] = Source(
immutable.Seq(
- WriteMessage.createIndexMessage("1", Book("Das Parfum").toJson.toString()),
- WriteMessage.createIndexMessage("2", Book("Faust").toJson.toString()),
- WriteMessage.createIndexMessage("3", Book("Die unendliche Geschichte").toJson.toString()))).via(
+ WriteMessage.createIndexMessage("1", Book("Das Parfum").toJson.compactPrint),
+ WriteMessage.createIndexMessage("2", Book("Faust").toJson.compactPrint),
+ WriteMessage.createIndexMessage("3", Book("Die unendliche Geschichte").toJson.compactPrint))).via(
ElasticsearchFlow.create(
constructElasticsearchParams(indexName, "_doc", ApiVersion.V5),
settings = baseWriteSettings,
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala
index 5eac2e95c..390fe9369 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala
@@ -147,9 +147,9 @@ class ElasticsearchV7Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
val write: Future[immutable.Seq[WriteResult[String, NotUsed]]] = Source(
immutable.Seq(
- WriteMessage.createIndexMessage("1", Book("Das Parfum").toJson.toString()),
- WriteMessage.createIndexMessage("2", Book("Faust").toJson.toString()),
- WriteMessage.createIndexMessage("3", Book("Die unendliche Geschichte").toJson.toString()))).via(
+ WriteMessage.createIndexMessage("1", Book("Das Parfum").toJson.compactPrint),
+ WriteMessage.createIndexMessage("2", Book("Faust").toJson.compactPrint),
+ WriteMessage.createIndexMessage("3", Book("Die unendliche Geschichte").toJson.compactPrint))).via(
ElasticsearchFlow.create(
constructElasticsearchParams(indexName, "_doc", ApiVersion.V7),
settings = baseWriteSettings,
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
index 7a486fd42..ecdb021e3 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
@@ -163,9 +163,9 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with ElasticsearchSpecUtils
// #string
val write: Future[immutable.Seq[WriteResult[String, NotUsed]]] = Source(
immutable.Seq(
- WriteMessage.createIndexMessage("1", Book("Das Parfum").toJson.toString()),
- WriteMessage.createIndexMessage("2", Book("Faust").toJson.toString()),
- WriteMessage.createIndexMessage("3", Book("Die unendliche Geschichte").toJson.toString()))).via(
+ WriteMessage.createIndexMessage("1", Book("Das Parfum").toJson.compactPrint),
+ WriteMessage.createIndexMessage("2", Book("Faust").toJson.compactPrint),
+ WriteMessage.createIndexMessage("3", Book("Die unendliche Geschichte").toJson.compactPrint))).via(
ElasticsearchFlow.create(
constructElasticsearchParams(indexName, "_doc", OpensearchApiVersion.V1),
settings = baseWriteSettings,
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/DatasetJsonProtocol.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/DatasetJsonProtocol.scala
index a9323a7f7..e56b10544 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/DatasetJsonProtocol.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/DatasetJsonProtocol.scala
@@ -32,7 +32,7 @@ import scala.collection.immutable.Seq
* @param labels the labels associated with this dataset
* @param location the geographic location where the dataset should reside
*/
-final case class Dataset private (datasetReference: DatasetReference,
+final case class Dataset private[bigquery] (datasetReference: DatasetReference,
friendlyName: Option[String],
labels: Option[Map[String, String]],
location: Option[String]) {
@@ -87,7 +87,7 @@ object Dataset {
* @param datasetId A unique ID for this dataset, without the project name
* @param projectId The ID of the project containing this dataset
*/
-final case class DatasetReference private (datasetId: Option[String], projectId: Option[String]) {
+final case class DatasetReference private[bigquery] (datasetId: Option[String], projectId: Option[String]) {
def getDatasetId = datasetId.toJava
def getProjectId = projectId.toJava
@@ -126,7 +126,7 @@ object DatasetReference {
* @param nextPageToken a token that can be used to request the next results page
* @param datasets an array of the dataset resources in the project
*/
-final case class DatasetListResponse private (nextPageToken: Option[String], datasets: Option[Seq[Dataset]]) {
+final case class DatasetListResponse private[bigquery] (nextPageToken: Option[String], datasets: Option[Seq[Dataset]]) {
def getNextPageToken = nextPageToken.toJava
def getDatasets = datasets.map(_.asJava).toJava
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/ErrorProtoJsonProtocol.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/ErrorProtoJsonProtocol.scala
index af887e7b5..dd814c8c0 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/ErrorProtoJsonProtocol.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/ErrorProtoJsonProtocol.scala
@@ -31,7 +31,8 @@ import scala.annotation.nowarn
* @param location specifies where the error occurred, if present
* @param message A human-readable description of the error
*/
-final case class ErrorProto private (reason: Option[String], location: Option[String], message: Option[String]) {
+final case class ErrorProto private[bigquery] (reason: Option[String], location: Option[String],
+ message: Option[String]) {
@nowarn("msg=never used")
@JsonCreator
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/JobJsonProtocol.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/JobJsonProtocol.scala
index 4799c0fe1..07b424e38 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/JobJsonProtocol.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/JobJsonProtocol.scala
@@ -33,7 +33,7 @@ import scala.collection.immutable.Seq
* @param jobReference reference describing the unique-per-user name of the job
* @param status the status of this job
*/
-final case class Job private (configuration: Option[JobConfiguration],
+final case class Job private[bigquery] (configuration: Option[JobConfiguration],
jobReference: Option[JobReference],
status: Option[JobStatus]) {
@@ -83,7 +83,8 @@ object Job {
* @param load configures a load job
* @param labels the labels associated with this job
*/
-final case class JobConfiguration private (load: Option[JobConfigurationLoad], labels: Option[Map[String, String]]) {
+final case class JobConfiguration private[bigquery] (load: Option[JobConfigurationLoad],
+ labels: Option[Map[String, String]]) {
def getLoad = load.toJava
def getLabels = labels.toJava
@@ -144,7 +145,7 @@ object JobConfiguration {
* @param writeDisposition specifies the action that occurs if the destination table already exists
* @param sourceFormat the format of the data files
*/
-final case class JobConfigurationLoad private (schema: Option[TableSchema],
+final case class JobConfigurationLoad private[bigquery] (schema: Option[TableSchema],
destinationTable: Option[TableReference],
createDisposition: Option[CreateDisposition],
writeDisposition: Option[WriteDisposition],
@@ -210,7 +211,7 @@ object JobConfigurationLoad {
implicit val configurationLoadFormat: JsonFormat[JobConfigurationLoad] = jsonFormat5(apply)
}
-final case class CreateDisposition private (value: String) extends StringEnum
+final case class CreateDisposition private[bigquery] (value: String) extends StringEnum
object CreateDisposition {
/**
@@ -227,7 +228,7 @@ object CreateDisposition {
implicit val format: JsonFormat[CreateDisposition] = StringEnum.jsonFormat(apply)
}
-final case class WriteDisposition private (value: String) extends StringEnum
+final case class WriteDisposition private[bigquery] (value: String) extends StringEnum
object WriteDisposition {
/**
@@ -269,7 +270,8 @@ object SourceFormat {
* @param jobId the ID of the job
* @param location the geographic location of the job
*/
-final case class JobReference private (projectId: Option[String], jobId: Option[String], location: Option[String]) {
+final case class JobReference private[bigquery] (projectId: Option[String], jobId: Option[String],
+ location: Option[String]) {
@nowarn("msg=never used")
@JsonCreator
@@ -323,7 +325,8 @@ object JobReference {
* @param errors the first errors encountered during the running of the job
* @param state running state of the job
*/
-final case class JobStatus private (errorResult: Option[ErrorProto], errors: Option[Seq[ErrorProto]], state: JobState) {
+final case class JobStatus private[bigquery] (errorResult: Option[ErrorProto], errors: Option[Seq[ErrorProto]],
+ state: JobState) {
def getErrorResult = errorResult.toJava
def getErrors = errors.map(_.asJava).toJava
@@ -360,7 +363,7 @@ object JobStatus {
implicit val format: JsonFormat[JobStatus] = jsonFormat3(apply)
}
-final case class JobState private (value: String) extends StringEnum
+final case class JobState private[bigquery] (value: String) extends StringEnum
object JobState {
/**
@@ -380,7 +383,7 @@ object JobState {
implicit val format: JsonFormat[JobState] = StringEnum.jsonFormat(apply)
}
-final case class JobCancelResponse private (job: Job) {
+final case class JobCancelResponse private[bigquery] (job: Job) {
def getJob = job
def withJob(job: Job) =
copy(job = job)
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/QueryJsonProtocol.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/QueryJsonProtocol.scala
index 5e66ee3cd..031a8d1d6 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/QueryJsonProtocol.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/QueryJsonProtocol.scala
@@ -21,11 +21,10 @@ import pekko.util.ccompat.JavaConverters._
import pekko.util.JavaDurationConverters._
import pekko.util.OptionConverters._
import com.fasterxml.jackson.annotation.{ JsonCreator, JsonIgnoreProperties, JsonProperty }
-import spray.json.{ RootJsonFormat, RootJsonReader }
+import spray.json.{ JsonFormat, RootJsonFormat, RootJsonReader }
import java.time.Duration
import java.{ lang, util }
-
import scala.annotation.nowarn
import scala.annotation.unchecked.uncheckedVariance
import scala.collection.immutable.Seq
@@ -46,7 +45,7 @@ import scala.concurrent.duration.FiniteDuration
* @param maximumBytesBilled limits the number of bytes billed for this query
* @param requestId a unique user provided identifier to ensure idempotent behavior for queries
*/
-final case class QueryRequest private (query: String,
+final case class QueryRequest private[bigquery] (query: String,
maxResults: Option[Int],
defaultDataset: Option[DatasetReference],
timeout: Option[FiniteDuration],
@@ -192,7 +191,7 @@ object QueryRequest {
* @tparam T the data model for each row
*/
@JsonIgnoreProperties(ignoreUnknown = true)
-final case class QueryResponse[+T] private (schema: Option[TableSchema],
+final case class QueryResponse[+T] private[bigquery] (schema: Option[TableSchema],
jobReference: JobReference,
totalRows: Option[Long],
pageToken: Option[String],
@@ -329,7 +328,7 @@ object QueryResponse {
implicit def reader[T <: AnyRef](
implicit reader: BigQueryRootJsonReader[T]): RootJsonReader[QueryResponse[T]] = {
- implicit val format = lift(reader)
+ implicit val format: JsonFormat[T] = lift(reader)
jsonFormat10(QueryResponse[T])
}
implicit val paginated: Paginated[QueryResponse[Any]] = _.pageToken
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableDataJsonProtocol.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableDataJsonProtocol.scala
index c370e7abf..030136af5 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableDataJsonProtocol.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableDataJsonProtocol.scala
@@ -38,7 +38,8 @@ import scala.collection.immutable.Seq
* @tparam T the data model of each row
*/
@JsonIgnoreProperties(ignoreUnknown = true)
-final case class TableDataListResponse[+T] private (totalRows: Long, pageToken: Option[String], rows: Option[Seq[T]]) {
+final case class TableDataListResponse[+T] private[bigquery] (totalRows: Long, pageToken: Option[String],
+ rows: Option[Seq[T]]) {
@nowarn("msg=never used")
@JsonCreator
@@ -82,7 +83,7 @@ object TableDataListResponse {
implicit def reader[T <: AnyRef](
implicit reader: BigQueryRootJsonReader[T]): RootJsonReader[TableDataListResponse[T]] = {
- implicit val format = lift(reader)
+ implicit val format: JsonFormat[T] = lift(reader)
jsonFormat3(TableDataListResponse[T])
}
implicit val paginated: Paginated[TableDataListResponse[Any]] = _.pageToken
@@ -99,7 +100,7 @@ object TableDataListResponse {
* @tparam T the data model of each row
*/
@JsonInclude(Include.NON_NULL)
-final case class TableDataInsertAllRequest[+T] private (skipInvalidRows: Option[Boolean],
+final case class TableDataInsertAllRequest[+T] private[bigquery] (skipInvalidRows: Option[Boolean],
ignoreUnknownValues: Option[Boolean],
templateSuffix: Option[String],
rows: Seq[Row[T]]) {
@@ -179,7 +180,7 @@ object TableDataInsertAllRequest {
* @param json the record this row contains
* @tparam T the data model of the record
*/
-final case class Row[+T] private (insertId: Option[String], json: T) {
+final case class Row[+T] private[bigquery] (insertId: Option[String], json: T) {
def getInsertId = insertId.toJava
def getJson = json
@@ -212,7 +213,7 @@ object Row {
* TableDataInsertAllResponse model
* @see [[https://cloud.google.com/bigquery/docs/reference/rest/v2/tabledata/insertAll#response-body BigQuery reference]]
*/
-final case class TableDataInsertAllResponse private (insertErrors: Option[Seq[InsertError]]) {
+final case class TableDataInsertAllResponse private[bigquery] (insertErrors: Option[Seq[InsertError]]) {
def getInsertErrors = insertErrors.map(_.asJava).toJava
def withInsertErrors(insertErrors: Option[Seq[InsertError]]) =
@@ -239,7 +240,7 @@ object TableDataInsertAllResponse {
* InsertError model
* @see [[https://cloud.google.com/bigquery/docs/reference/rest/v2/tabledata/insertAll#response-body BigQuery reference]]
*/
-final case class InsertError private (index: Int, errors: Option[Seq[ErrorProto]]) {
+final case class InsertError private[bigquery] (index: Int, errors: Option[Seq[ErrorProto]]) {
def getIndex = index
def getErrors = errors.map(_.asJava).toJava
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableJsonProtocol.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableJsonProtocol.scala
index 9d4427ae1..187fb5e8b 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableJsonProtocol.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableJsonProtocol.scala
@@ -37,7 +37,7 @@ import scala.collection.immutable.Seq
* @param numRows the number of rows of data in this table
* @param location the geographic location where the table resides
*/
-final case class Table private (tableReference: TableReference,
+final case class Table private[bigquery] (tableReference: TableReference,
labels: Option[Map[String, String]],
schema: Option[TableSchema],
numRows: Option[Long],
@@ -109,7 +109,8 @@ object Table {
* @param datasetId the ID of the dataset containing this table
* @param tableId the ID of the table
*/
-final case class TableReference private (projectId: Option[String], datasetId: String, tableId: Option[String]) {
+final case class TableReference private[bigquery] (projectId: Option[String], datasetId: String,
+ tableId: Option[String]) {
def getProjectId = projectId.toJava
def getDatasetId = datasetId
@@ -152,7 +153,7 @@ object TableReference {
*
* @param fields describes the fields in a table
*/
-final case class TableSchema private (fields: Seq[TableFieldSchema]) {
+final case class TableSchema private[bigquery] (fields: Seq[TableFieldSchema]) {
@nowarn("msg=never used")
@JsonCreator
@@ -200,7 +201,7 @@ object TableSchema {
* @param mode the field mode
* @param fields describes the nested schema fields if the type property is set to `RECORD`
*/
-final case class TableFieldSchema private (name: String,
+final case class TableFieldSchema private[bigquery] (name: String,
`type`: TableFieldSchemaType,
mode: Option[TableFieldSchemaMode],
fields: Option[Seq[TableFieldSchema]]) {
@@ -278,7 +279,7 @@ object TableFieldSchema {
jsonFormat(apply, "name", "type", "mode", "fields"))
}
-final case class TableFieldSchemaType private (value: String) extends StringEnum
+final case class TableFieldSchemaType private[bigquery] (value: String) extends StringEnum
object TableFieldSchemaType {
/**
@@ -328,7 +329,7 @@ object TableFieldSchemaType {
implicit val format: JsonFormat[TableFieldSchemaType] = StringEnum.jsonFormat(apply)
}
-final case class TableFieldSchemaMode private (value: String) extends StringEnum
+final case class TableFieldSchemaMode private[bigquery] (value: String) extends StringEnum
object TableFieldSchemaMode {
/**
@@ -356,7 +357,7 @@ object TableFieldSchemaMode {
* @param tables tables in the requested dataset
* @param totalItems the total number of tables in the dataset
*/
-final case class TableListResponse private (nextPageToken: Option[String],
+final case class TableListResponse private[bigquery] (nextPageToken: Option[String],
tables: Option[Seq[Table]],
totalItems: Option[Int]) {
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryCollectionFormats.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryCollectionFormats.scala
index b0c147b79..60171b025 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryCollectionFormats.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryCollectionFormats.scala
@@ -44,18 +44,26 @@ trait BigQueryCollectionFormats {
import collection.{ immutable => imm }
- implicit def immIterableFormat[T: BigQueryJsonFormat] = viaSeq[imm.Iterable[T], T](seq => imm.Iterable(seq: _*))
- implicit def immSeqFormat[T: BigQueryJsonFormat] = viaSeq[imm.Seq[T], T](seq => imm.Seq(seq: _*))
- implicit def immIndexedSeqFormat[T: BigQueryJsonFormat] = viaSeq[imm.IndexedSeq[T], T](seq => imm.IndexedSeq(seq: _*))
- implicit def immLinearSeqFormat[T: BigQueryJsonFormat] = viaSeq[imm.LinearSeq[T], T](seq => imm.LinearSeq(seq: _*))
- implicit def vectorFormat[T: BigQueryJsonFormat] = viaSeq[Vector[T], T](seq => Vector(seq: _*))
+ implicit def immIterableFormat[T: BigQueryJsonFormat]: BigQueryJsonFormat[imm.Iterable[T]] =
+ viaSeq[imm.Iterable[T], T](seq => imm.Iterable(seq: _*))
+ implicit def immSeqFormat[T: BigQueryJsonFormat]: BigQueryJsonFormat[imm.Seq[T]] =
+ viaSeq[imm.Seq[T], T](seq => imm.Seq(seq: _*))
+ implicit def immIndexedSeqFormat[T: BigQueryJsonFormat]: BigQueryJsonFormat[imm.IndexedSeq[T]] =
+ viaSeq[imm.IndexedSeq[T], T](seq => imm.IndexedSeq(seq: _*))
+ implicit def immLinearSeqFormat[T: BigQueryJsonFormat]: BigQueryJsonFormat[imm.LinearSeq[T]] =
+ viaSeq[imm.LinearSeq[T], T](seq => imm.LinearSeq(seq: _*))
+ implicit def vectorFormat[T: BigQueryJsonFormat]: BigQueryJsonFormat[Vector[T]] =
+ viaSeq[Vector[T], T](seq => Vector(seq: _*))
import collection._
- implicit def iterableFormat[T: BigQueryJsonFormat] = viaSeq[Iterable[T], T](seq => Iterable(seq: _*))
- implicit def seqFormat[T: BigQueryJsonFormat] = viaSeq[Seq[T], T](seq => Seq(seq: _*))
- implicit def indexedSeqFormat[T: BigQueryJsonFormat] = viaSeq[IndexedSeq[T], T](seq => IndexedSeq(seq: _*))
- implicit def linearSeqFormat[T: BigQueryJsonFormat] = viaSeq[LinearSeq[T], T](seq => LinearSeq(seq: _*))
+ implicit def iterableFormat[T: BigQueryJsonFormat]: BigQueryJsonFormat[Iterable[T]] =
+ viaSeq[Iterable[T], T](seq => Iterable(seq: _*))
+ implicit def seqFormat[T: BigQueryJsonFormat]: BigQueryJsonFormat[Seq[T]] = viaSeq[Seq[T], T](seq => Seq(seq: _*))
+ implicit def indexedSeqFormat[T: BigQueryJsonFormat]: BigQueryJsonFormat[IndexedSeq[T]] =
+ viaSeq[IndexedSeq[T], T](seq => IndexedSeq(seq: _*))
+ implicit def linearSeqFormat[T: BigQueryJsonFormat]: BigQueryJsonFormat[collection.LinearSeq[T]] =
+ viaSeq[collection.LinearSeq[T], T](seq => collection.LinearSeq(seq: _*))
/**
* A BigQueryJsonFormat construction helper that creates a BigQueryJsonFormat for an Iterable type I from a builder function
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryRestBasicFormats.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryRestBasicFormats.scala
index 24ee2a519..1338742ee 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryRestBasicFormats.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryRestBasicFormats.scala
@@ -22,18 +22,18 @@ import scala.concurrent.duration.{ DurationLong, FiniteDuration }
*/
trait BigQueryRestBasicFormats {
- implicit val IntJsonFormat = DefaultJsonProtocol.IntJsonFormat
- implicit val FloatJsonFormat = DefaultJsonProtocol.FloatJsonFormat
- implicit val DoubleJsonFormat = DefaultJsonProtocol.DoubleJsonFormat
- implicit val ByteJsonFormat = DefaultJsonProtocol.ByteJsonFormat
- implicit val ShortJsonFormat = DefaultJsonProtocol.ShortJsonFormat
- implicit val BigDecimalJsonFormat = DefaultJsonProtocol.BigDecimalJsonFormat
- implicit val BigIntJsonFormat = DefaultJsonProtocol.BigIntJsonFormat
- implicit val UnitJsonFormat = DefaultJsonProtocol.UnitJsonFormat
- implicit val BooleanJsonFormat = DefaultJsonProtocol.BooleanJsonFormat
- implicit val CharJsonFormat = DefaultJsonProtocol.CharJsonFormat
- implicit val StringJsonFormat = DefaultJsonProtocol.StringJsonFormat
- implicit val SymbolJsonFormat = DefaultJsonProtocol.SymbolJsonFormat
+ implicit val IntJsonFormat: JsonFormat[Int] = DefaultJsonProtocol.IntJsonFormat
+ implicit val FloatJsonFormat: JsonFormat[Float] = DefaultJsonProtocol.FloatJsonFormat
+ implicit val DoubleJsonFormat: JsonFormat[Double] = DefaultJsonProtocol.DoubleJsonFormat
+ implicit val ByteJsonFormat: JsonFormat[Byte] = DefaultJsonProtocol.ByteJsonFormat
+ implicit val ShortJsonFormat: JsonFormat[Short] = DefaultJsonProtocol.ShortJsonFormat
+ implicit val BigDecimalJsonFormat: JsonFormat[BigDecimal] = DefaultJsonProtocol.BigDecimalJsonFormat
+ implicit val BigIntJsonFormat: JsonFormat[BigInt] = DefaultJsonProtocol.BigIntJsonFormat
+ implicit val UnitJsonFormat: JsonFormat[Unit] = DefaultJsonProtocol.UnitJsonFormat
+ implicit val BooleanJsonFormat: JsonFormat[Boolean] = DefaultJsonProtocol.BooleanJsonFormat
+ implicit val CharJsonFormat: JsonFormat[Char] = DefaultJsonProtocol.CharJsonFormat
+ implicit val StringJsonFormat: JsonFormat[String] = DefaultJsonProtocol.StringJsonFormat
+ implicit val SymbolJsonFormat: JsonFormat[Symbol] = DefaultJsonProtocol.SymbolJsonFormat
implicit object BigQueryLongJsonFormat extends JsonFormat[Long] {
def write(x: Long) = JsNumber(x)
diff --git a/google-cloud-bigquery/src/test/scala/docs/scaladsl/BigQueryDoc.scala b/google-cloud-bigquery/src/test/scala/docs/scaladsl/BigQueryDoc.scala
index 2efa71f6e..5bfe9de59 100644
--- a/google-cloud-bigquery/src/test/scala/docs/scaladsl/BigQueryDoc.scala
+++ b/google-cloud-bigquery/src/test/scala/docs/scaladsl/BigQueryDoc.scala
@@ -28,9 +28,11 @@ import pekko.stream.connectors.googlecloud.bigquery.model.{
TableDataListResponse,
TableListResponse
}
-import pekko.stream.connectors.googlecloud.bigquery.scaladsl.BigQuery
import pekko.stream.connectors.googlecloud.bigquery.scaladsl.schema.BigQuerySchemas._
+import pekko.stream.connectors.googlecloud.bigquery.scaladsl.schema.TableSchemaWriter
+import pekko.stream.connectors.googlecloud.bigquery.scaladsl.spray.BigQueryRootJsonFormat
import pekko.stream.connectors.googlecloud.bigquery.scaladsl.spray.BigQueryJsonProtocol._
+import pekko.stream.connectors.googlecloud.bigquery.scaladsl.BigQuery
import pekko.stream.scaladsl.{ Flow, Sink, Source }
import pekko.{ Done, NotUsed }
@@ -48,8 +50,8 @@ class BigQueryDoc {
// #setup
case class Person(name: String, age: Int, addresses: Seq[Address], isHakker: Boolean)
case class Address(street: String, city: String, postalCode: Option[Int])
- implicit val addressFormat = bigQueryJsonFormat3(Address)
- implicit val personFormat = bigQueryJsonFormat4(Person)
+ implicit val addressFormat: BigQueryRootJsonFormat[Address] = bigQueryJsonFormat3(Address.apply)
+ implicit val personFormat: BigQueryRootJsonFormat[Person] = bigQueryJsonFormat4(Person.apply)
// #setup
@nowarn("msg=dead code")
@@ -113,8 +115,8 @@ class BigQueryDoc {
// #table-methods
// #create-table
- implicit val addressSchema = bigQuerySchema3(Address)
- implicit val personSchema = bigQuerySchema4(Person)
+ implicit val addressSchema: TableSchemaWriter[Address] = bigQuerySchema3(Address.apply)
+ implicit val personSchema: TableSchemaWriter[Person] = bigQuerySchema4(Person.apply)
val newTable: Future[Table] = BigQuery.createTable[Person](datasetId, "newTableId")
// #create-table
diff --git a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/e2e/A.scala b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/e2e/A.scala
index 21ee0cb22..0335213d2 100644
--- a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/e2e/A.scala
+++ b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/e2e/A.scala
@@ -39,7 +39,7 @@ case class A(integer: Int, long: Long, float: Float, double: Double, string: Str
new B(f.get(6).get("v")))
def getInteger = integer
- @JsonSerialize(using = classOf[ToStringSerializer])
+ @JsonSerialize(`using` = classOf[ToStringSerializer])
def getLong = long
def getFloat = float
def getDouble = double
@@ -74,7 +74,7 @@ case class C(numeric: BigDecimal, date: LocalDate, time: LocalTime, dateTime: Lo
LocalDateTime.parse(node.get("f").get(3).get("v").textValue()),
Instant.ofEpochMilli((BigDecimal(node.get("f").get(4).get("v").textValue()) * 1000).toLong))
- @JsonSerialize(using = classOf[ToStringSerializer])
+ @JsonSerialize(`using` = classOf[ToStringSerializer])
def getNumeric = numeric
def getDate = date
def getTime = time
diff --git a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/e2e/scaladsl/BigQueryEndToEndSpec.scala b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/e2e/scaladsl/BigQueryEndToEndSpec.scala
index 649cc16ce..52f087243 100644
--- a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/e2e/scaladsl/BigQueryEndToEndSpec.scala
+++ b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/e2e/scaladsl/BigQueryEndToEndSpec.scala
@@ -14,12 +14,14 @@
package org.apache.pekko.stream.connectors.googlecloud.bigquery.e2e.scaladsl
import org.apache.pekko
-import pekko.actor.ActorSystem
+import pekko.actor.{ ActorSystem, Scheduler }
import pekko.{ pattern, Done }
import pekko.stream.connectors.googlecloud.bigquery.HoverflySupport
import pekko.stream.connectors.googlecloud.bigquery.e2e.{ A, B, C }
import pekko.stream.connectors.googlecloud.bigquery.model.JobState
import pekko.stream.connectors.googlecloud.bigquery.model.TableReference
+import pekko.stream.connectors.googlecloud.bigquery.scaladsl.schema.TableSchemaWriter
+import pekko.stream.connectors.googlecloud.bigquery.scaladsl.spray.BigQueryRootJsonFormat
import pekko.testkit.TestKit
import io.specto.hoverfly.junit.core.{ HoverflyMode, SimulationSource }
import org.scalatest.BeforeAndAfterAll
@@ -56,7 +58,7 @@ class BigQueryEndToEndSpec
super.afterAll()
}
- implicit def scheduler = system.scheduler
+ implicit def scheduler: Scheduler = system.scheduler
"BigQuery Scala DSL" should {
@@ -66,12 +68,12 @@ class BigQueryEndToEndSpec
import pekko.stream.connectors.googlecloud.bigquery.scaladsl.spray.BigQueryJsonProtocol._
import pekko.stream.scaladsl.{ Sink, Source }
- implicit val cFormat = bigQueryJsonFormat5(C)
- implicit val bFormat = bigQueryJsonFormat3(B)
- implicit val aFormat = bigQueryJsonFormat7(A)
- implicit val cSchema = bigQuerySchema5(C)
- implicit val bSchema = bigQuerySchema3(B)
- implicit val aSchema = bigQuerySchema7(A)
+ implicit val cFormat: BigQueryRootJsonFormat[C] = bigQueryJsonFormat5(C.apply)
+ implicit val bFormat: BigQueryRootJsonFormat[B] = bigQueryJsonFormat3(B.apply)
+ implicit val aFormat: BigQueryRootJsonFormat[A] = bigQueryJsonFormat7(A.apply)
+ implicit val cSchema: TableSchemaWriter[C] = bigQuerySchema5(C.apply)
+ implicit val bSchema: TableSchemaWriter[B] = bigQuerySchema3(B.apply)
+ implicit val aSchema: TableSchemaWriter[A] = bigQuerySchema7(A.apply)
"create dataset" in {
BigQuery.createDataset(datasetId).map { dataset =>
diff --git a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueriesSpec.scala b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueriesSpec.scala
index f98c30999..169f6325c 100644
--- a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueriesSpec.scala
+++ b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueriesSpec.scala
@@ -94,7 +94,7 @@ class BigQueryQueriesSpec
.post(BigQueryEndpoints.queries(settings.projectId).path.toString)
.queryParam("prettyPrint", "false")
.anyBody()
- .willReturn(success(completeQuery.toJson.toString(), "application/json"))))
+ .willReturn(success(completeQuery.toJson.compactPrint, "application/json"))))
query[JsValue]("SQL")
.addAttributes(GoogleAttributes.settings(settings))
@@ -111,7 +111,7 @@ class BigQueryQueriesSpec
.post(BigQueryEndpoints.queries(settings.projectId).path.toString)
.queryParam("prettyPrint", "false")
.anyBody()
- .willReturn(success(completeQueryWith2ndPage.toJson.toString(), "application/json"))
+ .willReturn(success(completeQueryWith2ndPage.toJson.compactPrint, "application/json"))
.get(BigQueryEndpoints.query(settings.projectId, jobId).path.toString)
.queryParam("pageToken", pageToken)
.queryParam("prettyPrint", "false")
@@ -132,7 +132,7 @@ class BigQueryQueriesSpec
.post(BigQueryEndpoints.queries(settings.projectId).path.toString)
.queryParam("prettyPrint", "false")
.anyBody()
- .willReturn(success(incompleteQuery.toJson.toString(), "application/json"))
+ .willReturn(success(incompleteQuery.toJson.compactPrint, "application/json"))
.get(BigQueryEndpoints.query(settings.projectId, jobId).path.toString)
.queryParam("prettyPrint", "false")
.willReturn(success(completeQuery.toJson.toString, "application/json"))))
@@ -152,7 +152,7 @@ class BigQueryQueriesSpec
.post(BigQueryEndpoints.queries(settings.projectId).path.toString)
.queryParam("prettyPrint", "false")
.anyBody()
- .willReturn(success(incompleteQuery.toJson.toString(), "application/json"))
+ .willReturn(success(incompleteQuery.toJson.compactPrint, "application/json"))
.get(BigQueryEndpoints.query(settings.projectId, jobId).path.toString)
.queryParam("prettyPrint", "false")
.willReturn(success(completeQueryWith2ndPage.toJson.toString, "application/json"))
@@ -176,7 +176,7 @@ class BigQueryQueriesSpec
.post(BigQueryEndpoints.queries(settings.projectId).path.toString)
.queryParam("prettyPrint", "false")
.anyBody()
- .willReturn(success(completeQueryWithoutJobId.toJson.toString(), "application/json"))))
+ .willReturn(success(completeQueryWithoutJobId.toJson.compactPrint, "application/json"))))
query[JsValue]("SQL")
.addAttributes(GoogleAttributes.settings(settings))
@@ -204,7 +204,7 @@ class BigQueryQueriesSpec
.post(BigQueryEndpoints.queries(settings.projectId).path.toString)
.queryParam("prettyPrint", "false")
.anyBody()
- .willReturn(success(completeQuery.toJson.toString(), "application/json"))))
+ .willReturn(success(completeQuery.toJson.compactPrint, "application/json"))))
recoverToSucceededIf[BrokenParserException] {
query[JsValue]("SQL")
diff --git a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/schema/BigQuerySchemasSpec.scala b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/schema/BigQuerySchemasSpec.scala
index cd4ab44af..2e4a7178b 100644
--- a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/schema/BigQuerySchemasSpec.scala
+++ b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/schema/BigQuerySchemasSpec.scala
@@ -50,20 +50,20 @@ class BigQuerySchemasSpec extends AnyWordSpecLike with Matchers {
"BigQuerySchemas" should {
"correctly generate schema" in {
- implicit val cSchemaWriter = bigQuerySchema1(C)
- implicit val bSchemaWriter = bigQuerySchema2(B)
- val generatedSchema = bigQuerySchema7(A).write
+ implicit val cSchemaWriter: TableSchemaWriter[C] = bigQuerySchema1(C.apply)
+ implicit val bSchemaWriter: TableSchemaWriter[B] = bigQuerySchema2(B.apply)
+ val generatedSchema = bigQuerySchema7(A.apply).write
generatedSchema shouldEqual schema
}
"throw exception when nesting options" in {
case class Invalid(invalid: Option[Option[String]])
- assertThrows[IllegalArgumentException](bigQuerySchema1(Invalid).write)
+ assertThrows[IllegalArgumentException](bigQuerySchema1(Invalid.apply).write)
}
"throw exception when nesting options inside seqs" in {
case class Invalid(invalid: Seq[Option[String]])
- assertThrows[IllegalArgumentException](bigQuerySchema1(Invalid).write)
+ assertThrows[IllegalArgumentException](bigQuerySchema1(Invalid.apply).write)
}
}
}
diff --git a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryJsonProtocolSpec.scala b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryJsonProtocolSpec.scala
index b70fd4745..a8fe82084 100644
--- a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryJsonProtocolSpec.scala
+++ b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/spray/BigQueryJsonProtocolSpec.scala
@@ -59,8 +59,8 @@ class BigQueryJsonProtocolSpec extends BigQueryJsonProtocol with AnyWordSpecLike
case class Record(name: Option[String], addresses: Seq[Address])
case class Address(street: Option[String], city: Option[String])
- implicit val addressFormat = bigQueryJsonFormat2(Address)
- implicit val recordFormat = bigQueryJsonFormat2(Record)
+ implicit val addressFormat: BigQueryRootJsonFormat[Address] = bigQueryJsonFormat2(Address.apply)
+ implicit val recordFormat: BigQueryRootJsonFormat[Record] = bigQueryJsonFormat2(Record.apply)
"BigQueryJsonProtocol" should {
diff --git a/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala b/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala
index ccabe2828..0eebe0ace 100644
--- a/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala
+++ b/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala
@@ -69,7 +69,7 @@ private[pubsub] trait PubSubApi {
def PubSubGoogleApisPort: Int
def isEmulated: Boolean
- private implicit val instantFormat = new RootJsonFormat[Instant] {
+ private implicit val instantFormat: RootJsonFormat[Instant] = new RootJsonFormat[Instant] {
override def read(jsValue: JsValue): Instant = jsValue match {
case JsString(time) => Instant.parse(time)
case _ => deserializationError("Instant required as a string of RFC3339 UTC Zulu format.")
@@ -77,7 +77,7 @@ private[pubsub] trait PubSubApi {
override def write(instant: Instant): JsValue = JsString(instant.toString)
}
- private implicit val pubSubMessageFormat =
+ private implicit val pubSubMessageFormat: RootJsonFormat[PubSubMessage] =
new RootJsonFormat[PubSubMessage] {
override def read(json: JsValue): PubSubMessage = {
val fields = json.asJsObject.fields
@@ -98,7 +98,7 @@ private[pubsub] trait PubSubApi {
++ m.attributes.map(attributes => "attributes" -> attributes.toJson): _*)
}
- private implicit val publishMessageFormat = new RootJsonFormat[PublishMessage] {
+ private implicit val publishMessageFormat: RootJsonFormat[PublishMessage] = new RootJsonFormat[PublishMessage] {
def read(json: JsValue): PublishMessage = {
val data = json.asJsObject.fields("data").convertTo[String]
val attributes = json.asJsObject.fields("attributes").convertTo[immutable.Map[String, String]]
@@ -112,37 +112,39 @@ private[pubsub] trait PubSubApi {
m.attributes.map(a => "attributes" -> a.toJson): _*)
}
- private implicit val pubSubRequestFormat = new RootJsonFormat[PublishRequest] {
+ private implicit val pubSubRequestFormat: RootJsonFormat[PublishRequest] = new RootJsonFormat[PublishRequest] {
def read(json: JsValue): PublishRequest =
PublishRequest(json.asJsObject.fields("messages").convertTo[immutable.Seq[PublishMessage]])
def write(pr: PublishRequest): JsValue = JsObject("messages" -> pr.messages.toJson)
}
- private implicit val gcePubSubResponseFormat = new RootJsonFormat[PublishResponse] {
+ private implicit val gcePubSubResponseFormat: RootJsonFormat[PublishResponse] = new RootJsonFormat[PublishResponse] {
def read(json: JsValue): PublishResponse =
PublishResponse(json.asJsObject.fields("messageIds").convertTo[immutable.Seq[String]])
def write(pr: PublishResponse): JsValue = JsObject("messageIds" -> pr.messageIds.toJson)
}
- private implicit val receivedMessageFormat = new RootJsonFormat[ReceivedMessage] {
+ private implicit val receivedMessageFormat: RootJsonFormat[ReceivedMessage] = new RootJsonFormat[ReceivedMessage] {
def read(json: JsValue): ReceivedMessage =
ReceivedMessage(json.asJsObject.fields("ackId").convertTo[String],
json.asJsObject.fields("message").convertTo[PubSubMessage])
def write(rm: ReceivedMessage): JsValue =
JsObject("ackId" -> rm.ackId.toJson, "message" -> rm.message.toJson)
}
- private implicit val pubSubPullResponseFormat = new RootJsonFormat[PullResponse] {
+ private implicit val pubSubPullResponseFormat: RootJsonFormat[PullResponse] = new RootJsonFormat[PullResponse] {
def read(json: JsValue): PullResponse =
PullResponse(json.asJsObject.fields.get("receivedMessages").map(_.convertTo[immutable.Seq[ReceivedMessage]]))
def write(pr: PullResponse): JsValue =
pr.receivedMessages.map(rm => JsObject("receivedMessages" -> rm.toJson)).getOrElse(JsObject.empty)
}
- private implicit val acknowledgeRequestFormat = new RootJsonFormat[AcknowledgeRequest] {
- def read(json: JsValue): AcknowledgeRequest =
- AcknowledgeRequest(json.asJsObject.fields("ackIds").convertTo[immutable.Seq[String]]: _*)
- def write(ar: AcknowledgeRequest): JsValue = JsObject("ackIds" -> ar.ackIds.toJson)
- }
- private implicit val pullRequestFormat = DefaultJsonProtocol.jsonFormat2(PullRequest.apply)
+ private implicit val acknowledgeRequestFormat: RootJsonFormat[AcknowledgeRequest] =
+ new RootJsonFormat[AcknowledgeRequest] {
+ def read(json: JsValue): AcknowledgeRequest =
+ AcknowledgeRequest(json.asJsObject.fields("ackIds").convertTo[immutable.Seq[String]]: _*)
+ def write(ar: AcknowledgeRequest): JsValue = JsObject("ackIds" -> ar.ackIds.toJson)
+ }
+ private implicit val pullRequestFormat: RootJsonFormat[PullRequest] =
+ DefaultJsonProtocol.jsonFormat2(PullRequest.apply)
private def scheme: String = if (isEmulated) "http" else "https"
diff --git a/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApiSpec.scala b/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApiSpec.scala
index bd8a57f78..d22c7b35d 100644
--- a/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApiSpec.scala
+++ b/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApiSpec.scala
@@ -60,7 +60,7 @@ class PubSubApiSpec extends AnyFlatSpec with BeforeAndAfterAll with ScalaFutures
s"pekko.connectors.google.credentials.none.project-id = ${TestCredentials.projectId}")
.withFallback(ConfigFactory.load()))
- implicit val defaultPatience =
+ implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = 5.seconds, interval = 100.millis)
def createInsecureSslEngine(host: String, port: Int): SSLEngine = {
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/GoogleSettings.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/GoogleSettings.scala
index 973cdb194..dd6e1725c 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/GoogleSettings.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/GoogleSettings.scala
@@ -90,7 +90,8 @@ object GoogleSettings {
}
-final case class GoogleSettings @InternalApi private[connectors] (projectId: String,
+@InternalApi
+final case class GoogleSettings(projectId: String,
credentials: Credentials,
requestSettings: RequestSettings) {
def getProjectId = projectId
@@ -134,7 +135,8 @@ object RequestSettings {
apply(userIp.toScala, quotaUser.toScala, prettyPrint, chunkSize, retrySettings, forwardProxy.toScala)
}
-final case class RequestSettings @InternalApi private[connectors] (
+@InternalApi
+final case class RequestSettings(
userIp: Option[String],
quotaUser: Option[String],
prettyPrint: Boolean,
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
index 81f9e7db6..71ac9641f 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
@@ -14,7 +14,6 @@
package org.apache.pekko.stream.connectors.google
import org.apache.pekko
-import pekko.actor.ActorSystem
import pekko.NotUsed
import pekko.annotation.InternalApi
import pekko.http.scaladsl.model.HttpMethods.{ POST, PUT }
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/NoCredentials.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/NoCredentials.scala
index ead4e2da0..87a202c3e 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/NoCredentials.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/NoCredentials.scala
@@ -32,7 +32,7 @@ private[connectors] object NoCredentials {
}
@InternalApi
-private[auth] final case class NoCredentials private (projectId: String, token: String) extends Credentials {
+private[connectors] final case class NoCredentials(projectId: String, token: String) extends Credentials {
private val futureToken = Future.successful(OAuth2BearerToken(token))
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index 6f7301efb..4cb19083d 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -206,7 +206,6 @@ object Dependencies {
) ++ Mockito)
val GoogleBigQuery = Seq(
- crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-jackson" % PekkoHttpVersion % Provided,
@@ -233,7 +232,6 @@ object Dependencies {
"org.apache.pekko" %% "pekko-discovery" % PekkoVersion) ++ Mockito)
val GooglePubSub = Seq(
- crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
[incubator-pekko-connectors] 01/05: try scala 3 (#123)
Posted by fa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch scala3
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors.git
commit efe1dd9471c3b76e3802586223a7fb98c9435b2e
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Tue May 30 14:44:18 2023 +0100
try scala 3 (#123)
---
.../amqp/impl/AmqpReplyToSinkStage.scala | 4 +-
.../connectors/amqp/impl/AmqpRpcFlowStage.scala | 5 +-
.../connectors/amqp/impl/AmqpSimpleFlowStage.scala | 4 +-
.../connectors/amqp/scaladsl/AmqpSource.scala | 4 +-
.../test/scala/docs/scaladsl/AmqpDocsSpec.scala | 2 +-
.../pekko/stream/connectors/amqp/AmqpSpec.scala | 6 +-
...AmqpGraphStageLogicConnectionShutdownSpec.scala | 9 ++-
.../docs/scaladsl/EventBridgePublisherSpec.scala | 2 +-
.../scala/docs/scaladsl/AwsLambdaFlowSpec.scala | 7 +-
.../storagequeue/javadsl/AzureQueueSink.scala | 2 +-
.../cassandra/javadsl/CassandraSession.scala | 2 +-
.../cassandra/scaladsl/CassandraLifecycle.scala | 5 +-
.../couchbase/scaladsl/CouchbaseFlow.scala | 10 +--
.../couchbase/scaladsl/DiscoverySupport.scala | 6 +-
.../stream/connectors/csv/scaladsl/CsvBench.scala | 9 ++-
csv/src/test/scala/docs/scaladsl/CsvSpec.scala | 2 +-
.../eip/scaladsl/PassThroughExamples.scala | 4 +-
.../src/test/scala/docs/scaladsl/ExampleSpec.scala | 3 +-
.../scaladsl/ElasticsearchConnectorBehaviour.scala | 6 +-
.../docs/scaladsl/ElasticsearchSpecUtils.scala | 2 +-
.../scala/docs/scaladsl/ElasticsearchV5Spec.scala | 18 ++---
.../scala/docs/scaladsl/ElasticsearchV7Spec.scala | 18 ++---
.../scaladsl/OpensearchConnectorBehaviour.scala | 6 +-
.../scala/docs/scaladsl/OpensearchV1Spec.scala | 18 ++---
.../connectors/file/javadsl/LogRotatorSink.scala | 33 +++++----
.../connectors/file/scaladsl/LogRotatorSink.scala | 2 +-
.../scala/docs/scaladsl/LogRotatorSinkSpec.scala | 4 +-
.../stream/connectors/ftp/CommonFtpStageSpec.scala | 13 ++--
.../test/scala/docs/scaladsl/GeodeBaseSpec.scala | 2 +-
.../googlecloud/bigquery/BigQueryException.scala | 2 +-
.../bigquery/model/TableDataJsonProtocol.scala | 5 +-
.../bigquery/scaladsl/BigQueryDatasets.scala | 4 +-
.../bigquery/scaladsl/BigQueryJobs.scala | 8 +--
.../bigquery/scaladsl/BigQueryQueries.scala | 11 +--
.../bigquery/scaladsl/BigQueryTableData.scala | 7 +-
.../bigquery/scaladsl/BigQueryTables.scala | 4 +-
.../bigquery/scaladsl/BigQueryQueriesSpec.scala | 2 +-
.../test/scala/docs/scaladsl/IntegrationSpec.scala | 2 +-
.../googlecloud/pubsub/impl/PubSubApi.scala | 16 ++---
.../test/scala/docs/scaladsl/ExampleUsage.scala | 2 +-
.../test/scala/docs/scaladsl/IntegrationSpec.scala | 2 +-
.../googlecloud/pubsub/GooglePubSubSpec.scala | 4 +-
.../googlecloud/pubsub/impl/PubSubApiSpec.scala | 2 +-
.../googlecloud/storage/impl/Formats.scala | 17 +++--
.../googlecloud/storage/impl/GCStorageStream.scala | 8 +--
.../googlecloud/storage/GCSExtSpec.scala | 2 +-
.../googlecloud/storage/GCStorageExtSpec.scala | 2 +-
.../storage/WithMaterializerGlobal.scala | 4 +-
.../connectors/google/PaginatedRequest.scala | 5 +-
.../stream/connectors/google/ResumableUpload.scala | 9 +--
.../google/auth/GoogleComputeMetadata.scala | 5 +-
.../connectors/google/auth/GoogleOAuth2.scala | 5 +-
.../google/auth/GoogleOAuth2Credentials.scala | 2 +-
.../google/auth/GoogleOAuth2Exception.scala | 3 +-
.../google/auth/ServiceAccountCredentials.scala | 2 +-
.../google/auth/UserAccessCredentials.scala | 2 +-
.../google/auth/UserAccessMetadata.scala | 3 +-
.../stream/connectors/google/http/GoogleHttp.scala | 6 +-
.../connectors/google/PaginatedRequestSpec.scala | 2 +-
.../connectors/google/ResumableUploadSpec.scala | 2 +-
.../connectors/google/auth/GoogleOAuth2Spec.scala | 8 +--
.../google/auth/OAuth2CredentialsSpec.scala | 3 +-
.../connectors/google/http/GoogleHttpSpec.scala | 4 +-
.../google/firebase/fcm/impl/FcmFlows.scala | 4 +-
.../google/firebase/fcm/impl/FcmJsonSupport.scala | 10 +--
.../firebase/fcm/v1/impl/FcmJsonSupport.scala | 4 +-
.../src/test/scala/docs/scaladsl/FcmExamples.scala | 4 +-
.../firebase/fcm/v1/impl/FcmSenderSpec.scala | 4 +-
.../connectors/hbase/impl/HBaseFlowStage.scala | 2 +-
.../connectors/hbase/impl/HBaseSourceStage.scala | 4 +-
.../test/scala/docs/scaladsl/HBaseStageSpec.scala | 2 +-
.../test/scala/docs/scaladsl/HdfsWriterSpec.scala | 2 +-
.../huawei/pushkit/impl/PushKitJsonSupport.scala | 2 +-
.../test/scala/docs/scaladsl/PushKitExamples.scala | 4 +-
.../huawei/pushkit/impl/HmsTokenApiSpec.scala | 2 +-
.../huawei/pushkit/impl/PushKitSenderSpec.scala | 2 +-
.../src/test/scala/docs/scaladsl/FlowSpec.scala | 6 +-
.../scala/docs/scaladsl/InfluxDbSourceSpec.scala | 2 +-
.../test/scala/docs/scaladsl/InfluxDbSpec.scala | 4 +-
.../connectors/ironmq/impl/IronMqPullStage.scala | 4 +-
.../stream/connectors/ironmq/IronMqSpec.scala | 2 +-
.../scaladsl/JmsBufferedAckConnectorsSpec.scala | 2 +-
.../docs/scaladsl/JmsIbmmqConnectorsSpec.scala | 2 +-
.../scala/docs/scaladsl/JmsTxConnectorsSpec.scala | 2 +-
.../pekko/stream/connectors/jms/JmsSpec.scala | 2 +-
.../jms/scaladsl/JmsAckConnectorsSpec.scala | 2 +-
.../stream/connectors/kinesis/ShardIterator.scala | 2 +-
.../kinesis/KinesisSchedulerSourceSpec.scala | 36 +++++-----
.../test/scala/docs/scaladsl/MongoSinkSpec.scala | 9 +--
.../test/scala/docs/scaladsl/MongoSourceSpec.scala | 4 +-
.../test/scala/docs/scaladsl/OrientDbSpec.scala | 12 ++--
.../scala/docs/scaladsl/PravegaReadWriteDocs.scala | 12 ++--
.../scala/docs/scaladsl/PravegaSettingsSpec.scala | 4 +-
.../src/test/scala/docs/scaladsl/Serializers.scala | 6 +-
project/Common.scala | 24 +++++--
project/Dependencies.scala | 32 ++++++++-
.../stream/connectors/reference/Resource.scala | 2 +-
.../connectors/s3/impl/HttpRequestsSpec.scala | 79 +++++++++++++---------
.../stream/connectors/s3/impl/S3StreamSpec.scala | 4 +-
.../connectors/s3/impl/auth/SignerSpec.scala | 2 +-
.../src/test/scala/docs/scaladsl/DocSnippets.scala | 22 +++---
slick/src/test/scala/docs/scaladsl/SlickSpec.scala | 10 +--
.../scala/docs/scaladsl/SnsPublisherSpec.scala | 2 +-
solr/src/test/scala/docs/scaladsl/SolrSpec.scala | 32 ++++-----
sqs/src/test/scala/docs/scaladsl/SqsAckSpec.scala | 8 +--
.../test/scala/docs/scaladsl/SqsSourceSpec.scala | 2 +-
.../test/scala/docs/scaladsl/EventSourceSpec.scala | 8 +--
udp/src/test/scala/docs/scaladsl/UdpSpec.scala | 4 +-
.../scaladsl/UnixDomainSocket.scala | 2 +-
109 files changed, 416 insertions(+), 349 deletions(-)
diff --git a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpReplyToSinkStage.scala b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpReplyToSinkStage.scala
index 641b2b6f5..4415c6290 100644
--- a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpReplyToSinkStage.scala
+++ b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpReplyToSinkStage.scala
@@ -28,7 +28,7 @@ import scala.concurrent.{ Future, Promise }
* the queue named in the replyTo options of the message instead of from settings declared at construction.
*/
@InternalApi
-private[amqp] final class AmqpReplyToSinkStage(settings: AmqpReplyToSinkSettings)
+private[amqp] final class AmqpReplyToSinkStage(replyToSinkSettings: AmqpReplyToSinkSettings)
extends GraphStageWithMaterializedValue[SinkShape[WriteMessage], Future[Done]] { stage =>
val in = Inlet[WriteMessage]("AmqpReplyToSink.in")
@@ -41,7 +41,7 @@ private[amqp] final class AmqpReplyToSinkStage(settings: AmqpReplyToSinkSettings
override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Future[Done]) = {
val streamCompletion = Promise[Done]()
(new GraphStageLogic(shape) with AmqpConnectorLogic {
- override val settings = stage.settings
+ override val settings: AmqpReplyToSinkSettings = stage.replyToSinkSettings
override def whenConnected(): Unit = pull(in)
diff --git a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpRpcFlowStage.scala b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpRpcFlowStage.scala
index 186f85ec5..39c7a6d5c 100644
--- a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpRpcFlowStage.scala
+++ b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpRpcFlowStage.scala
@@ -37,7 +37,8 @@ import scala.util.Success
* can be overridden per message by including `expectedReplies` in the the header of the [[pekko.stream.connectors.amqp.WriteMessage]]
*/
@InternalApi
-private[amqp] final class AmqpRpcFlowStage(settings: AmqpWriteSettings, bufferSize: Int, responsesPerMessage: Int = 1)
+private[amqp] final class AmqpRpcFlowStage(writeSettings: AmqpWriteSettings, bufferSize: Int,
+ responsesPerMessage: Int = 1)
extends GraphStageWithMaterializedValue[FlowShape[WriteMessage, CommittableReadResult], Future[String]] {
stage =>
@@ -53,7 +54,7 @@ private[amqp] final class AmqpRpcFlowStage(settings: AmqpWriteSettings, bufferSi
val streamCompletion = Promise[String]()
(new GraphStageLogic(shape) with AmqpConnectorLogic {
- override val settings = stage.settings
+ override val settings: AmqpWriteSettings = stage.writeSettings
private val exchange = settings.exchange.getOrElse("")
private val routingKey = settings.routingKey.getOrElse("")
private val queue = mutable.Queue[CommittableReadResult]()
diff --git a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpSimpleFlowStage.scala b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpSimpleFlowStage.scala
index 3f87fb685..efb5888f4 100644
--- a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpSimpleFlowStage.scala
+++ b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/impl/AmqpSimpleFlowStage.scala
@@ -32,7 +32,7 @@ import scala.concurrent.{ Future, Promise }
* instead of complete [[WriteResult]] (possibly it would be less confusing for users), but [[WriteResult]] is used
* for consistency with other variants and to make the flow ready for any possible future [[WriteResult]] extensions.
*/
-@InternalApi private[amqp] final class AmqpSimpleFlowStage[T](settings: AmqpWriteSettings)
+@InternalApi private[amqp] final class AmqpSimpleFlowStage[T](writeSettings: AmqpWriteSettings)
extends GraphStageWithMaterializedValue[FlowShape[(WriteMessage, T), (WriteResult, T)], Future[Done]] { stage =>
private val in: Inlet[(WriteMessage, T)] = Inlet(Logging.simpleName(this) + ".in")
@@ -45,7 +45,7 @@ import scala.concurrent.{ Future, Promise }
override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Future[Done]) = {
val streamCompletion = Promise[Done]()
- (new AbstractAmqpFlowStageLogic[T](settings, streamCompletion, shape) {
+ (new AbstractAmqpFlowStageLogic[T](writeSettings, streamCompletion, shape) {
override def publish(message: WriteMessage, passThrough: T): Unit = {
log.debug("Publishing message {}.", message)
diff --git a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpSource.scala b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpSource.scala
index 9266fd9f3..3b8d8ad73 100644
--- a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpSource.scala
+++ b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpSource.scala
@@ -20,8 +20,10 @@ import pekko.stream.connectors.amqp.impl
import pekko.stream.connectors.amqp.{ AmqpSourceSettings, ReadResult }
import pekko.stream.scaladsl.Source
+import scala.concurrent.ExecutionContext
+
object AmqpSource {
- private implicit val executionContext = ExecutionContexts.parasitic
+ private implicit val executionContext: ExecutionContext = ExecutionContexts.parasitic
/**
* Scala API: Convenience for "at-most once delivery" semantics. Each message is acked to RabbitMQ
diff --git a/amqp/src/test/scala/docs/scaladsl/AmqpDocsSpec.scala b/amqp/src/test/scala/docs/scaladsl/AmqpDocsSpec.scala
index a2dc099f7..8199d8f8b 100644
--- a/amqp/src/test/scala/docs/scaladsl/AmqpDocsSpec.scala
+++ b/amqp/src/test/scala/docs/scaladsl/AmqpDocsSpec.scala
@@ -33,7 +33,7 @@ import scala.collection.immutable
*/
class AmqpDocsSpec extends AmqpSpec {
- override implicit val patienceConfig = PatienceConfig(10.seconds)
+ override implicit val patienceConfig: PatienceConfig = PatienceConfig(10.seconds)
val businessLogic: CommittableReadResult => Future[CommittableReadResult] = Future.successful(_)
diff --git a/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/AmqpSpec.scala b/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/AmqpSpec.scala
index 63068932b..87d29f1d2 100644
--- a/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/AmqpSpec.scala
+++ b/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/AmqpSpec.scala
@@ -22,10 +22,12 @@ import org.scalatest.concurrent.ScalaFutures
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec
+import scala.concurrent.ExecutionContext
+
abstract class AmqpSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with ScalaFutures with LogCapturing {
- implicit val system = ActorSystem(this.getClass.getSimpleName)
- implicit val executionContext = ExecutionContexts.parasitic
+ implicit val system: ActorSystem = ActorSystem(this.getClass.getSimpleName)
+ implicit val executionContext: ExecutionContext = ExecutionContexts.parasitic
override protected def afterAll(): Unit =
system.terminate()
diff --git a/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpGraphStageLogicConnectionShutdownSpec.scala b/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpGraphStageLogicConnectionShutdownSpec.scala
index ed8e8a4bb..0f58a2513 100644
--- a/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpGraphStageLogicConnectionShutdownSpec.scala
+++ b/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpGraphStageLogicConnectionShutdownSpec.scala
@@ -15,7 +15,6 @@ package org.apache.pekko.stream.connectors.amqp.scaladsl
import java.util.concurrent.ExecutorService
import java.util.concurrent.atomic.AtomicInteger
-
import org.apache.pekko
import pekko.actor.ActorSystem
import pekko.dispatch.ExecutionContexts
@@ -33,7 +32,7 @@ import com.rabbitmq.client.{ AddressResolver, Connection, ConnectionFactory, Shu
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.BeforeAndAfterEach
-import scala.concurrent.Future
+import scala.concurrent.{ ExecutionContext, Future }
import scala.concurrent.duration._
import scala.util.control.NonFatal
import org.scalatest.matchers.should.Matchers
@@ -50,8 +49,8 @@ class AmqpGraphStageLogicConnectionShutdownSpec
with BeforeAndAfterEach
with LogCapturing {
- override implicit val patienceConfig = PatienceConfig(10.seconds)
- private implicit val executionContext = ExecutionContexts.parasitic
+ override implicit val patienceConfig: PatienceConfig = PatienceConfig(10.seconds)
+ private implicit val executionContext: ExecutionContext = ExecutionContexts.parasitic
val shutdownsAdded = new AtomicInteger()
val shutdownsRemoved = new AtomicInteger()
@@ -76,7 +75,7 @@ class AmqpGraphStageLogicConnectionShutdownSpec
"registers and unregisters a single connection shutdown hook per graph" in {
// actor system is within this test as it has to be shut down in order
// to verify graph stage termination
- implicit val system = ActorSystem(this.getClass.getSimpleName + System.currentTimeMillis())
+ implicit val system: ActorSystem = ActorSystem(this.getClass.getSimpleName + System.currentTimeMillis())
val connectionFactory = new ConnectionFactory() {
override def newConnection(es: ExecutorService, ar: AddressResolver, name: String): Connection =
diff --git a/aws-event-bridge/src/test/scala/docs/scaladsl/EventBridgePublisherSpec.scala b/aws-event-bridge/src/test/scala/docs/scaladsl/EventBridgePublisherSpec.scala
index 9923f262e..382e5e300 100644
--- a/aws-event-bridge/src/test/scala/docs/scaladsl/EventBridgePublisherSpec.scala
+++ b/aws-event-bridge/src/test/scala/docs/scaladsl/EventBridgePublisherSpec.scala
@@ -28,7 +28,7 @@ import scala.concurrent.duration._
class EventBridgePublisherSpec extends AnyFlatSpec with Matchers with ScalaFutures with IntegrationTestContext {
- implicit val defaultPatience =
+ implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = 15.seconds, interval = 100.millis)
"EventBridge Publisher sink" should "send PutEventsEntry message" in {
diff --git a/awslambda/src/test/scala/docs/scaladsl/AwsLambdaFlowSpec.scala b/awslambda/src/test/scala/docs/scaladsl/AwsLambdaFlowSpec.scala
index 0affabe3a..163e1471e 100644
--- a/awslambda/src/test/scala/docs/scaladsl/AwsLambdaFlowSpec.scala
+++ b/awslambda/src/test/scala/docs/scaladsl/AwsLambdaFlowSpec.scala
@@ -14,7 +14,6 @@
package docs.scaladsl
import java.util.concurrent.CompletableFuture
-
import org.apache.pekko
import pekko.actor.ActorSystem
import pekko.stream.connectors.awslambda.scaladsl.AwsLambdaFlow
@@ -36,7 +35,7 @@ import software.amazon.awssdk.core.SdkBytes
import software.amazon.awssdk.services.lambda.LambdaAsyncClient
import software.amazon.awssdk.services.lambda.model.{ InvokeRequest, InvokeResponse }
-import scala.concurrent.Await
+import scala.concurrent.{ Await, ExecutionContext }
import scala.concurrent.duration._
class AwsLambdaFlowSpec
@@ -49,9 +48,9 @@ class AwsLambdaFlowSpec
with MockitoSugar
with LogCapturing {
- implicit val ec = system.dispatcher
+ implicit val ec: ExecutionContext = system.dispatcher
- implicit val awsLambdaClient = mock[LambdaAsyncClient]
+ implicit val awsLambdaClient: LambdaAsyncClient = mock[LambdaAsyncClient]
override protected def afterEach(): Unit = {
reset(awsLambdaClient)
diff --git a/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSink.scala b/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSink.scala
index 36488a3e7..ea8c9adcd 100644
--- a/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSink.scala
+++ b/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSink.scala
@@ -51,7 +51,7 @@ object AzureQueueWithTimeoutsSink {
* of a [[com.microsoft.azure.storage.queue.CloudQueueMessage]] a [[MessageWithTimeouts]].
*/
def create(cloudQueue: Supplier[CloudQueue]): Sink[MessageWithTimeouts, CompletionStage[Done]] =
- AzureQueueSink.fromFunction { input: MessageWithTimeouts =>
+ AzureQueueSink.fromFunction[MessageWithTimeouts] { input =>
AzureQueueSinkFunctions
.addMessage(() => cloudQueue.get)(input.message, input.timeToLive, input.initialVisibility)
}
diff --git a/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/javadsl/CassandraSession.scala b/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/javadsl/CassandraSession.scala
index 9fa9f8afc..10459641b 100644
--- a/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/javadsl/CassandraSession.scala
+++ b/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/javadsl/CassandraSession.scala
@@ -83,7 +83,7 @@ final class CassandraSession(@InternalApi private[pekko] val delegate: scaladsl.
onClose: java.lang.Runnable) =
this(system.classicSystem, sessionProvider, executionContext, log, metricsCategory, init, onClose)
- implicit private val ec = delegate.ec
+ implicit private val ec: ExecutionContext = delegate.ec
/**
* Closes the underlying Cassandra session.
diff --git a/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraLifecycle.scala b/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraLifecycle.scala
index 00674384a..556c4c8e5 100644
--- a/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraLifecycle.scala
+++ b/cassandra/src/test/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraLifecycle.scala
@@ -15,7 +15,6 @@ package org.apache.pekko.stream.connectors.cassandra.scaladsl
import java.util.concurrent.CompletionStage
import java.util.concurrent.atomic.AtomicInteger
-
import org.apache.pekko
import pekko.Done
import pekko.testkit.TestKitBase
@@ -27,7 +26,7 @@ import org.scalatest.concurrent.{ PatienceConfiguration, ScalaFutures }
import scala.collection.immutable
import scala.concurrent.duration._
-import scala.concurrent.{ Await, Future }
+import scala.concurrent.{ Await, ExecutionContext, Future }
import scala.util.control.NonFatal
trait CassandraLifecycleBase {
@@ -71,7 +70,7 @@ trait CassandraLifecycleBase {
executeCql(lifecycleSession, statements.asScala.toList).asJava
def withSchemaMetadataDisabled(block: => Future[Done]): Future[Done] = {
- implicit val ec = lifecycleSession.ec
+ implicit val ec: ExecutionContext = lifecycleSession.ec
lifecycleSession.underlying().flatMap { cqlSession =>
cqlSession.setSchemaMetadataEnabled(false)
val blockResult =
diff --git a/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/CouchbaseFlow.scala b/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/CouchbaseFlow.scala
index d8089d208..79039403d 100644
--- a/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/CouchbaseFlow.scala
+++ b/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/CouchbaseFlow.scala
@@ -29,6 +29,8 @@ import pekko.stream.connectors.couchbase.{
import pekko.stream.scaladsl.Flow
import com.couchbase.client.java.document.{ Document, JsonDocument }
+import scala.concurrent.ExecutionContext
+
/**
* Scala API: Factory methods for Couchbase flows.
*/
@@ -104,7 +106,7 @@ object CouchbaseFlow {
val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
Flow[T]
.mapAsync(writeSettings.parallelism)(doc => {
- implicit val executor = materializer.system.dispatcher
+ implicit val executor: ExecutionContext = materializer.system.dispatcher
session
.flatMap(_.upsertDoc(doc, writeSettings))
.map(_ => CouchbaseWriteSuccess(doc))
@@ -157,7 +159,7 @@ object CouchbaseFlow {
val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
Flow[T]
.mapAsync(writeSettings.parallelism)(doc => {
- implicit val executor = materializer.system.dispatcher
+ implicit val executor: ExecutionContext = materializer.system.dispatcher
session
.flatMap(_.replaceDoc(doc, writeSettings))
.map(_ => CouchbaseWriteSuccess(doc))
@@ -179,7 +181,7 @@ object CouchbaseFlow {
val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
Flow[String]
.mapAsync(writeSettings.parallelism)(id => {
- implicit val executor = materializer.system.dispatcher
+ implicit val executor: ExecutionContext = materializer.system.dispatcher
session
.flatMap(_.remove(id, writeSettings))
.map(_ => id)
@@ -198,7 +200,7 @@ object CouchbaseFlow {
val session = CouchbaseSessionRegistry(materializer.system).sessionFor(sessionSettings, bucketName)
Flow[String]
.mapAsync(writeSettings.parallelism)(id => {
- implicit val executor = materializer.system.dispatcher
+ implicit val executor: ExecutionContext = materializer.system.dispatcher
session
.flatMap(_.remove(id, writeSettings))
.map(_ => CouchbaseDeleteSuccess(id))
diff --git a/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/DiscoverySupport.scala b/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/DiscoverySupport.scala
index dd7184e71..83979974f 100644
--- a/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/DiscoverySupport.scala
+++ b/couchbase/src/main/scala/org/apache/pekko/stream/connectors/couchbase/scaladsl/DiscoverySupport.scala
@@ -25,7 +25,7 @@ import pekko.util.FutureConverters._
import com.typesafe.config.Config
import scala.collection.immutable
-import scala.concurrent.Future
+import scala.concurrent.{ ExecutionContext, Future }
import scala.concurrent.duration.FiniteDuration
/**
@@ -39,7 +39,7 @@ sealed class DiscoverySupport private {
private def readNodes(
serviceName: String,
lookupTimeout: FiniteDuration)(implicit system: ClassicActorSystemProvider): Future[immutable.Seq[String]] = {
- implicit val ec = system.classicSystem.dispatcher
+ implicit val ec: ExecutionContext = system.classicSystem.dispatcher
val discovery = Discovery(system).discovery
discovery.lookup(serviceName, lookupTimeout).map { resolved =>
resolved.addresses.map(_.host)
@@ -63,7 +63,7 @@ sealed class DiscoverySupport private {
def nodes(
config: Config)(
implicit system: ClassicActorSystemProvider): CouchbaseSessionSettings => Future[CouchbaseSessionSettings] = {
- implicit val ec = system.classicSystem.dispatcher
+ implicit val ec: ExecutionContext = system.classicSystem.dispatcher
settings =>
readNodes(config)
.map { nodes =>
diff --git a/csv-bench/src/main/scala/org/apache/pekko/stream/connectors/csv/scaladsl/CsvBench.scala b/csv-bench/src/main/scala/org/apache/pekko/stream/connectors/csv/scaladsl/CsvBench.scala
index b661296ee..25bcc4fb9 100644
--- a/csv-bench/src/main/scala/org/apache/pekko/stream/connectors/csv/scaladsl/CsvBench.scala
+++ b/csv-bench/src/main/scala/org/apache/pekko/stream/connectors/csv/scaladsl/CsvBench.scala
@@ -5,7 +5,6 @@
package org.apache.pekko.stream.connectors.csv.scaladsl
import java.util.concurrent.TimeUnit
-
import org.apache.pekko
import pekko.NotUsed
import pekko.actor.ActorSystem
@@ -15,7 +14,7 @@ import pekko.util.ByteString
import org.openjdk.jmh.annotations._
import org.openjdk.jmh.infra.Blackhole
-import scala.concurrent.Await
+import scala.concurrent.{ Await, ExecutionContext }
import scala.concurrent.duration.Duration
/**
@@ -50,9 +49,9 @@ import scala.concurrent.duration.Duration
@State(Scope.Benchmark)
class CsvBench {
- implicit val system = ActorSystem()
- implicit val executionContext = system.dispatcher
- implicit val mat = ActorMaterializer()
+ implicit val system: ActorSystem = ActorSystem()
+ implicit val executionContext: ExecutionContext = system.dispatcher
+ implicit val mat: ActorMaterializer = ActorMaterializer()
/**
* Size of [[ByteString]] chunks in bytes.
diff --git a/csv/src/test/scala/docs/scaladsl/CsvSpec.scala b/csv/src/test/scala/docs/scaladsl/CsvSpec.scala
index 36c8bc911..57a52ee7c 100644
--- a/csv/src/test/scala/docs/scaladsl/CsvSpec.scala
+++ b/csv/src/test/scala/docs/scaladsl/CsvSpec.scala
@@ -30,7 +30,7 @@ abstract class CsvSpec
with ScalaFutures
with LogCapturing {
- implicit val system = ActorSystem(this.getClass.getSimpleName)
+ implicit val system: ActorSystem = ActorSystem(this.getClass.getSimpleName)
override protected def afterAll(): Unit =
TestKit.shutdownActorSystem(system)
diff --git a/doc-examples/src/test/scala/org/apache/pekko/stream/connectors/eip/scaladsl/PassThroughExamples.scala b/doc-examples/src/test/scala/org/apache/pekko/stream/connectors/eip/scaladsl/PassThroughExamples.scala
index 5eaf3cf98..a411fcc47 100644
--- a/doc-examples/src/test/scala/org/apache/pekko/stream/connectors/eip/scaladsl/PassThroughExamples.scala
+++ b/doc-examples/src/test/scala/org/apache/pekko/stream/connectors/eip/scaladsl/PassThroughExamples.scala
@@ -29,7 +29,7 @@ import org.scalatest.wordspec.AnyWordSpec
class PassThroughExamples extends AnyWordSpec with BeforeAndAfterAll with Matchers with ScalaFutures {
- implicit val system = ActorSystem("Test")
+ implicit val system: ActorSystem = ActorSystem("Test")
"PassThroughFlow" should {
" original message is maintained " in {
@@ -103,7 +103,7 @@ object PassThroughFlow {
//#PassThrough
object PassThroughFlowKafkaCommitExample {
- implicit val system = ActorSystem("Test")
+ implicit val system: ActorSystem = ActorSystem("Test")
def dummy(): Unit = {
// #passThroughKafkaFlow
diff --git a/dynamodb/src/test/scala/docs/scaladsl/ExampleSpec.scala b/dynamodb/src/test/scala/docs/scaladsl/ExampleSpec.scala
index 1f1ff9a31..1ec8ca9cf 100644
--- a/dynamodb/src/test/scala/docs/scaladsl/ExampleSpec.scala
+++ b/dynamodb/src/test/scala/docs/scaladsl/ExampleSpec.scala
@@ -17,15 +17,14 @@ import java.net.URI
import org.apache.pekko
import pekko.NotUsed
+import pekko.actor.ActorSystem
import pekko.stream.connectors.testkit.scaladsl.LogCapturing
import pekko.stream.scaladsl.{ FlowWithContext, SourceWithContext }
import scala.util.{ Failure, Success, Try }
//#init-client
-import org.apache.pekko.actor.ActorSystem
//#init-client
-import org.apache.pekko
import pekko.stream.connectors.dynamodb.DynamoDbOp._
import pekko.stream.connectors.dynamodb.scaladsl._
import pekko.stream.scaladsl.{ Sink, Source }
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala
index 925632a53..3f0f98977 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala
@@ -364,7 +364,7 @@ trait ElasticsearchConnectorBehaviour {
val indexName = "sink7"
val createBooks = Source(books)
- .map { book: (String, Book) =>
+ .map { (book: (String, Book)) =>
WriteMessage.createUpsertMessage(id = book._1, source = book._2)
}
.via(
@@ -391,7 +391,7 @@ trait ElasticsearchConnectorBehaviour {
// Update sink7/_doc with the second dataset
val upserts = Source(updatedBooks)
- .map { book: (String, JsObject) =>
+ .map { (book: (String, JsObject)) =>
WriteMessage.createUpsertMessage(id = book._1, source = book._2)
}
.via(
@@ -453,7 +453,7 @@ trait ElasticsearchConnectorBehaviour {
"read and write document-version if configured to do so" in {
case class VersionTestDoc(id: String, name: String, value: Int)
- implicit val formatVersionTestDoc: JsonFormat[VersionTestDoc] = jsonFormat3(VersionTestDoc)
+ implicit val formatVersionTestDoc: JsonFormat[VersionTestDoc] = jsonFormat3(VersionTestDoc.apply)
val indexName = "version-test-scala"
val typeName = "_doc"
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchSpecUtils.scala b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchSpecUtils.scala
index f60e4caab..03f83f61b 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchSpecUtils.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchSpecUtils.scala
@@ -45,7 +45,7 @@ trait ElasticsearchSpecUtils { this: AnyWordSpec with ScalaFutures =>
case class Book(title: String, shouldSkip: Option[Boolean] = None, price: Int = 10)
- implicit val format: JsonFormat[Book] = jsonFormat3(Book)
+ implicit val format: JsonFormat[Book] = jsonFormat3(Book.apply)
// #define-class
def register(connectionSettings: ElasticsearchConnectionSettings,
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala
index eb4e64a2b..f8293cad9 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala
@@ -64,7 +64,7 @@ class ElasticsearchV5Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
constructElasticsearchParams("source", "_doc", ApiVersion.V5),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[spray.json.JsObject] =>
+ .map { (message: ReadResult[spray.json.JsObject]) =>
val book: Book = jsonReader[Book].read(message.source)
WriteMessage.createIndexMessage(message.id, book)
}
@@ -97,7 +97,7 @@ class ElasticsearchV5Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
constructElasticsearchParams("source", "_doc", ApiVersion.V5),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[Book] =>
+ .map { (message: ReadResult[Book]) =>
WriteMessage.createIndexMessage(message.id, message.source)
}
.runWith(
@@ -129,7 +129,7 @@ class ElasticsearchV5Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
constructElasticsearchParams("source", "_doc", ApiVersion.V5),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[Book] =>
+ .map { (message: ReadResult[Book]) =>
WriteMessage.createIndexMessage(message.id, message.source)
}
.via(
@@ -200,7 +200,7 @@ class ElasticsearchV5Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
val indexName = "sink6"
val kafkaToEs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -251,7 +251,7 @@ class ElasticsearchV5Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
val indexName = "sink6-bulk"
val kafkaToEs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -306,7 +306,7 @@ class ElasticsearchV5Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
val indexName = "sink6-nop"
val kafkaToEs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -363,7 +363,7 @@ class ElasticsearchV5Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
register(connectionSettings, indexName, "dummy", 10) // need to create index else exception in reading below
val kafkaToEs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -447,7 +447,7 @@ class ElasticsearchV5Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
constructElasticsearchParams("source", "_doc", ApiVersion.V5),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[Book] =>
+ .map { (message: ReadResult[Book]) =>
WriteMessage
.createIndexMessage(message.id, message.source)
.withIndexName(customIndexName) // Setting the index-name to use for this document
@@ -480,7 +480,7 @@ class ElasticsearchV5Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
case class TestDoc(id: String, a: String, b: Option[String], c: String)
// #custom-search-params
- implicit val formatVersionTestDoc: JsonFormat[TestDoc] = jsonFormat4(TestDoc)
+ implicit val formatVersionTestDoc: JsonFormat[TestDoc] = jsonFormat4(TestDoc.apply)
val indexName = "custom-search-params-test-scala"
val typeName = "_doc"
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala
index 4d0650490..5eac2e95c 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala
@@ -56,7 +56,7 @@ class ElasticsearchV7Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
constructElasticsearchParams("source", "_doc", ApiVersion.V7),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[spray.json.JsObject] =>
+ .map { (message: ReadResult[spray.json.JsObject]) =>
val book: Book = jsonReader[Book].read(message.source)
WriteMessage.createIndexMessage(message.id, book)
}
@@ -88,7 +88,7 @@ class ElasticsearchV7Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
constructElasticsearchParams("source", "_doc", ApiVersion.V7),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[Book] =>
+ .map { (message: ReadResult[Book]) =>
WriteMessage.createIndexMessage(message.id, message.source)
}
.runWith(
@@ -119,7 +119,7 @@ class ElasticsearchV7Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
constructElasticsearchParams("source", "_doc", ApiVersion.V7),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[Book] =>
+ .map { (message: ReadResult[Book]) =>
WriteMessage.createIndexMessage(message.id, message.source)
}
.via(
@@ -187,7 +187,7 @@ class ElasticsearchV7Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
val indexName = "sink6"
val kafkaToEs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -237,7 +237,7 @@ class ElasticsearchV7Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
val indexName = "sink6-bulk"
val kafkaToEs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -291,7 +291,7 @@ class ElasticsearchV7Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
val indexName = "sink6-nop"
val kafkaToEs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -347,7 +347,7 @@ class ElasticsearchV7Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
register(connectionSettings, indexName, "dummy", 10) // need to create index else exception in reading below
val kafkaToEs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -428,7 +428,7 @@ class ElasticsearchV7Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
constructElasticsearchParams("source", "_doc", ApiVersion.V7),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[Book] =>
+ .map { (message: ReadResult[Book]) =>
WriteMessage
.createIndexMessage(message.id, message.source)
.withIndexName(customIndexName) // Setting the index-name to use for this document
@@ -458,7 +458,7 @@ class ElasticsearchV7Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
case class TestDoc(id: String, a: String, b: Option[String], c: String)
- implicit val formatVersionTestDoc: JsonFormat[TestDoc] = jsonFormat4(TestDoc)
+ implicit val formatVersionTestDoc: JsonFormat[TestDoc] = jsonFormat4(TestDoc.apply)
val indexName = "custom-search-params-test-scala"
val typeName = "_doc"
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchConnectorBehaviour.scala b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchConnectorBehaviour.scala
index 09e6073f6..563f9a5dd 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchConnectorBehaviour.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchConnectorBehaviour.scala
@@ -364,7 +364,7 @@ trait OpensearchConnectorBehaviour {
val indexName = "sink7"
val createBooks = Source(books)
- .map { book: (String, Book) =>
+ .map { (book: (String, Book)) =>
WriteMessage.createUpsertMessage(id = book._1, source = book._2)
}
.via(
@@ -391,7 +391,7 @@ trait OpensearchConnectorBehaviour {
// Update sink7/_doc with the second dataset
val upserts = Source(updatedBooks)
- .map { book: (String, JsObject) =>
+ .map { (book: (String, JsObject)) =>
WriteMessage.createUpsertMessage(id = book._1, source = book._2)
}
.via(
@@ -453,7 +453,7 @@ trait OpensearchConnectorBehaviour {
"read and write document-version if configured to do so" in {
case class VersionTestDoc(id: String, name: String, value: Int)
- implicit val formatVersionTestDoc: JsonFormat[VersionTestDoc] = jsonFormat3(VersionTestDoc)
+ implicit val formatVersionTestDoc: JsonFormat[VersionTestDoc] = jsonFormat3(VersionTestDoc.apply)
val indexName = "version-test-scala"
val typeName = "_doc"
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
index be2c37ae2..7a486fd42 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
@@ -65,7 +65,7 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with ElasticsearchSpecUtils
constructElasticsearchParams("source", "_doc", OpensearchApiVersion.V1),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[spray.json.JsObject] =>
+ .map { (message: ReadResult[spray.json.JsObject]) =>
val book: Book = jsonReader[Book].read(message.source)
WriteMessage.createIndexMessage(message.id, book)
}
@@ -100,7 +100,7 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with ElasticsearchSpecUtils
constructElasticsearchParams("source", "_doc", OpensearchApiVersion.V1),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[Book] =>
+ .map { (message: ReadResult[Book]) =>
WriteMessage.createIndexMessage(message.id, message.source)
}
.runWith(
@@ -133,7 +133,7 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with ElasticsearchSpecUtils
constructElasticsearchParams("source", "_doc", OpensearchApiVersion.V1),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[Book] =>
+ .map { (message: ReadResult[Book]) =>
WriteMessage.createIndexMessage(message.id, message.source)
}
.via(
@@ -205,7 +205,7 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with ElasticsearchSpecUtils
val indexName = "sink6"
val kafkaToOs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -255,7 +255,7 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with ElasticsearchSpecUtils
val indexName = "sink6-bulk"
val kafkaToOs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -309,7 +309,7 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with ElasticsearchSpecUtils
val indexName = "sink6-nop"
val kafkaToOs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -365,7 +365,7 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with ElasticsearchSpecUtils
register(connectionSettings, indexName, "dummy", 10) // need to create index else exception in reading below
val kafkaToOs = Source(messagesFromKafka) // Assume we get this from Kafka
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
@@ -449,7 +449,7 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with ElasticsearchSpecUtils
constructElasticsearchParams("source", "_doc", OpensearchApiVersion.V1),
query = """{"match_all": {}}""",
settings = baseSourceSettings)
- .map { message: ReadResult[Book] =>
+ .map { (message: ReadResult[Book]) =>
WriteMessage
.createIndexMessage(message.id, message.source)
.withIndexName(customIndexName) // Setting the index-name to use for this document
@@ -480,7 +480,7 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with ElasticsearchSpecUtils
case class TestDoc(id: String, a: String, b: Option[String], c: String)
- implicit val formatVersionTestDoc: JsonFormat[TestDoc] = jsonFormat4(TestDoc)
+ implicit val formatVersionTestDoc: JsonFormat[TestDoc] = jsonFormat4(TestDoc.apply)
val indexName = "custom-search-params-test-scala"
val typeName = "_doc"
diff --git a/file/src/main/scala/org/apache/pekko/stream/connectors/file/javadsl/LogRotatorSink.scala b/file/src/main/scala/org/apache/pekko/stream/connectors/file/javadsl/LogRotatorSink.scala
index d439c58d7..56f504edb 100644
--- a/file/src/main/scala/org/apache/pekko/stream/connectors/file/javadsl/LogRotatorSink.scala
+++ b/file/src/main/scala/org/apache/pekko/stream/connectors/file/javadsl/LogRotatorSink.scala
@@ -16,12 +16,10 @@ package org.apache.pekko.stream.connectors.file.javadsl
import java.nio.file.{ Path, StandardOpenOption }
import java.util.Optional
import java.util.concurrent.CompletionStage
-
import org.apache.pekko
import pekko.Done
import pekko.stream.javadsl
import pekko.stream.scaladsl
-import pekko.stream.javadsl.Sink
import pekko.util.ByteString
import pekko.japi.function
import pekko.util.ccompat.JavaConverters._
@@ -42,11 +40,11 @@ object LogRotatorSink {
*/
def createFromFunction(
triggerGeneratorCreator: function.Creator[function.Function[ByteString, Optional[Path]]])
- : javadsl.Sink[ByteString, CompletionStage[Done]] =
- new Sink(
- pekko.stream.connectors.file.scaladsl
- .LogRotatorSink(asScala(triggerGeneratorCreator))
- .toCompletionStage())
+ : javadsl.Sink[ByteString, CompletionStage[Done]] = {
+ val logRotatorSink = new scaladsl.SinkToCompletionStage[ByteString, Done](pekko.stream.connectors.file.scaladsl
+ .LogRotatorSink(asScala(triggerGeneratorCreator)))
+ new javadsl.Sink(logRotatorSink.toCompletionStage())
+ }
/**
* Sink directing the incoming `ByteString`s to new files whenever `triggerGenerator` returns a value.
@@ -56,11 +54,12 @@ object LogRotatorSink {
*/
def createFromFunctionAndOptions(
triggerGeneratorCreator: function.Creator[function.Function[ByteString, Optional[Path]]],
- fileOpenOptions: java.util.Set[StandardOpenOption]): javadsl.Sink[ByteString, CompletionStage[Done]] =
- new Sink(
- pekko.stream.connectors.file.scaladsl
- .LogRotatorSink(asScala(triggerGeneratorCreator), fileOpenOptions.asScala.toSet)
- .toCompletionStage())
+ fileOpenOptions: java.util.Set[StandardOpenOption]): javadsl.Sink[ByteString, CompletionStage[Done]] = {
+
+ val logRotatorSink = new scaladsl.SinkToCompletionStage[ByteString, Done](pekko.stream.connectors.file.scaladsl
+ .LogRotatorSink(asScala(triggerGeneratorCreator), fileOpenOptions.asScala.toSet))
+ new javadsl.Sink(logRotatorSink.toCompletionStage())
+ }
/**
* Sink directing the incoming `ByteString`s to a new `Sink` created by `sinkFactory` whenever `triggerGenerator` returns a value.
@@ -72,14 +71,14 @@ object LogRotatorSink {
*/
def withSinkFactory[C, R](
triggerGeneratorCreator: function.Creator[function.Function[ByteString, Optional[C]]],
- sinkFactory: function.Function[C, Sink[ByteString, CompletionStage[R]]])
+ sinkFactory: function.Function[C, javadsl.Sink[ByteString, CompletionStage[R]]])
: javadsl.Sink[ByteString, CompletionStage[Done]] = {
val t: C => scaladsl.Sink[ByteString, Future[R]] = path =>
sinkFactory.apply(path).asScala.mapMaterializedValue(_.asScala)
- new Sink(
- pekko.stream.connectors.file.scaladsl.LogRotatorSink
- .withSinkFactory(asScala[C](triggerGeneratorCreator), t)
- .toCompletionStage())
+ val logRotatorSink =
+ new scaladsl.SinkToCompletionStage[ByteString, Done](pekko.stream.connectors.file.scaladsl.LogRotatorSink
+ .withSinkFactory(asScala[C](triggerGeneratorCreator), t))
+ new javadsl.Sink(logRotatorSink.toCompletionStage())
}
private def asScala[C](
diff --git a/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/LogRotatorSink.scala b/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/LogRotatorSink.scala
index 1fe7f4f59..fca14c150 100644
--- a/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/LogRotatorSink.scala
+++ b/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/LogRotatorSink.scala
@@ -167,7 +167,7 @@ final private class LogRotatorSink[T, C, R](triggerGeneratorCreator: () => T =>
override def postStop(): Unit =
promise.completeWith {
- implicit val ec = materializer.executionContext
+ implicit val ec: ExecutionContext = materializer.executionContext
Future
.sequence(sinkCompletions)
.map(_ => Done)(pekko.dispatch.ExecutionContexts.parasitic)
diff --git a/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala b/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala
index df2e48092..13d32e893 100644
--- a/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala
+++ b/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala
@@ -95,7 +95,7 @@ class LogRotatorSinkSpec
"complete when consuming an empty source" in assertAllStagesStopped {
val triggerCreator: () => ByteString => Option[Path] = () => {
- element: ByteString => fail("trigger creator should not be called")
+ (element: ByteString) => fail("trigger creator should not be called")
}
val rotatorSink: Sink[ByteString, Future[Done]] =
@@ -112,7 +112,7 @@ class LogRotatorSinkSpec
val fileSizeTriggerCreator: () => ByteString => Option[Path] = () => {
val max = 10 * 1024 * 1024
var size: Long = max
- element: ByteString =>
+ (element: ByteString) =>
if (size + element.size > max) {
val path = Files.createTempFile("out-", ".log")
size = element.size
diff --git a/ftp/src/test/scala/org/apache/pekko/stream/connectors/ftp/CommonFtpStageSpec.scala b/ftp/src/test/scala/org/apache/pekko/stream/connectors/ftp/CommonFtpStageSpec.scala
index 8c3b24cdb..7020f08de 100644
--- a/ftp/src/test/scala/org/apache/pekko/stream/connectors/ftp/CommonFtpStageSpec.scala
+++ b/ftp/src/test/scala/org/apache/pekko/stream/connectors/ftp/CommonFtpStageSpec.scala
@@ -20,8 +20,9 @@ import java.nio.file.{ Files, Paths }
import java.time.Instant
import java.util.concurrent.TimeUnit
import org.apache.pekko
-import pekko.stream.{ IOOperationIncompleteException, IOResult }
+import pekko.stream.{ IOOperationIncompleteException, IOResult, Materializer }
import BaseSftpSupport.{ CLIENT_PRIVATE_KEY_PASSPHRASE => ClientPrivateKeyPassphrase }
+import pekko.actor.ActorSystem
import pekko.stream.scaladsl.{ Keep, Sink, Source }
import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import pekko.stream.testkit.scaladsl.TestSink
@@ -30,7 +31,7 @@ import org.scalatest.concurrent.Eventually
import org.scalatest.time.{ Millis, Seconds, Span }
import scala.collection.immutable
-import scala.concurrent.{ Await, ExecutionContextExecutor }
+import scala.concurrent.{ Await, ExecutionContext, ExecutionContextExecutor }
import scala.concurrent.duration._
import scala.util.Random
@@ -85,14 +86,14 @@ final class UnconfirmedReadsSftpSourceSpec extends BaseSftpSpec with CommonFtpSt
trait CommonFtpStageSpec extends BaseSpec with Eventually {
- implicit val system = getSystem
- implicit val mat = getMaterializer
- implicit val defaultPatience =
+ implicit val system: ActorSystem = getSystem
+ implicit val mat: Materializer = getMaterializer
+ implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = Span(30, Seconds), interval = Span(600, Millis))
"FtpBrowserSource" should {
"complete with a failed Future, when the credentials supplied were wrong" in assertAllStagesStopped {
- implicit val ec = system.getDispatcher
+ implicit val ec: ExecutionContext = system.getDispatcher
listFilesWithWrongCredentials("")
.toMat(Sink.seq)(Keep.right)
.run()
diff --git a/geode/src/test/scala/docs/scaladsl/GeodeBaseSpec.scala b/geode/src/test/scala/docs/scaladsl/GeodeBaseSpec.scala
index c4d922720..c1369a083 100644
--- a/geode/src/test/scala/docs/scaladsl/GeodeBaseSpec.scala
+++ b/geode/src/test/scala/docs/scaladsl/GeodeBaseSpec.scala
@@ -30,7 +30,7 @@ import org.scalatest.wordspec.AnyWordSpec
class GeodeBaseSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with LogCapturing {
- implicit val system = ActorSystem("test")
+ implicit val system: ActorSystem = ActorSystem("test")
// #region
val personsRegionSettings: RegionSettings[Int, Person] = RegionSettings("persons", (p: Person) => p.id)
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/BigQueryException.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/BigQueryException.scala
index 717cd57ae..eb9e7e910 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/BigQueryException.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/BigQueryException.scala
@@ -77,5 +77,5 @@ object BigQueryException {
}
private final case class ErrorResponse(error: Option[ErrorProto])
- private implicit val errorResponseFormat: RootJsonFormat[ErrorResponse] = jsonFormat1(ErrorResponse)
+ private implicit val errorResponseFormat: RootJsonFormat[ErrorResponse] = jsonFormat1(ErrorResponse.apply)
}
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableDataJsonProtocol.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableDataJsonProtocol.scala
index b2b0107e4..c370e7abf 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableDataJsonProtocol.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/model/TableDataJsonProtocol.scala
@@ -24,7 +24,6 @@ import com.fasterxml.jackson.annotation._
import spray.json.{ JsonFormat, RootJsonFormat, RootJsonReader, RootJsonWriter }
import java.{ lang, util }
-
import scala.annotation.nowarn
import scala.annotation.unchecked.uncheckedVariance
import scala.collection.immutable.Seq
@@ -166,8 +165,8 @@ object TableDataInsertAllRequest {
implicit def writer[T](
implicit writer: BigQueryRootJsonWriter[T]): RootJsonWriter[TableDataInsertAllRequest[T]] = {
- implicit val format = lift(writer)
- implicit val rowFormat = jsonFormat2(Row[T])
+ implicit val format: RootJsonFormat[T] = lift(writer)
+ implicit val rowFormat: RootJsonFormat[Row[T]] = jsonFormat2(Row[T])
jsonFormat4(TableDataInsertAllRequest[T])
}
}
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryDatasets.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryDatasets.scala
index 421051bcc..0bb653809 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryDatasets.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryDatasets.scala
@@ -28,7 +28,7 @@ import pekko.stream.connectors.googlecloud.bigquery.{ BigQueryEndpoints, BigQuer
import pekko.stream.scaladsl.Source
import pekko.{ Done, NotUsed }
-import scala.concurrent.Future
+import scala.concurrent.{ ExecutionContext, Future }
private[scaladsl] trait BigQueryDatasets { this: BigQueryRest =>
@@ -102,7 +102,7 @@ private[scaladsl] trait BigQueryDatasets { this: BigQueryRest =>
settings: GoogleSettings): Future[Dataset] = {
import BigQueryException._
import SprayJsonSupport._
- implicit val ec = ExecutionContexts.parasitic
+ implicit val ec: ExecutionContext = ExecutionContexts.parasitic
val uri = BigQueryEndpoints.datasets(settings.projectId)
Marshal(dataset).to[RequestEntity].flatMap { entity =>
val request = HttpRequest(POST, uri, entity = entity)
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryJobs.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryJobs.scala
index 69c969e26..e1f82dbf7 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryJobs.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryJobs.scala
@@ -43,7 +43,7 @@ import pekko.stream.scaladsl.{ Flow, GraphDSL, Keep, Sink }
import pekko.util.ByteString
import scala.annotation.nowarn
-import scala.concurrent.Future
+import scala.concurrent.{ ExecutionContext, Future }
private[scaladsl] trait BigQueryJobs { this: BigQueryRest =>
@@ -114,7 +114,7 @@ private[scaladsl] trait BigQueryJobs { this: BigQueryRest =>
.fromMaterializer { (mat, attr) =>
import SprayJsonSupport._
import mat.executionContext
- implicit val settings = GoogleAttributes.resolveSettings(mat, attr)
+ implicit val settings: GoogleSettings = GoogleAttributes.resolveSettings(mat, attr)
val BigQuerySettings(loadJobPerTableQuota) = BigQueryAttributes.resolveSettings(mat, attr)
val job = Job(
@@ -168,8 +168,8 @@ private[scaladsl] trait BigQueryJobs { this: BigQueryRest =>
Sink
.fromMaterializer { (mat, attr) =>
import BigQueryException._
- implicit val settings = GoogleAttributes.resolveSettings(mat, attr)
- implicit val ec = ExecutionContexts.parasitic
+ implicit val settings: GoogleSettings = GoogleAttributes.resolveSettings(mat, attr)
+ implicit val ec: ExecutionContext = ExecutionContexts.parasitic
val uri = BigQueryMediaEndpoints.jobs(settings.projectId).withQuery(Query("uploadType" -> "resumable"))
Sink
.lazyFutureSink { () =>
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueries.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueries.scala
index 6330733db..9dac9b823 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueries.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueries.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.stream.connectors.googlecloud.bigquery.scaladsl
import org.apache.pekko
+import pekko.actor.ActorSystem
import pekko.NotUsed
import pekko.dispatch.ExecutionContexts
import pekko.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
@@ -23,14 +24,14 @@ import pekko.http.scaladsl.model.Uri.Query
import pekko.http.scaladsl.model.{ HttpRequest, RequestEntity }
import pekko.http.scaladsl.unmarshalling.FromEntityUnmarshaller
import pekko.stream.RestartSettings
-import pekko.stream.connectors.google.GoogleAttributes
+import pekko.stream.connectors.google.{ GoogleAttributes, GoogleSettings }
import pekko.stream.connectors.google.implicits._
import pekko.stream.connectors.googlecloud.bigquery.model.JobReference
import pekko.stream.connectors.googlecloud.bigquery.model.{ QueryRequest, QueryResponse }
import pekko.stream.connectors.googlecloud.bigquery.{ BigQueryEndpoints, BigQueryException }
import pekko.stream.scaladsl.{ Keep, RestartSource, Sink, Source }
-import scala.concurrent.Future
+import scala.concurrent.{ ExecutionContext, Future }
import scala.concurrent.duration.FiniteDuration
import scala.util.{ Failure, Success }
@@ -70,9 +71,9 @@ private[scaladsl] trait BigQueryQueries { this: BigQueryRest =>
.fromMaterializer { (mat, attr) =>
import BigQueryException._
import SprayJsonSupport._
- implicit val system = mat.system
- implicit val ec = ExecutionContexts.parasitic
- implicit val settings = GoogleAttributes.resolveSettings(mat, attr)
+ implicit val system: ActorSystem = mat.system
+ implicit val ec: ExecutionContext = ExecutionContexts.parasitic
+ implicit val settings: GoogleSettings = GoogleAttributes.resolveSettings(mat, attr)
Source.lazyFutureSource { () =>
for {
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryTableData.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryTableData.scala
index 0e18e4974..3e2f8db7b 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryTableData.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryTableData.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.stream.connectors.googlecloud.bigquery.scaladsl
import org.apache.pekko
+import pekko.actor.ActorSystem
import pekko.NotUsed
import pekko.dispatch.ExecutionContexts
import pekko.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
@@ -22,7 +23,7 @@ import pekko.http.scaladsl.model.HttpMethods.POST
import pekko.http.scaladsl.model.Uri.Query
import pekko.http.scaladsl.model.{ HttpRequest, RequestEntity }
import pekko.http.scaladsl.unmarshalling.{ FromEntityUnmarshaller, FromResponseUnmarshaller }
-import pekko.stream.connectors.google.GoogleAttributes
+import pekko.stream.connectors.google.{ GoogleAttributes, GoogleSettings }
import pekko.stream.connectors.google.http.GoogleHttp
import pekko.stream.connectors.google.implicits._
import pekko.stream.connectors.googlecloud.bigquery.model.{
@@ -131,9 +132,9 @@ private[scaladsl] trait BigQueryTableData { this: BigQueryRest =>
.fromMaterializer { (mat, attr) =>
import BigQueryException._
import SprayJsonSupport._
- implicit val system = mat.system
+ implicit val system: ActorSystem = mat.system
implicit val ec = ExecutionContexts.parasitic
- implicit val settings = GoogleAttributes.resolveSettings(mat, attr)
+ implicit val settings: GoogleSettings = GoogleAttributes.resolveSettings(mat, attr)
val uri = BigQueryEndpoints.tableDataInsertAll(settings.projectId, datasetId, tableId)
val request = HttpRequest(POST, uri)
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryTables.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryTables.scala
index c94e3e2f1..54ede0ed7 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryTables.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryTables.scala
@@ -29,7 +29,7 @@ import pekko.stream.connectors.googlecloud.bigquery.scaladsl.schema.TableSchemaW
import pekko.stream.connectors.googlecloud.bigquery.{ BigQueryEndpoints, BigQueryException }
import pekko.stream.scaladsl.{ Keep, Sink, Source }
-import scala.concurrent.Future
+import scala.concurrent.{ ExecutionContext, Future }
private[scaladsl] trait BigQueryTables { this: BigQueryRest =>
@@ -94,7 +94,7 @@ private[scaladsl] trait BigQueryTables { this: BigQueryRest =>
settings: GoogleSettings): Future[Table] = {
import BigQueryException._
import SprayJsonSupport._
- implicit val ec = ExecutionContexts.parasitic
+ implicit val ec: ExecutionContext = ExecutionContexts.parasitic
val projectId = table.tableReference.projectId.getOrElse(settings.projectId)
val datasetId = table.tableReference.datasetId
val uri = BigQueryEndpoints.tables(projectId, datasetId)
diff --git a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueriesSpec.scala b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueriesSpec.scala
index e4d293453..f98c30999 100644
--- a/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueriesSpec.scala
+++ b/google-cloud-bigquery/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/BigQueryQueriesSpec.scala
@@ -50,7 +50,7 @@ class BigQueryQueriesSpec
jsonFormat10(QueryResponse[T])
}
- implicit val settings = GoogleSettings().copy(credentials = NoCredentials("", ""))
+ implicit val settings: GoogleSettings = GoogleSettings().copy(credentials = NoCredentials("", ""))
val jobId = "jobId"
val pageToken = "pageToken"
diff --git a/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/IntegrationSpec.scala b/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/IntegrationSpec.scala
index 69f7beb12..c6616c520 100644
--- a/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/IntegrationSpec.scala
+++ b/google-cloud-pub-sub-grpc/src/test/scala/docs/scaladsl/IntegrationSpec.scala
@@ -50,7 +50,7 @@ class IntegrationSpec
with OptionValues
with LogCapturing {
- implicit val system = ActorSystem("IntegrationSpec")
+ implicit val system: ActorSystem = ActorSystem("IntegrationSpec")
implicit val defaultPatience = PatienceConfig(timeout = 15.seconds, interval = 50.millis)
diff --git a/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala b/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala
index e66a13982..75b28878e 100644
--- a/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala
+++ b/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala
@@ -23,7 +23,7 @@ import pekko.http.scaladsl.marshalling.Marshal
import pekko.http.scaladsl.model.HttpMethods.POST
import pekko.http.scaladsl.model._
import pekko.http.scaladsl.unmarshalling.{ FromResponseUnmarshaller, Unmarshal, Unmarshaller }
-import pekko.stream.connectors.google.GoogleAttributes
+import pekko.stream.connectors.google.{ GoogleAttributes, GoogleSettings, RequestSettings }
import pekko.stream.connectors.google.http.GoogleHttp
import pekko.stream.connectors.google.implicits._
import pekko.stream.connectors.googlecloud.pubsub._
@@ -142,7 +142,7 @@ private[pubsub] trait PubSubApi {
AcknowledgeRequest(json.asJsObject.fields("ackIds").convertTo[immutable.Seq[String]]: _*)
def write(ar: AcknowledgeRequest): JsValue = JsObject("ackIds" -> ar.ackIds.toJson)
}
- private implicit val pullRequestFormat = DefaultJsonProtocol.jsonFormat2(PullRequest)
+ private implicit val pullRequestFormat = DefaultJsonProtocol.jsonFormat2(PullRequest.apply)
private def scheme: String = if (isEmulated) "http" else "https"
@@ -150,8 +150,8 @@ private[pubsub] trait PubSubApi {
Flow
.fromMaterializer { (mat, attr) =>
import mat.executionContext
- implicit val settings = GoogleAttributes.resolveSettings(mat, attr)
- implicit val requestSettings = settings.requestSettings
+ implicit val settings: GoogleSettings = GoogleAttributes.resolveSettings(mat, attr)
+ implicit val requestSettings: RequestSettings = settings.requestSettings
val url: Uri = Uri.from(
scheme = scheme,
@@ -188,8 +188,8 @@ private[pubsub] trait PubSubApi {
Flow
.fromMaterializer { (mat, attr) =>
import mat.executionContext
- implicit val settings = GoogleAttributes.resolveSettings(mat, attr)
- implicit val requestSettings = settings.requestSettings
+ implicit val settings: GoogleSettings = GoogleAttributes.resolveSettings(mat, attr)
+ implicit val requestSettings: RequestSettings = settings.requestSettings
val url: Uri = Uri.from(
scheme = scheme,
@@ -239,8 +239,8 @@ private[pubsub] trait PubSubApi {
Flow
.fromMaterializer { (mat, attr) =>
import mat.executionContext
- implicit val system = mat.system
- implicit val settings = GoogleAttributes.resolveSettings(mat, attr)
+ implicit val system: ActorSystem = mat.system
+ implicit val settings: GoogleSettings = GoogleAttributes.resolveSettings(mat, attr)
val url: Uri = s"/v1/projects/${settings.projectId}/topics/$topic:publish"
FlowWithContext[PublishRequest, T]
.mapAsync(parallelism) { request =>
diff --git a/google-cloud-pub-sub/src/test/scala/docs/scaladsl/ExampleUsage.scala b/google-cloud-pub-sub/src/test/scala/docs/scaladsl/ExampleUsage.scala
index 42709fc21..afb86f5d4 100644
--- a/google-cloud-pub-sub/src/test/scala/docs/scaladsl/ExampleUsage.scala
+++ b/google-cloud-pub-sub/src/test/scala/docs/scaladsl/ExampleUsage.scala
@@ -31,7 +31,7 @@ import scala.concurrent.{ Future, Promise }
class ExampleUsage {
// #init-system
- implicit val system = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
val config = PubSubConfig()
val topic = "topic1"
val subscription = "subscription1"
diff --git a/google-cloud-pub-sub/src/test/scala/docs/scaladsl/IntegrationSpec.scala b/google-cloud-pub-sub/src/test/scala/docs/scaladsl/IntegrationSpec.scala
index 4bc3ba82d..0849f1dde 100644
--- a/google-cloud-pub-sub/src/test/scala/docs/scaladsl/IntegrationSpec.scala
+++ b/google-cloud-pub-sub/src/test/scala/docs/scaladsl/IntegrationSpec.scala
@@ -45,7 +45,7 @@ class IntegrationSpec
with OptionValues
with LogCapturing {
- private implicit val system = ActorSystem("IntegrationSpec")
+ private implicit val system: ActorSystem = ActorSystem("IntegrationSpec")
override def afterAll(): Unit = TestKit.shutdownActorSystem(system)
diff --git a/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/GooglePubSubSpec.scala b/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/GooglePubSubSpec.scala
index bc9e9e189..29d94c795 100644
--- a/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/GooglePubSubSpec.scala
+++ b/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/GooglePubSubSpec.scala
@@ -42,10 +42,10 @@ class GooglePubSubSpec
with LogCapturing
with BeforeAndAfterAll {
- implicit val defaultPatience =
+ implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = 5.seconds, interval = 100.millis)
- implicit val system = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
override protected def afterAll(): Unit = {
TestKit.shutdownActorSystem(system)
diff --git a/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApiSpec.scala b/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApiSpec.scala
index 73e4cad71..bd8a57f78 100644
--- a/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApiSpec.scala
+++ b/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApiSpec.scala
@@ -53,7 +53,7 @@ class NoopTrustManager extends X509TrustManager {
class PubSubApiSpec extends AnyFlatSpec with BeforeAndAfterAll with ScalaFutures with Matchers with LogCapturing {
- implicit val system = ActorSystem(
+ implicit val system: ActorSystem = ActorSystem(
"PubSubApiSpec",
ConfigFactory
.parseString(
diff --git a/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/Formats.scala b/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/Formats.scala
index 93d5f6b61..003d0f244 100644
--- a/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/Formats.scala
+++ b/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/Formats.scala
@@ -26,13 +26,15 @@ import scala.util.Try
object Formats extends DefaultJsonProtocol {
private final case class CustomerEncryption(encryptionAlgorithm: String, keySha256: String)
- private implicit val customerEncryptionJsonFormat = jsonFormat2(CustomerEncryption)
+
+ private implicit val customerEncryptionJsonFormat: RootJsonFormat[CustomerEncryption] =
+ jsonFormat2(CustomerEncryption.apply)
private final case class Owner(entity: String, entityId: Option[String])
- private implicit val OwnerJsonFormat = jsonFormat2(Owner)
+ private implicit val OwnerJsonFormat: RootJsonFormat[Owner] = jsonFormat2(Owner.apply)
private final case class ProjectTeam(projectNumber: String, team: String)
- private implicit val ProjectTeamJsonFormat = jsonFormat2(ProjectTeam)
+ private implicit val ProjectTeamJsonFormat: RootJsonFormat[ProjectTeam] = jsonFormat2(ProjectTeam.apply)
private final case class ObjectAccessControls(kind: String,
id: String,
@@ -129,7 +131,7 @@ object Formats extends DefaultJsonProtocol {
prefixes: Option[List[String]],
items: Option[List[StorageObjectJson]])
- private implicit val bucketInfoJsonFormat = jsonFormat6(BucketInfoJson)
+ private implicit val bucketInfoJsonFormat: RootJsonFormat[BucketInfoJson] = jsonFormat6(BucketInfoJson.apply)
/**
* Google API rewrite response object
@@ -144,7 +146,8 @@ object Formats extends DefaultJsonProtocol {
rewriteToken: Option[String],
resource: Option[StorageObjectJson])
- private implicit val rewriteResponseFormat = jsonFormat6(RewriteResponseJson)
+ private implicit val rewriteResponseFormat: RootJsonFormat[RewriteResponseJson] =
+ jsonFormat6(RewriteResponseJson.apply)
/**
* Google API bucket response object
@@ -159,7 +162,7 @@ object Formats extends DefaultJsonProtocol {
selfLink: String,
etag: String)
- implicit val bucketInfoFormat = jsonFormat2(BucketInfo)
+ implicit val bucketInfoFormat: RootJsonFormat[BucketInfo] = jsonFormat2(BucketInfo.apply)
implicit object BucketListResultReads extends RootJsonReader[BucketListResult] {
override def read(json: JsValue): BucketListResult = {
@@ -172,7 +175,7 @@ object Formats extends DefaultJsonProtocol {
}
}
- private implicit val bucketListResultJsonReads = jsonFormat4(BucketListResultJson)
+ private implicit val bucketListResultJsonReads = jsonFormat4(BucketListResultJson.apply)
implicit object RewriteResponseReads extends RootJsonReader[RewriteResponse] {
override def read(json: JsValue): RewriteResponse = {
diff --git a/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStream.scala b/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStream.scala
index b9c8b9a77..c704d093b 100644
--- a/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStream.scala
+++ b/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStream.scala
@@ -37,7 +37,7 @@ import pekko.{ Done, NotUsed }
import spray.json._
import scala.annotation.nowarn
-import scala.concurrent.Future
+import scala.concurrent.{ ExecutionContext, Future }
@InternalApi private[storage] object GCStorageStream {
@@ -55,7 +55,7 @@ import scala.concurrent.Future
val uri = Uri(gcsSettings.endpointUrl)
.withPath(Path(gcsSettings.basePath) / "b")
.withQuery(Query("project" -> settings.projectId))
- implicit val ec = parasitic
+ implicit val ec: ExecutionContext = parasitic
val request = Marshal(BucketInfo(bucketName, location)).to[RequestEntity].map { entity =>
HttpRequest(POST, uri, entity = entity)
}
@@ -142,7 +142,7 @@ import scala.concurrent.Future
metadata: Option[Map[String, String]] = None): Sink[ByteString, Future[StorageObject]] =
Sink
.fromMaterializer { (mat, attr) =>
- implicit val settings = {
+ implicit val settings: GoogleSettings = {
val s = resolveSettings(mat, attr)
s.copy(requestSettings = s.requestSettings.copy(uploadChunkSize = chunkSize))
}
@@ -226,7 +226,7 @@ import scala.concurrent.Future
private def makeRequestSource[T: FromResponseUnmarshaller](request: Future[HttpRequest]): Source[T, NotUsed] =
Source
.fromMaterializer { (mat, attr) =>
- implicit val settings = resolveSettings(mat, attr)
+ implicit val settings: GoogleSettings = resolveSettings(mat, attr)
Source.lazyFuture { () =>
request.flatMap { request =>
GoogleHttp()(mat.system).singleAuthenticatedRequest[T](request)
diff --git a/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/GCSExtSpec.scala b/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/GCSExtSpec.scala
index 0f4272037..659e9b1d0 100644
--- a/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/GCSExtSpec.scala
+++ b/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/GCSExtSpec.scala
@@ -31,7 +31,7 @@ class GCSExtSpec extends AnyFlatSpec with Matchers with LogCapturing {
"pekko.connectors.google.cloud-storage.endpoint-url" -> endpointUrl,
"pekko.connectors.google.cloud-storage.base-path" -> basePath).asJava)
- implicit val system = ActorSystem.create("gcs", config)
+ implicit val system: ActorSystem = ActorSystem.create("gcs", config)
val ext = GCSExt(system)
ext.settings.endpointUrl shouldBe endpointUrl
diff --git a/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/GCStorageExtSpec.scala b/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/GCStorageExtSpec.scala
index e2836ea71..9fe31dd4d 100644
--- a/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/GCStorageExtSpec.scala
+++ b/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/GCStorageExtSpec.scala
@@ -42,7 +42,7 @@ class GCStorageExtSpec extends AnyFlatSpec with Matchers with LogCapturing {
"pekko.connectors.google.cloud.storage.base-path" -> basePath,
"pekko.connectors.google.cloud.storage.token-url" -> tokenUrl,
"pekko.connectors.google.cloud.storage.token-scope" -> tokenScope).asJava)
- implicit val system = ActorSystem.create("gcStorage", config)
+ implicit val system: ActorSystem = ActorSystem.create("gcStorage", config)
@nowarn("msg=deprecated")
val ext = GCStorageExt(system)
diff --git a/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/WithMaterializerGlobal.scala b/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/WithMaterializerGlobal.scala
index 6efa9399e..5edc3fdba 100644
--- a/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/WithMaterializerGlobal.scala
+++ b/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/WithMaterializerGlobal.scala
@@ -19,7 +19,7 @@ import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
import org.scalatest.{ BeforeAndAfterAll, BeforeAndAfterEach }
-import scala.concurrent.Await
+import scala.concurrent.{ Await, ExecutionContext }
import scala.concurrent.duration._
trait WithMaterializerGlobal
@@ -30,7 +30,7 @@ trait WithMaterializerGlobal
with IntegrationPatience
with Matchers {
implicit val actorSystem = ActorSystem("test")
- implicit val ec = actorSystem.dispatcher
+ implicit val ec: ExecutionContext = actorSystem.dispatcher
override protected def afterAll(): Unit = {
super.afterAll()
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/PaginatedRequest.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/PaginatedRequest.scala
index 48f1cdb2c..2c0baca4a 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/PaginatedRequest.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/PaginatedRequest.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.stream.connectors.google
import org.apache.pekko
+import pekko.actor.ActorSystem
import pekko.annotation.InternalApi
import pekko.dispatch.ExecutionContexts
import pekko.http.scaladsl.model.HttpMethods.GET
@@ -51,8 +52,8 @@ private[connectors] object PaginatedRequest {
Source
.fromMaterializer { (mat, attr) =>
- implicit val system = mat.system
- implicit val settings = GoogleAttributes.resolveSettings(mat, attr)
+ implicit val system: ActorSystem = mat.system
+ implicit val settings: GoogleSettings = GoogleAttributes.resolveSettings(mat, attr)
val requestWithPageToken = addPageToken(request, query)
Source.unfoldAsync[Either[Done, Option[String]], Out](Right(initialPageToken)) {
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
index f80864062..630381f59 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.stream.connectors.google
import org.apache.pekko
+import pekko.actor.ActorSystem
import pekko.NotUsed
import pekko.annotation.InternalApi
import pekko.http.scaladsl.model.HttpMethods.{ POST, PUT }
@@ -56,7 +57,7 @@ private[connectors] object ResumableUpload {
.fromMaterializer { (mat, attr) =>
import mat.executionContext
implicit val materializer = mat
- implicit val settings = GoogleAttributes.resolveSettings(mat, attr)
+ implicit val settings: GoogleSettings = GoogleAttributes.resolveSettings(mat, attr)
val uploadChunkSize = settings.requestSettings.uploadChunkSize
val in = Flow[ByteString]
@@ -95,7 +96,7 @@ private[connectors] object ResumableUpload {
private def initiateSession(request: HttpRequest)(implicit mat: Materializer,
settings: GoogleSettings): Future[Uri] = {
- implicit val system = mat.system
+ implicit val system: ActorSystem = mat.system
import implicits._
implicit val um = Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
@@ -111,7 +112,7 @@ private[connectors] object ResumableUpload {
private def uploadChunk[T: FromResponseUnmarshaller](
request: HttpRequest)(implicit mat: Materializer): Flow[Either[T, MaybeLast[Chunk]], Try[Option[T]], NotUsed] = {
- implicit val system = mat.system
+ implicit val system: ActorSystem = mat.system
val um = Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
response.status match {
@@ -146,7 +147,7 @@ private[connectors] object ResumableUpload {
request: HttpRequest,
chunk: Future[MaybeLast[Chunk]])(
implicit mat: Materializer, settings: GoogleSettings): Future[Either[T, MaybeLast[Chunk]]] = {
- implicit val system = mat.system
+ implicit val system: ActorSystem = mat.system
import implicits._
implicit val um = Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleComputeMetadata.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleComputeMetadata.scala
index dbe5dc73e..5acc124a3 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleComputeMetadata.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleComputeMetadata.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.stream.connectors.google.auth
import org.apache.pekko
+import pekko.actor.ActorSystem
import pekko.annotation.InternalApi
import pekko.http.scaladsl.Http
import pekko.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
@@ -42,7 +43,7 @@ private[auth] object GoogleComputeMetadata {
clock: Clock): Future[AccessToken] = {
import SprayJsonSupport._
import mat.executionContext
- implicit val system = mat.system
+ implicit val system: ActorSystem = mat.system
for {
response <- Http().singleRequest(tokenRequest)
token <- Unmarshal(response.entity).to[AccessToken]
@@ -52,7 +53,7 @@ private[auth] object GoogleComputeMetadata {
def getProjectId()(
implicit mat: Materializer): Future[String] = {
import mat.executionContext
- implicit val system = mat.system
+ implicit val system: ActorSystem = mat.system
for {
response <- Http().singleRequest(projectIdRequest)
projectId <- Unmarshal(response.entity).to[String]
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala
index 0da749c2d..259770100 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.stream.connectors.google.auth
import org.apache.pekko
+import pekko.actor.ActorSystem
import pekko.annotation.InternalApi
import pekko.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import pekko.http.scaladsl.model.HttpMethods.POST
@@ -42,7 +43,7 @@ private[auth] object GoogleOAuth2 {
import GoogleOAuth2Exception._
import SprayJsonSupport._
import implicits._
- implicit val system = mat.system
+ implicit val system: ActorSystem = mat.system
try {
val entity = FormData(
@@ -71,5 +72,5 @@ private[auth] object GoogleOAuth2 {
}
final case class JwtClaimContent(scope: String)
- implicit val jwtClaimContentFormat: JsonFormat[JwtClaimContent] = jsonFormat1(JwtClaimContent)
+ implicit val jwtClaimContentFormat: JsonFormat[JwtClaimContent] = jsonFormat1(JwtClaimContent.apply)
}
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Credentials.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Credentials.scala
index c28868c6d..5c30f45e7 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Credentials.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Credentials.scala
@@ -38,7 +38,7 @@ private[auth] final class GoogleOAuth2Credentials(credentials: OAuth2Credentials
Await.result(requestMetadata, Duration.Inf)
override def getRequestMetadata(uri: URI, executor: Executor, callback: RequestMetadataCallback): Unit = {
- implicit val ec = ExecutionContext.fromExecutor(executor)
+ implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(executor)
requestMetadata.onComplete {
case Success(metadata) => callback.onSuccess(metadata)
case Failure(ex) => callback.onFailure(ex)
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Exception.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Exception.scala
index 3b6b02db8..ed392199a 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Exception.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Exception.scala
@@ -28,7 +28,8 @@ private[google] object GoogleOAuth2Exception {
private val internalFailure = "internal_failure"
private final case class OAuth2ErrorResponse(error: Option[String], error_description: Option[String])
- private implicit val oAuth2ErrorResponseFormat: RootJsonFormat[OAuth2ErrorResponse] = jsonFormat2(OAuth2ErrorResponse)
+ private implicit val oAuth2ErrorResponseFormat: RootJsonFormat[OAuth2ErrorResponse] =
+ jsonFormat2(OAuth2ErrorResponse.apply)
implicit val unmarshaller: FromResponseUnmarshaller[Throwable] =
Unmarshaller
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/ServiceAccountCredentials.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/ServiceAccountCredentials.scala
index de20e49cb..5f38c016e 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/ServiceAccountCredentials.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/ServiceAccountCredentials.scala
@@ -57,7 +57,7 @@ private[connectors] object ServiceAccountCredentials {
final case class ServiceAccountCredentialsFile(project_id: String, client_email: String, private_key: String)
implicit val serviceAccountCredentialsFormat: RootJsonFormat[ServiceAccountCredentialsFile] = jsonFormat3(
- ServiceAccountCredentialsFile)
+ ServiceAccountCredentialsFile.apply)
}
@InternalApi
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessCredentials.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessCredentials.scala
index fa81d92d9..cb63a4b28 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessCredentials.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessCredentials.scala
@@ -61,7 +61,7 @@ private[connectors] object UserAccessCredentials {
refresh_token: String,
quota_project_id: String)
implicit val userAccessCredentialsFormat: RootJsonFormat[UserAccessCredentialsFile] = jsonFormat4(
- UserAccessCredentialsFile)
+ UserAccessCredentialsFile.apply)
}
@InternalApi
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessMetadata.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessMetadata.scala
index 93bb65010..0ae2dbb5a 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessMetadata.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessMetadata.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.stream.connectors.google.auth
import org.apache.pekko
+import pekko.actor.ActorSystem
import pekko.annotation.InternalApi
import pekko.http.scaladsl.Http
import pekko.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
@@ -45,7 +46,7 @@ private[auth] object UserAccessMetadata {
clock: Clock): Future[AccessToken] = {
import SprayJsonSupport._
import mat.executionContext
- implicit val system = mat.system
+ implicit val system: ActorSystem = mat.system
for {
response <- Http().singleRequest(tokenRequest(clientId, clientSecret, refreshToken))
token <- Unmarshal(response.entity).to[AccessToken]
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttp.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttp.scala
index 0f570a5dd..ba364683e 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttp.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttp.scala
@@ -76,7 +76,7 @@ private[connectors] final class GoogleHttp private (val http: HttpExt) extends A
def singleAuthenticatedRequest[T](request: HttpRequest)(
implicit settings: GoogleSettings,
um: FromResponseUnmarshaller[T]): Future[T] = Retry(settings.requestSettings.retrySettings) {
- implicit val requestSettings = settings.requestSettings
+ implicit val requestSettings: RequestSettings = settings.requestSettings
addAuth(request).flatMap(singleRequest(_))(ExecutionContexts.parasitic)
}
@@ -110,7 +110,7 @@ private[connectors] final class GoogleHttp private (val http: HttpExt) extends A
parallelism: Int = 1): FlowWithContext[HttpRequest, Ctx, Try[T], Ctx, Future[HostConnectionPool]] =
FlowWithContext.fromTuples {
Flow.fromMaterializer { (mat, attr) =>
- implicit val settings = GoogleAttributes.resolveSettings(mat, attr)
+ implicit val settings: GoogleSettings = GoogleAttributes.resolveSettings(mat, attr)
val p = if (port == -1) if (https) 443 else 80 else port
val uriFlow = FlowWithContext[HttpRequest, Ctx].map(addStandardQuery)
@@ -163,7 +163,7 @@ private[connectors] final class GoogleHttp private (val http: HttpExt) extends A
.fold(settings.requestSettings.queryString)(_.concat(settings.requestSettings.`&queryString`)))))
private def addAuth(request: HttpRequest)(implicit settings: GoogleSettings): Future[HttpRequest] = {
- implicit val requestSettings = settings.requestSettings
+ implicit val requestSettings: RequestSettings = settings.requestSettings
settings.credentials
.get()
.map { token =>
diff --git a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/PaginatedRequestSpec.scala b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/PaginatedRequestSpec.scala
index b637861e7..d62aa6e95 100644
--- a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/PaginatedRequestSpec.scala
+++ b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/PaginatedRequestSpec.scala
@@ -43,7 +43,7 @@ class PaginatedRequestSpec
super.afterAll()
}
- implicit val patience = PatienceConfig(remainingOrDefault)
+ implicit val patience: PatienceConfig = PatienceConfig(remainingOrDefault)
implicit val paginated: Paginated[JsValue] = _.asJsObject.fields.get("pageToken").flatMap {
case JsString(value) => Some(value)
case _ => None
diff --git a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/ResumableUploadSpec.scala b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/ResumableUploadSpec.scala
index f5e27aebc..aa3559cab 100644
--- a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/ResumableUploadSpec.scala
+++ b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/ResumableUploadSpec.scala
@@ -40,7 +40,7 @@ class ResumableUploadSpec
with ScalaFutures
with HoverflySupport {
- implicit val patience = PatienceConfig(remainingOrDefault)
+ implicit val patience: PatienceConfig = PatienceConfig(remainingOrDefault)
override def afterAll(): Unit = {
TestKit.shutdownActorSystem(system)
diff --git a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Spec.scala b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Spec.scala
index 5eb4f3956..6c20d12a9 100644
--- a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Spec.scala
+++ b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Spec.scala
@@ -15,7 +15,7 @@ package org.apache.pekko.stream.connectors.google.auth
import org.apache.pekko
import pekko.actor.ActorSystem
-import pekko.stream.connectors.google.{ GoogleSettings, HoverflySupport }
+import pekko.stream.connectors.google.{ GoogleSettings, HoverflySupport, RequestSettings }
import pekko.testkit.TestKit
import io.specto.hoverfly.junit.core.SimulationSource.dsl
import io.specto.hoverfly.junit.core.model.RequestFieldMatcher.newRegexMatcher
@@ -42,10 +42,10 @@ class GoogleOAuth2Spec
TestKit.shutdownActorSystem(system)
super.afterAll()
}
- implicit val defaultPatience = PatienceConfig(remainingOrDefault)
+ implicit val defaultPatience: PatienceConfig = PatienceConfig(remainingOrDefault)
implicit val executionContext: ExecutionContext = system.dispatcher
- implicit val settings = GoogleSettings(system)
+ implicit val settings: GoogleSettings = GoogleSettings(system)
implicit val clock = Clock.systemUTC()
lazy val privateKey = {
@@ -73,7 +73,7 @@ class GoogleOAuth2Spec
success("""{"access_token": "token", "token_type": "String", "expires_in": 3600}""",
"application/json"))))
- implicit val settings = GoogleSettings().requestSettings
+ implicit val settings: RequestSettings = GoogleSettings().requestSettings
GoogleOAuth2.getAccessToken("email", privateKey, scopes).futureValue should matchPattern {
case AccessToken("token", exp) if exp > (System.currentTimeMillis / 1000L + 3000L) =>
}
diff --git a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala
index b3376ebba..967a067e0 100644
--- a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala
+++ b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala
@@ -43,7 +43,8 @@ class OAuth2CredentialsSpec
}
import system.dispatcher
- implicit val settings = GoogleSettings().requestSettings
+
+ implicit val settings: RequestSettings = GoogleSettings().requestSettings
implicit val clock = Clock.systemUTC()
final object AccessTokenProvider {
diff --git a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttpSpec.scala b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttpSpec.scala
index ae2928f2f..125d406cf 100644
--- a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttpSpec.scala
+++ b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttpSpec.scala
@@ -76,7 +76,7 @@ class GoogleHttpSpec
http
}
- implicit val settings = GoogleSettings().requestSettings
+ implicit val settings: RequestSettings = GoogleSettings().requestSettings
"GoogleHttp" must {
@@ -163,7 +163,7 @@ class GoogleHttpSpec
when(credentials.get()(any[ExecutionContext], any[RequestSettings])).thenReturn(
Future.failed(GoogleOAuth2Exception(ErrorInfo())),
Future.failed(new AnotherException))
- implicit val settingsWithMockedCredentials = GoogleSettings().copy(credentials = credentials)
+ implicit val settingsWithMockedCredentials: GoogleSettings = GoogleSettings().copy(credentials = credentials)
val http = mockHttp
when(
diff --git a/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmFlows.scala b/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmFlows.scala
index 8e7503ab8..be9962604 100644
--- a/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmFlows.scala
+++ b/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmFlows.scala
@@ -37,7 +37,7 @@ private[fcm] object FcmFlows {
private[fcm] def fcmWithData[T](conf: FcmSettings): Flow[(FcmNotification, T), (FcmResponse, T), NotUsed] =
Flow
.fromMaterializer { (mat, attr) =>
- implicit val settings = resolveSettings(conf)(mat, attr)
+ implicit val settings: GoogleSettings = resolveSettings(conf)(mat, attr)
val sender = new FcmSender()
Flow[(FcmNotification, T)].mapAsync(conf.maxConcurrentConnections) {
case (notification, data) =>
@@ -54,7 +54,7 @@ private[fcm] object FcmFlows {
private[fcm] def fcm(conf: FcmSettings): Flow[FcmNotification, FcmResponse, NotUsed] =
Flow
.fromMaterializer { (mat, attr) =>
- implicit val settings = resolveSettings(conf)(mat, attr)
+ implicit val settings: GoogleSettings = resolveSettings(conf)(mat, attr)
val sender = new FcmSender()
Flow[FcmNotification].mapAsync(conf.maxConcurrentConnections) { notification =>
sender.send(Http(mat.system), FcmSend(conf.isTest, notification))(mat, settings)
diff --git a/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmJsonSupport.scala b/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmJsonSupport.scala
index ed50203e5..f61707748 100644
--- a/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmJsonSupport.scala
+++ b/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmJsonSupport.scala
@@ -86,11 +86,13 @@ private[fcm] object FcmJsonSupport extends DefaultJsonProtocol with SprayJsonSup
}
// app -> google
- implicit val webPushNotificationJsonFormat: RootJsonFormat[WebPushNotification] = jsonFormat3(WebPushNotification)
+ implicit val webPushNotificationJsonFormat: RootJsonFormat[WebPushNotification] =
+ jsonFormat3(WebPushNotification.apply)
implicit val webPushConfigJsonFormat: RootJsonFormat[WebPushConfig] = jsonFormat3(WebPushConfig.apply)
- implicit val androidNotificationJsonFormat: RootJsonFormat[AndroidNotification] = jsonFormat11(AndroidNotification)
+ implicit val androidNotificationJsonFormat: RootJsonFormat[AndroidNotification] =
+ jsonFormat11(AndroidNotification.apply)
implicit val androidConfigJsonFormat: RootJsonFormat[AndroidConfig] = jsonFormat6(AndroidConfig.apply)
- implicit val basicNotificationJsonFormat: RootJsonFormat[BasicNotification] = jsonFormat2(BasicNotification)
+ implicit val basicNotificationJsonFormat: RootJsonFormat[BasicNotification] = jsonFormat2(BasicNotification.apply)
implicit val sendableFcmNotificationJsonFormat: RootJsonFormat[FcmNotification] = jsonFormat8(FcmNotification.apply)
- implicit val fcmSendJsonFormat: RootJsonFormat[FcmSend] = jsonFormat2(FcmSend)
+ implicit val fcmSendJsonFormat: RootJsonFormat[FcmSend] = jsonFormat2(FcmSend.apply)
}
diff --git a/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmJsonSupport.scala b/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmJsonSupport.scala
index c021488e5..7364d0e40 100644
--- a/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmJsonSupport.scala
+++ b/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmJsonSupport.scala
@@ -249,10 +249,10 @@ private[fcm] object FcmJsonSupport extends DefaultJsonProtocol with SprayJsonSup
implicit val fcmOptionsJsonFormat: RootJsonFormat[FcmOptions] = jsonFormat1(FcmOptions.apply)
implicit val apnsFcmOptionsJsonFormat: RootJsonFormat[ApnsFcmOptions] = jsonFormat2(ApnsFcmOptions.apply)
implicit val webPushFcmOptionsJsonFormat: RootJsonFormat[WebPushFcmOptions] = jsonFormat2(WebPushFcmOptions.apply)
- implicit val androidColorJsonFormat: RootJsonFormat[Color] = jsonFormat4(Color)
+ implicit val androidColorJsonFormat: RootJsonFormat[Color] = jsonFormat4(Color.apply)
implicit val androidLightSettingsJsonFormat: RootJsonFormat[LightSettings] = jsonFormat3(LightSettings.apply)
implicit val androidConfigJsonFormat: RootJsonFormat[AndroidConfig] = jsonFormat8(AndroidConfig.apply)
implicit val basicNotificationJsonFormat: RootJsonFormat[BasicNotification] = jsonFormat3(BasicNotification.apply)
implicit val mainFcmNotificationJsonFormat: RootJsonFormat[FcmNotification] = jsonFormat9(FcmNotification.apply)
- implicit val fcmSendJsonFormat: RootJsonFormat[FcmSend] = jsonFormat2(FcmSend)
+ implicit val fcmSendJsonFormat: RootJsonFormat[FcmSend] = jsonFormat2(FcmSend.apply)
}
diff --git a/google-fcm/src/test/scala/docs/scaladsl/FcmExamples.scala b/google-fcm/src/test/scala/docs/scaladsl/FcmExamples.scala
index 4817286f2..072879b2e 100644
--- a/google-fcm/src/test/scala/docs/scaladsl/FcmExamples.scala
+++ b/google-fcm/src/test/scala/docs/scaladsl/FcmExamples.scala
@@ -13,9 +13,9 @@
package docs.scaladsl
-import org.apache.pekko.actor.ActorSystem
//#imports
import org.apache.pekko
+import pekko.actor.ActorSystem
import pekko.stream.connectors.google.firebase.fcm.FcmSettings
import pekko.stream.connectors.google.firebase.fcm.v1.models._
import pekko.stream.connectors.google.firebase.fcm.v1.scaladsl.GoogleFcm
@@ -28,7 +28,7 @@ import scala.concurrent.Future
class FcmExamples {
- implicit val system = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
// #simple-send
val fcmConfig = FcmSettings()
diff --git a/google-fcm/src/test/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSenderSpec.scala b/google-fcm/src/test/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSenderSpec.scala
index 961925fa4..c37111b3e 100644
--- a/google-fcm/src/test/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSenderSpec.scala
+++ b/google-fcm/src/test/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSenderSpec.scala
@@ -52,13 +52,13 @@ class FcmSenderSpec
override def afterAll(): Unit =
TestKit.shutdownActorSystem(system)
- implicit val defaultPatience =
+ implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = 2.seconds, interval = 50.millis)
implicit val executionContext: ExecutionContext = system.dispatcher
implicit val conf = FcmSettings()
- implicit val settings = GoogleSettings().copy(projectId = "projectId")
+ implicit val settings: GoogleSettings = GoogleSettings().copy(projectId = "projectId")
"FcmSender" should {
diff --git a/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseFlowStage.scala b/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseFlowStage.scala
index 35c807965..952dace9b 100644
--- a/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseFlowStage.scala
+++ b/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseFlowStage.scala
@@ -36,7 +36,7 @@ private[hbase] class HBaseFlowStage[A](settings: HTableSettings[A]) extends Grap
override protected def logSource = classOf[HBaseFlowStage[A]]
- implicit val connection = connect(settings.conf)
+ implicit val connection: Connection = connect(settings.conf)
lazy val table: Table = getOrCreateTable(settings.tableName, settings.columnFamilies).get
diff --git a/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseSourceStage.scala b/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseSourceStage.scala
index 182b65fbd..015401eef 100644
--- a/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseSourceStage.scala
+++ b/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/impl/HBaseSourceStage.scala
@@ -17,7 +17,7 @@ import org.apache.pekko
import pekko.stream.{ Attributes, Outlet, SourceShape }
import pekko.stream.connectors.hbase.HTableSettings
import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler, StageLogging }
-import org.apache.hadoop.hbase.client.{ Result, Scan, Table }
+import org.apache.hadoop.hbase.client.{ Connection, Result, Scan, Table }
import scala.util.control.NonFatal
@@ -40,7 +40,7 @@ private[hbase] final class HBaseSourceLogic[A](scan: Scan,
with StageLogging
with HBaseCapabilities {
- implicit val connection = connect(settings.conf)
+ implicit val connection: Connection = connect(settings.conf)
lazy val table: Table = getOrCreateTable(settings.tableName, settings.columnFamilies).get
private var results: java.util.Iterator[Result] = null
diff --git a/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala b/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
index e80a1a8ad..3720dc6da 100644
--- a/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
+++ b/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
@@ -41,7 +41,7 @@ class HBaseStageSpec
with BeforeAndAfterAll
with LogCapturing {
- implicit val defaultPatience =
+ implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = 5.seconds, interval = 500.millis)
// #create-converter-put
diff --git a/hdfs/src/test/scala/docs/scaladsl/HdfsWriterSpec.scala b/hdfs/src/test/scala/docs/scaladsl/HdfsWriterSpec.scala
index 062b6a1f5..74810923e 100644
--- a/hdfs/src/test/scala/docs/scaladsl/HdfsWriterSpec.scala
+++ b/hdfs/src/test/scala/docs/scaladsl/HdfsWriterSpec.scala
@@ -253,7 +253,7 @@ class HdfsWriterSpec
committedOffsets = committedOffsets :+ offset
val resF = Source(messagesFromKafka)
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
// Transform message so that we can write to hdfs
HdfsWriteMessage(ByteString(book.title), kafkaMessage.offset)
diff --git a/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitJsonSupport.scala b/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitJsonSupport.scala
index c396b624a..19f33fed2 100644
--- a/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitJsonSupport.scala
+++ b/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitJsonSupport.scala
@@ -227,5 +227,5 @@ private[pushkit] object PushKitJsonSupport extends DefaultJsonProtocol with Spra
implicit val webNotificationJsonFormat: RootJsonFormat[WebNotification] = jsonFormat14(WebNotification.apply)
implicit val pushKitNotificationJsonFormat: RootJsonFormat[PushKitNotification] = jsonFormat8(
PushKitNotification.apply)
- implicit val pushKitSendJsonFormat: RootJsonFormat[PushKitSend] = jsonFormat2(PushKitSend)
+ implicit val pushKitSendJsonFormat: RootJsonFormat[PushKitSend] = jsonFormat2(PushKitSend.apply)
}
diff --git a/huawei-push-kit/src/test/scala/docs/scaladsl/PushKitExamples.scala b/huawei-push-kit/src/test/scala/docs/scaladsl/PushKitExamples.scala
index 76bde5add..39c613cf0 100644
--- a/huawei-push-kit/src/test/scala/docs/scaladsl/PushKitExamples.scala
+++ b/huawei-push-kit/src/test/scala/docs/scaladsl/PushKitExamples.scala
@@ -13,9 +13,9 @@
package docs.scaladsl
-import org.apache.pekko.actor.ActorSystem
//#imports
import org.apache.pekko
+import pekko.actor.ActorSystem
import pekko.stream.connectors.huawei.pushkit._
import pekko.stream.connectors.huawei.pushkit.scaladsl.HmsPushKit
import pekko.stream.connectors.huawei.pushkit.models.AndroidConfig
@@ -38,7 +38,7 @@ import scala.concurrent.Future
class PushKitExamples {
- implicit val system = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
// #simple-send
val config = HmsSettings()
diff --git a/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApiSpec.scala b/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApiSpec.scala
index ba6568b91..eed0b781b 100644
--- a/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApiSpec.scala
+++ b/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApiSpec.scala
@@ -48,7 +48,7 @@ class HmsTokenApiSpec
override def afterAll() =
TestKit.shutdownActorSystem(system)
- implicit val defaultPatience =
+ implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = 2.seconds, interval = 50.millis)
val config = HmsSettings()
diff --git a/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSenderSpec.scala b/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSenderSpec.scala
index dda2d6d8b..8f739a32d 100644
--- a/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSenderSpec.scala
+++ b/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSenderSpec.scala
@@ -50,7 +50,7 @@ class PushKitSenderSpec
override def afterAll() =
TestKit.shutdownActorSystem(system)
- implicit val defaultPatience =
+ implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = 6.seconds, interval = 50.millis)
implicit val executionContext: ExecutionContext = system.dispatcher
diff --git a/influxdb/src/test/scala/docs/scaladsl/FlowSpec.scala b/influxdb/src/test/scala/docs/scaladsl/FlowSpec.scala
index a5ee57d67..f2f9f51c4 100644
--- a/influxdb/src/test/scala/docs/scaladsl/FlowSpec.scala
+++ b/influxdb/src/test/scala/docs/scaladsl/FlowSpec.scala
@@ -43,7 +43,7 @@ class FlowSpec
with ScalaFutures
with LogCapturing {
- implicit val system = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
final val DatabaseName = this.getClass.getSimpleName
@@ -102,7 +102,7 @@ class FlowSpec
committedOffsets = committedOffsets :+ offset
val f1 = Source(messagesFromKafka)
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val cpu = kafkaMessage.cpu
println("hostname: " + cpu.getHostname)
@@ -111,7 +111,7 @@ class FlowSpec
.groupedWithin(10, 50.millis)
.via(
InfluxDbFlow.typedWithPassThrough(classOf[InfluxDbFlowCpu]))
- .map { messages: Seq[InfluxDbWriteResult[InfluxDbFlowCpu, KafkaOffset]] =>
+ .map { (messages: Seq[InfluxDbWriteResult[InfluxDbFlowCpu, KafkaOffset]]) =>
messages.foreach { message =>
commitToKafka(message.writeMessage.passThrough)
}
diff --git a/influxdb/src/test/scala/docs/scaladsl/InfluxDbSourceSpec.scala b/influxdb/src/test/scala/docs/scaladsl/InfluxDbSourceSpec.scala
index 14a842a49..b4bea095c 100644
--- a/influxdb/src/test/scala/docs/scaladsl/InfluxDbSourceSpec.scala
+++ b/influxdb/src/test/scala/docs/scaladsl/InfluxDbSourceSpec.scala
@@ -39,7 +39,7 @@ class InfluxDbSourceSpec
final val DatabaseName = "InfluxDbSourceSpec"
- implicit val system = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
implicit var influxDB: InfluxDB = _
diff --git a/influxdb/src/test/scala/docs/scaladsl/InfluxDbSpec.scala b/influxdb/src/test/scala/docs/scaladsl/InfluxDbSpec.scala
index 5da0b057a..89f76bee9 100644
--- a/influxdb/src/test/scala/docs/scaladsl/InfluxDbSpec.scala
+++ b/influxdb/src/test/scala/docs/scaladsl/InfluxDbSpec.scala
@@ -41,7 +41,7 @@ class InfluxDbSpec
with ScalaFutures
with LogCapturing {
- implicit val system = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
final val DatabaseName = this.getClass.getSimpleName
@@ -81,7 +81,7 @@ class InfluxDbSpec
// #run-typed
val f1 = InfluxDbSource
.typed(classOf[InfluxDbSpecCpu], InfluxDbReadSettings(), influxDB, query)
- .map { cpu: InfluxDbSpecCpu =>
+ .map { (cpu: InfluxDbSpecCpu) =>
{
val clonedCpu = cpu.cloneAt(cpu.getTime.plusSeconds(60000))
List(InfluxDbWriteMessage(clonedCpu))
diff --git a/ironmq/src/main/scala/org/apache/pekko/stream/connectors/ironmq/impl/IronMqPullStage.scala b/ironmq/src/main/scala/org/apache/pekko/stream/connectors/ironmq/impl/IronMqPullStage.scala
index 40d62a432..7df3cbc73 100644
--- a/ironmq/src/main/scala/org/apache/pekko/stream/connectors/ironmq/impl/IronMqPullStage.scala
+++ b/ironmq/src/main/scala/org/apache/pekko/stream/connectors/ironmq/impl/IronMqPullStage.scala
@@ -125,12 +125,12 @@ private[ironmq] final class IronMqPullStage(queueName: String, settings: IronMqS
buffer = buffer.tail
}
- private val updateBuffer = getAsyncCallback { xs: List[ReservedMessage] =>
+ private val updateBuffer = getAsyncCallback { (xs: List[ReservedMessage]) =>
buffer = buffer ::: xs
deliveryMessages()
}
- private val updateFetching = getAsyncCallback { x: Boolean =>
+ private val updateFetching = getAsyncCallback { (x: Boolean) =>
fetching = x
}
}
diff --git a/ironmq/src/test/scala/org/apache/pekko/stream/connectors/ironmq/IronMqSpec.scala b/ironmq/src/test/scala/org/apache/pekko/stream/connectors/ironmq/IronMqSpec.scala
index 87d04f4bb..ba080457e 100644
--- a/ironmq/src/test/scala/org/apache/pekko/stream/connectors/ironmq/IronMqSpec.scala
+++ b/ironmq/src/test/scala/org/apache/pekko/stream/connectors/ironmq/IronMqSpec.scala
@@ -40,7 +40,7 @@ abstract class IronMqSpec
override implicit val patienceConfig: PatienceConfig = PatienceConfig(timeout = 15.seconds, interval = 1.second)
val DefaultActorSystemTerminateTimeout: Duration = 10.seconds
- private implicit val ec = ExecutionContext.global
+ private implicit val ec: ExecutionContext = ExecutionContext.global
private var mutableIronMqClient = Option.empty[IronMqClient]
private var mutableConfig = Option.empty[Config]
diff --git a/jms/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala b/jms/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala
index 6c6850835..7f686ae8e 100644
--- a/jms/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala
+++ b/jms/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala
@@ -35,7 +35,7 @@ import scala.util.{ Failure, Success }
class JmsBufferedAckConnectorsSpec extends JmsSharedServerSpec {
- override implicit val patienceConfig = PatienceConfig(2.minutes)
+ override implicit val patienceConfig: PatienceConfig = PatienceConfig(2.minutes)
"The JMS Ack Connectors" should {
"publish and consume strings through a queue" in withConnectionFactory() { connectionFactory =>
diff --git a/jms/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala b/jms/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala
index 47de052ce..ef58d3f67 100644
--- a/jms/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala
+++ b/jms/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala
@@ -26,7 +26,7 @@ import scala.concurrent.duration._
import scala.concurrent.Future
class JmsIbmmqConnectorsSpec extends JmsSpec {
- override implicit val patienceConfig = PatienceConfig(2.minutes)
+ override implicit val patienceConfig: PatienceConfig = PatienceConfig(2.minutes)
"The JMS Ibmmq Connectors" should {
val queueConnectionFactory = {
diff --git a/jms/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala b/jms/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala
index 68e3cf844..b07435160 100644
--- a/jms/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala
+++ b/jms/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala
@@ -35,7 +35,7 @@ class JmsTxConnectorsSpec extends JmsSharedServerSpec {
private final val log = LoggerFactory.getLogger(classOf[JmsTxConnectorsSpec])
- override implicit val patienceConfig = PatienceConfig(2.minutes)
+ override implicit val patienceConfig: PatienceConfig = PatienceConfig(2.minutes)
"The JMS Transactional Connectors" should {
"publish and consume strings through a queue" in withConnectionFactory() { connectionFactory =>
diff --git a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsSpec.scala b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsSpec.scala
index 421c1ba4b..4f955f5e1 100644
--- a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsSpec.scala
+++ b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsSpec.scala
@@ -37,7 +37,7 @@ abstract class JmsSpec
with MockitoSugar
with LogCapturing {
- implicit val system = ActorSystem(this.getClass.getSimpleName)
+ implicit val system: ActorSystem = ActorSystem(this.getClass.getSimpleName)
val consumerConfig = system.settings.config.getConfig(JmsConsumerSettings.configPath)
val producerConfig = system.settings.config.getConfig(JmsProducerSettings.configPath)
diff --git a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsAckConnectorsSpec.scala b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsAckConnectorsSpec.scala
index 1bd575a0f..2fa9a8da4 100644
--- a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsAckConnectorsSpec.scala
+++ b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsAckConnectorsSpec.scala
@@ -33,7 +33,7 @@ import scala.util.{ Failure, Success }
class JmsAckConnectorsSpec extends JmsSpec {
- override implicit val patienceConfig = PatienceConfig(2.minutes)
+ override implicit val patienceConfig: PatienceConfig = PatienceConfig(2.minutes)
"The JMS Ack Connectors" should {
"publish and consume strings through a queue" in withConnectionFactory() { connectionFactory =>
diff --git a/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/ShardIterator.scala b/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/ShardIterator.scala
index 13170a038..75de3c97c 100644
--- a/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/ShardIterator.scala
+++ b/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/ShardIterator.scala
@@ -40,7 +40,7 @@ object ShardIterator {
override final val shardIteratorType: ShardIteratorType = ShardIteratorType.TRIM_HORIZON
}
- case class AtTimestamp private (value: Instant) extends ShardIterator {
+ case class AtTimestamp private[kinesis] (value: Instant) extends ShardIterator {
override final val timestamp: Option[Instant] = Some(value)
override final val startingSequenceNumber: Option[String] = None
diff --git a/kinesis/src/test/scala/org/apache/pekko/stream/connectors/kinesis/KinesisSchedulerSourceSpec.scala b/kinesis/src/test/scala/org/apache/pekko/stream/connectors/kinesis/KinesisSchedulerSourceSpec.scala
index 6550771b0..3cdcbcd8c 100644
--- a/kinesis/src/test/scala/org/apache/pekko/stream/connectors/kinesis/KinesisSchedulerSourceSpec.scala
+++ b/kinesis/src/test/scala/org/apache/pekko/stream/connectors/kinesis/KinesisSchedulerSourceSpec.scala
@@ -269,7 +269,7 @@ class KinesisSchedulerSourceSpec
var recordProcessor: ShardRecordProcessor = _
var otherRecordProcessor: ShardRecordProcessor = _
- private val schedulerBuilder = { x: ShardRecordProcessorFactory =>
+ private val schedulerBuilder = { (x: ShardRecordProcessorFactory) =>
recordProcessor = x.shardRecordProcessor()
otherRecordProcessor = x.shardRecordProcessor()
semaphore.release()
@@ -334,12 +334,12 @@ class KinesisSchedulerSourceSpec
var latestRecord: KinesisClientRecord = _
val allRecordsPushed: Future[Unit] = Future {
for (i <- 1 to 3) {
- val record = org.mockito.Mockito.mock(classOf[KinesisClientRecord])
- when(record.sequenceNumber).thenReturn("1")
- when(record.subSequenceNumber).thenReturn(i.toLong)
+ val clientRecord = org.mockito.Mockito.mock(classOf[KinesisClientRecord])
+ when(clientRecord.sequenceNumber).thenReturn("1")
+ when(clientRecord.subSequenceNumber).thenReturn(i.toLong)
sourceProbe.sendNext(
new CommittableRecord(
- record,
+ clientRecord,
new BatchData(null, null, false, 0),
new ShardProcessorData(
"shard-1",
@@ -349,7 +349,7 @@ class KinesisSchedulerSourceSpec
override def forceCheckpoint(): Unit = checkpointer(record)
})
- latestRecord = record
+ latestRecord = clientRecord
}
}
@@ -374,11 +374,11 @@ class KinesisSchedulerSourceSpec
val allRecordsPushed: Future[Unit] = Future {
for (i <- 1 to 3) {
- val record = org.mockito.Mockito.mock(classOf[KinesisClientRecord])
- when(record.sequenceNumber).thenReturn(i.toString)
+ val clientRecord = org.mockito.Mockito.mock(classOf[KinesisClientRecord])
+ when(clientRecord.sequenceNumber).thenReturn(i.toString)
sourceProbe.sendNext(
new CommittableRecord(
- record,
+ clientRecord,
new BatchData(null, null, false, 0),
new ShardProcessorData(
"shard-1",
@@ -388,14 +388,14 @@ class KinesisSchedulerSourceSpec
override def forceCheckpoint(): Unit = checkpointerShard1(record)
})
- latestRecordShard1 = record
+ latestRecordShard1 = clientRecord
}
for (i <- 1 to 3) {
- val record = org.mockito.Mockito.mock(classOf[KinesisClientRecord])
- when(record.sequenceNumber).thenReturn(i.toString)
+ val clientRecord = org.mockito.Mockito.mock(classOf[KinesisClientRecord])
+ when(clientRecord.sequenceNumber).thenReturn(i.toString)
sourceProbe.sendNext(
new CommittableRecord(
- record,
+ clientRecord,
new BatchData(null, null, false, 0),
new ShardProcessorData(
"shard-2",
@@ -405,7 +405,7 @@ class KinesisSchedulerSourceSpec
override def forceCheckpoint(): Unit = checkpointerShard2(record)
})
- latestRecordShard2 = record
+ latestRecordShard2 = clientRecord
}
}
@@ -422,12 +422,12 @@ class KinesisSchedulerSourceSpec
}
"fail with Exception if checkpoint action fails" in new KinesisSchedulerCheckpointContext {
- val record: KinesisClientRecord = org.mockito.Mockito.mock(classOf[KinesisClientRecord])
- when(record.sequenceNumber).thenReturn("1")
+ val clientRecord: KinesisClientRecord = org.mockito.Mockito.mock(classOf[KinesisClientRecord])
+ when(clientRecord.sequenceNumber).thenReturn("1")
val checkpointer: KinesisClientRecord => Unit =
org.mockito.Mockito.mock(classOf[KinesisClientRecord => Unit])
val committableRecord: CommittableRecord = new CommittableRecord(
- record,
+ clientRecord,
new BatchData(null, null, false, 0),
new ShardProcessorData(
"shard-1",
@@ -439,7 +439,7 @@ class KinesisSchedulerSourceSpec
sourceProbe.sendNext(committableRecord)
val failure = new RuntimeException()
- when(checkpointer.apply(record)).thenThrow(failure)
+ when(checkpointer.apply(clientRecord)).thenThrow(failure)
sinkProbe.request(1)
diff --git a/mongodb/src/test/scala/docs/scaladsl/MongoSinkSpec.scala b/mongodb/src/test/scala/docs/scaladsl/MongoSinkSpec.scala
index 911dd2860..a0c1e8a45 100644
--- a/mongodb/src/test/scala/docs/scaladsl/MongoSinkSpec.scala
+++ b/mongodb/src/test/scala/docs/scaladsl/MongoSinkSpec.scala
@@ -52,11 +52,11 @@ class MongoSinkSpec
fromRegistries(fromProviders(classOf[Number], classOf[DomainObject]), DEFAULT_CODEC_REGISTRY): @nowarn(
"msg=match may not be exhaustive")
- implicit val system = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
+ implicit val defaultPatience: PatienceConfig =
+ PatienceConfig(timeout = 5.seconds, interval = 50.millis)
override protected def beforeAll(): Unit = {
- implicit val patienceConfig =
- PatienceConfig(timeout = 2.seconds, interval = 100.millis)
Source.fromPublisher(db.drop()).runWith(Sink.headOption).futureValue
}
@@ -69,9 +69,6 @@ class MongoSinkSpec
db.getCollection("domainObjectsSink", classOf[DomainObject]).withCodecRegistry(codecRegistry)
private val domainObjectsDocumentColl = db.getCollection("domainObjectsSink")
- implicit val defaultPatience =
- PatienceConfig(timeout = 5.seconds, interval = 50.millis)
-
override def afterEach(): Unit = {
Source.fromPublisher(numbersDocumentColl.deleteMany(new Document())).runWith(Sink.head).futureValue
Source.fromPublisher(domainObjectsDocumentColl.deleteMany(new Document())).runWith(Sink.head).futureValue
diff --git a/mongodb/src/test/scala/docs/scaladsl/MongoSourceSpec.scala b/mongodb/src/test/scala/docs/scaladsl/MongoSourceSpec.scala
index f1ac264d2..f4789be9d 100644
--- a/mongodb/src/test/scala/docs/scaladsl/MongoSourceSpec.scala
+++ b/mongodb/src/test/scala/docs/scaladsl/MongoSourceSpec.scala
@@ -43,7 +43,7 @@ class MongoSourceSpec
with LogCapturing {
// #init-system
- implicit val system = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
// #init-system
override protected def beforeAll(): Unit =
@@ -74,7 +74,7 @@ class MongoSourceSpec
private val numbersDocumentColl = db.getCollection("numbers")
- implicit val defaultPatience =
+ implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = 5.seconds, interval = 50.millis)
override def afterEach(): Unit =
diff --git a/orientdb/src/test/scala/docs/scaladsl/OrientDbSpec.scala b/orientdb/src/test/scala/docs/scaladsl/OrientDbSpec.scala
index a2fbe01fa..6dc76596f 100644
--- a/orientdb/src/test/scala/docs/scaladsl/OrientDbSpec.scala
+++ b/orientdb/src/test/scala/docs/scaladsl/OrientDbSpec.scala
@@ -159,7 +159,7 @@ class OrientDbSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with
// Copy source to sink1 through ODocument stream
val f1 = OrientDbSource(
sourceClass,
- OrientDbSourceSettings(oDatabase)).map { message: OrientDbReadResult[ODocument] =>
+ OrientDbSourceSettings(oDatabase)).map { (message: OrientDbReadResult[ODocument]) =>
OrientDbWriteMessage(message.oDocument)
}
.groupedWithin(10, 50.millis)
@@ -173,7 +173,7 @@ class OrientDbSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with
// #run-odocument
val result: Future[immutable.Seq[String]] = OrientDbSource(
sink4,
- OrientDbSourceSettings(oDatabase)).map { message: OrientDbReadResult[ODocument] =>
+ OrientDbSourceSettings(oDatabase)).map { (message: OrientDbReadResult[ODocument]) =>
message.oDocument.field[String]("book_title")
}
.runWith(Sink.seq)
@@ -197,7 +197,7 @@ class OrientDbSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with
val f1 = OrientDbSource(
sourceClass,
- OrientDbSourceSettings(oDatabase)).map { message: OrientDbReadResult[ODocument] =>
+ OrientDbSourceSettings(oDatabase)).map { (message: OrientDbReadResult[ODocument]) =>
OrientDbWriteMessage(message.oDocument)
}
.groupedWithin(10, 50.millis)
@@ -231,7 +231,7 @@ class OrientDbSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with
// #run-typed
val streamCompletion: Future[Done] = OrientDbSource
.typed(sourceClass, OrientDbSourceSettings(oDatabase), classOf[OrientDbTest.source1])
- .map { m: OrientDbReadResult[OrientDbTest.source1] =>
+ .map { (m: OrientDbReadResult[OrientDbTest.source1]) =>
val db: ODatabaseDocumentTx = oDatabase.acquire
db.setDatabaseOwner(new OObjectDatabaseTx(db))
ODatabaseRecordThreadLocal.instance.set(db)
@@ -267,7 +267,7 @@ class OrientDbSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with
committedOffsets = committedOffsets :+ offset
val f1 = Source(messagesFromKafka)
- .map { kafkaMessage: KafkaMessage =>
+ .map { (kafkaMessage: KafkaMessage) =>
val book = kafkaMessage.book
val id = book.title
println("title: " + book.title)
@@ -279,7 +279,7 @@ class OrientDbSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with
OrientDbFlow.createWithPassThrough(
sink7,
OrientDbWriteSettings(oDatabase)))
- .map { messages: Seq[OrientDbWriteMessage[ODocument, KafkaOffset]] =>
+ .map { (messages: Seq[OrientDbWriteMessage[ODocument, KafkaOffset]]) =>
messages.foreach { message =>
commitToKafka(message.passThrough)
}
diff --git a/pravega/src/test/scala/docs/scaladsl/PravegaReadWriteDocs.scala b/pravega/src/test/scala/docs/scaladsl/PravegaReadWriteDocs.scala
index 330c36420..f89232c62 100644
--- a/pravega/src/test/scala/docs/scaladsl/PravegaReadWriteDocs.scala
+++ b/pravega/src/test/scala/docs/scaladsl/PravegaReadWriteDocs.scala
@@ -18,6 +18,8 @@ import pekko.actor.ActorSystem
import pekko.stream.connectors.pravega.{
PravegaEvent,
ReaderSettingsBuilder,
+ TableReaderSettingsBuilder,
+ TableWriterSettings,
TableWriterSettingsBuilder,
WriterSettingsBuilder
}
@@ -27,21 +29,21 @@ import io.pravega.client.stream.Serializer
import io.pravega.client.stream.impl.UTF8StringSerializer
import java.nio.ByteBuffer
-import pekko.stream.connectors.pravega.TableReaderSettingsBuilder
import pekko.stream.connectors.pravega.scaladsl.PravegaTable
import pekko.stream.connectors.pravega.scaladsl.Pravega
+
import scala.util.Using
import io.pravega.client.tables.TableKey
class PravegaReadWriteDocs {
- implicit val system = ActorSystem("PravegaDocs")
+ implicit val system: ActorSystem = ActorSystem("PravegaDocs")
val serializer = new UTF8StringSerializer
implicit def personSerialiser: Serializer[Person] = ???
- implicit val intSerializer = new Serializer[Int] {
+ implicit val intSerializer: Serializer[Int] = new Serializer[Int] {
override def serialize(value: Int): ByteBuffer = {
val buff = ByteBuffer.allocate(4).putInt(value)
buff.position(0)
@@ -89,7 +91,7 @@ class PravegaReadWriteDocs {
Pravega
.source(readerGroup, readerSettings)
- .to(Sink.foreach { event: PravegaEvent[String] =>
+ .to(Sink.foreach { (event: PravegaEvent[String]) =>
val message: String = event.message
processMessage(message)
})
@@ -99,7 +101,7 @@ class PravegaReadWriteDocs {
}
- implicit val tablewriterSettings = TableWriterSettingsBuilder[Int, Person]()
+ implicit val tablewriterSettings: TableWriterSettings[Int, Person] = TableWriterSettingsBuilder[Int, Person]()
.withKeyExtractor(id => new TableKey(intSerializer.serialize(id)))
.build()
diff --git a/pravega/src/test/scala/docs/scaladsl/PravegaSettingsSpec.scala b/pravega/src/test/scala/docs/scaladsl/PravegaSettingsSpec.scala
index f374d276c..e763396e5 100644
--- a/pravega/src/test/scala/docs/scaladsl/PravegaSettingsSpec.scala
+++ b/pravega/src/test/scala/docs/scaladsl/PravegaSettingsSpec.scala
@@ -31,9 +31,9 @@ import io.pravega.client.tables.TableKey
class PravegaSettingsSpec extends PravegaBaseSpec with Matchers {
- implicit val serializer = new UTF8StringSerializer
+ implicit val serializer: UTF8StringSerializer = new UTF8StringSerializer
- implicit val intSerializer = new Serializer[Int] {
+ implicit val intSerializer: Serializer[Int] = new Serializer[Int] {
override def serialize(value: Int): ByteBuffer = {
val buff = ByteBuffer.allocate(4).putInt(value)
buff.position(0)
diff --git a/pravega/src/test/scala/docs/scaladsl/Serializers.scala b/pravega/src/test/scala/docs/scaladsl/Serializers.scala
index 6cc016190..62952b9e7 100644
--- a/pravega/src/test/scala/docs/scaladsl/Serializers.scala
+++ b/pravega/src/test/scala/docs/scaladsl/Serializers.scala
@@ -19,9 +19,9 @@ import io.pravega.client.stream.impl.UTF8StringSerializer
object Serializers {
- implicit val stringSerializer = new UTF8StringSerializer()
+ implicit val stringSerializer: UTF8StringSerializer = new UTF8StringSerializer()
- implicit val personSerializer = new Serializer[Person] {
+ implicit val personSerializer: Serializer[Person] = new Serializer[Person] {
def serialize(x: Person): ByteBuffer = {
val name = x.firstname.getBytes("UTF-8")
val buff = ByteBuffer.allocate(4 + name.length).putInt(x.id)
@@ -38,7 +38,7 @@ object Serializers {
}
- implicit val intSerializer = new Serializer[Int] {
+ implicit val intSerializer: Serializer[Int] = new Serializer[Int] {
override def serialize(value: Int): ByteBuffer = {
val buff = ByteBuffer.allocate(4).putInt(value)
buff.position(0)
diff --git a/project/Common.scala b/project/Common.scala
index c152452e9..8cc19f2ad 100644
--- a/project/Common.scala
+++ b/project/Common.scala
@@ -30,6 +30,8 @@ object Common extends AutoPlugin {
override def requires = JvmPlugin && HeaderPlugin && ApacheSonatypePlugin && DynVerPlugin
+ val isScala3 = Def.setting(scalaBinaryVersion.value == "3")
+
override def globalSettings = Seq(
scmInfo := Some(ScmInfo(url("https://github.com/apache/incubator-pekko-connectors"),
"git@github.com:apache/incubator-pekko-connectors.git")),
@@ -43,6 +45,12 @@ object Common extends AutoPlugin {
// Ignore unused keys which affect documentation
excludeLintKeys ++= Set(scmInfo, projectInfoVersion, autoAPIMappings))
+ val packagesToSkip = "org.apache.pekko.pattern:" + // for some reason Scaladoc creates this
+ "org.mongodb.scala:" + // this one is a mystery as well
+ // excluding generated grpc classes, except the model ones (com.google.pubsub)
+ "com.google.api:com.google.cloud:com.google.iam:com.google.logging:" +
+ "com.google.longrunning:com.google.protobuf:com.google.rpc:com.google.type"
+
override lazy val projectSettings = Dependencies.Common ++ Seq(
projectInfoVersion := (if (isSnapshot.value) "snapshot" else version.value),
crossVersion := CrossVersion.binary,
@@ -64,13 +72,8 @@ object Common extends AutoPlugin {
"-doc-version",
version.value,
"-sourcepath",
- (ThisBuild / baseDirectory).value.toString,
- "-skip-packages",
- "org.apache.pekko.pattern:" + // for some reason Scaladoc creates this
- "org.mongodb.scala:" + // this one is a mystery as well
- // excluding generated grpc classes, except the model ones (com.google.pubsub)
- "com.google.api:com.google.cloud:com.google.iam:com.google.logging:" +
- "com.google.longrunning:com.google.protobuf:com.google.rpc:com.google.type"),
+ (ThisBuild / baseDirectory).value.toString),
+ Compile / doc / scalacOptions := scalacOptions.value,
Compile / doc / scalacOptions ++=
Seq(
"-doc-source-url", {
@@ -79,6 +82,13 @@ object Common extends AutoPlugin {
},
"-doc-canonical-base-url",
"https://pekko.apache.org/api/pekko-connectors/current/"),
+ Compile / doc / scalacOptions ++= {
+ if (isScala3.value) {
+ Seq("-skip-packages:" + packagesToSkip)
+ } else {
+ Seq("-skip-packages", packagesToSkip)
+ }
+ },
Compile / doc / scalacOptions -= "-Werror",
compile / javacOptions ++= Seq(
"-Xlint:cast",
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index cf4500882..f8d605989 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -16,7 +16,8 @@ object Dependencies {
val Scala213 = "2.13.10" // update even in link-validator.conf
val Scala212 = "2.12.17"
- val ScalaVersions = Seq(Scala213, Scala212)
+ val Scala3 = "3.3.0"
+ val ScalaVersions = Seq(Scala213, Scala212, Scala3)
val PekkoVersion = "0.0.0+26669-ec5b6764-SNAPSHOT"
val PekkoBinaryVersion = "current"
@@ -62,7 +63,7 @@ object Dependencies {
val testkit = Seq(
libraryDependencies := Seq(
- "org.scala-lang.modules" %% "scala-collection-compat" % "2.2.0",
+ "org.scala-lang.modules" %% "scala-collection-compat" % "2.10.0",
"org.apache.pekko" %% "pekko-stream" % PekkoVersion,
"org.apache.pekko" %% "pekko-stream-testkit" % PekkoVersion,
"org.apache.pekko" %% "pekko-slf4j" % PekkoVersion,
@@ -88,6 +89,7 @@ object Dependencies {
"com.fasterxml.jackson.core" % "jackson-databind" % JacksonDatabindVersion)
val Amqp = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"com.rabbitmq" % "amqp-client" % "5.14.2" // APLv2
) ++ Mockito)
@@ -104,6 +106,7 @@ object Dependencies {
ExclusionRule("software.amazon.awssdk", "netty-nio-client"))) ++ Mockito)
val AzureStorageQueue = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"com.microsoft.azure" % "azure-storage" % "8.0.0" // ApacheV2
))
@@ -113,6 +116,7 @@ object Dependencies {
val CassandraDriverVersionInDocs = "4.15"
val Cassandra = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
("com.datastax.oss" % "java-driver-core" % CassandraDriverVersion)
.exclude("com.github.spotbugs", "spotbugs-annotations")
@@ -121,6 +125,7 @@ object Dependencies {
"org.apache.pekko" %% "pekko-discovery" % PekkoVersion % Provided))
val Couchbase = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"com.couchbase.client" % "java-client" % CouchbaseVersion, // ApacheV2
"io.reactivex" % "rxjava-reactive-streams" % "1.2.1", // ApacheV2
@@ -151,6 +156,7 @@ object Dependencies {
))
val Elasticsearch = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion,
@@ -162,17 +168,19 @@ object Dependencies {
))
val AvroParquet = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.parquet" % "parquet-avro" % "1.10.1", // Apache2
("org.apache.hadoop" % "hadoop-client" % "3.2.1" % Test).exclude("log4j", "log4j"), // Apache2
("org.apache.hadoop" % "hadoop-common" % "3.2.1" % Test).exclude("log4j", "log4j"), // Apache2
"com.sksamuel.avro4s" %% "avro4s-core" % "4.1.1" % Test,
"org.scalacheck" %% "scalacheck" % scalaCheckVersion % Test,
- "org.specs2" %% "specs2-core" % "4.8.3" % Test, // MIT like: https://github.com/etorreborre/specs2/blob/master/LICENSE.txt
+ "org.specs2" %% "specs2-core" % "4.20.0" % Test, // MIT like: https://github.com/etorreborre/specs2/blob/master/LICENSE.txt
"org.slf4j" % "log4j-over-slf4j" % log4jOverSlf4jVersion % Test // MIT like: http://www.slf4j.org/license.html
))
val Ftp = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"commons-net" % "commons-net" % "3.8.0", // ApacheV2
"com.hierynomus" % "sshj" % "0.33.0" // ApacheV2
@@ -182,6 +190,7 @@ object Dependencies {
val GeodeVersionForDocs = "115"
val Geode = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++=
Seq("geode-core", "geode-cq")
.map("org.apache.geode" % _ % GeodeVersion) ++
@@ -190,6 +199,7 @@ object Dependencies {
"org.apache.logging.log4j" % "log4j-to-slf4j" % "2.17.1" % Test) ++ JacksonDatabindDependencies)
val GoogleCommon = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion,
@@ -199,6 +209,7 @@ object Dependencies {
) ++ Mockito)
val GoogleBigQuery = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-jackson" % PekkoHttpVersion % Provided,
@@ -209,6 +220,7 @@ object Dependencies {
"io.specto" % "hoverfly-java" % hoverflyVersion % Test // ApacheV2
) ++ Mockito)
val GoogleBigQueryStorage = Seq(
+ crossScalaVersions -= Scala3,
// see Pekko gRPC version in plugins.sbt
libraryDependencies ++= Seq(
// https://github.com/googleapis/java-bigquerystorage/tree/master/proto-google-cloud-bigquerystorage-v1
@@ -224,6 +236,7 @@ object Dependencies {
"org.apache.pekko" %% "pekko-discovery" % PekkoVersion) ++ Mockito)
val GooglePubSub = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion,
@@ -231,6 +244,7 @@ object Dependencies {
) ++ Mockito)
val GooglePubSubGrpc = Seq(
+ crossScalaVersions -= Scala3,
// see Pekko gRPC version in plugins.sbt
libraryDependencies ++= Seq(
// https://github.com/googleapis/java-pubsub/tree/master/proto-google-cloud-pubsub-v1/
@@ -241,11 +255,13 @@ object Dependencies {
"org.apache.pekko" %% "pekko-discovery" % PekkoVersion))
val GoogleFcm = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion) ++ Mockito)
val GoogleStorage = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion,
@@ -271,6 +287,7 @@ object Dependencies {
val HadoopVersion = "3.2.1"
val Hdfs = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
("org.apache.hadoop" % "hadoop-client" % HadoopVersion).exclude("log4j", "log4j").exclude("org.slf4j",
"slf4j-log4j12"), // ApacheV2
@@ -285,6 +302,7 @@ object Dependencies {
))
val HuaweiPushKit = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion,
@@ -297,6 +315,7 @@ object Dependencies {
))
val IronMq = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.mdedetrich" %% "pekko-stream-circe" % "0.0.0+94-dbf3173f-SNAPSHOT", // ApacheV2
@@ -304,6 +323,7 @@ object Dependencies {
))
val Jms = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"javax.jms" % "jms" % "1.1" % Provided, // CDDL + GPLv2
"com.ibm.mq" % "com.ibm.mq.allclient" % "9.2.5.0" % Test, // IBM International Program License Agreement https://public.dhe.ibm.com/ibmdl/export/pub/software/websphere/messaging/mqdev/maven/licenses/L-APIG-AZYF2E/LI_en.html
@@ -316,6 +336,7 @@ object Dependencies {
"https://repository.jboss.org/nexus/content/groups/public")) +: externalResolvers.value)
val JsonStreaming = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"com.github.jsurfer" % "jsurfer-jackson" % "1.6.0" // MIT
) ++ JacksonDatabindDependencies)
@@ -341,6 +362,7 @@ object Dependencies {
))
val MongoDb = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.mongodb.scala" %% "mongo-scala-driver" % "4.4.0" // ApacheV2
))
@@ -351,6 +373,7 @@ object Dependencies {
))
val MqttStreaming = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-actor-typed" % PekkoVersion, // ApacheV2
"org.apache.pekko" %% "pekko-actor-testkit-typed" % PekkoVersion % Test, // ApacheV2
@@ -382,6 +405,7 @@ object Dependencies {
))
val S3 = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-xml" % PekkoHttpVersion,
@@ -408,6 +432,7 @@ object Dependencies {
val SlickVersion = "3.3.3"
val Slick = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"com.typesafe.slick" %% "slick" % SlickVersion, // BSD 2-clause "Simplified" License
"com.typesafe.slick" %% "slick-hikaricp" % SlickVersion, // BSD 2-clause "Simplified" License
@@ -468,6 +493,7 @@ object Dependencies {
"org.apache.pekko" %% "pekko-http-testkit" % PekkoHttpVersion % Test))
val UnixDomainSocket = Seq(
+ crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"com.github.jnr" % "jffi" % "1.3.1", // classifier "complete", // Is the classifier needed anymore?
"com.github.jnr" % "jnr-unixsocket" % "0.38.5" // BSD/ApacheV2/CPL/MIT as per https://github.com/akka/alpakka/issues/620#issuecomment-348727265
diff --git a/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/Resource.scala b/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/Resource.scala
index fe7d278ac..16660013e 100644
--- a/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/Resource.scala
+++ b/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/Resource.scala
@@ -116,7 +116,7 @@ object ResourceSettings {
* there is only one instance of the resource instantiated per Actor System.
*/
final class ResourceExt private (sys: ExtendedActorSystem) extends Extension {
- implicit val resource = Resource(ResourceSettings()(sys))
+ implicit val resource: Resource = Resource(ResourceSettings()(sys))
sys.registerOnTermination(resource.cleanup())
}
diff --git a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/HttpRequestsSpec.scala b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/HttpRequestsSpec.scala
index 6c30070af..3877e1da8 100644
--- a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/HttpRequestsSpec.scala
+++ b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/HttpRequestsSpec.scala
@@ -14,7 +14,6 @@
package org.apache.pekko.stream.connectors.s3.impl
import java.util.UUID
-
import org.apache.pekko
import pekko.actor.ActorSystem
import pekko.http.scaladsl.Http
@@ -33,6 +32,8 @@ import software.amazon.awssdk.auth.credentials._
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.regions.providers._
+import scala.concurrent.ExecutionContext
+
class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with IntegrationPatience with LogCapturing {
// test fixtures
@@ -55,7 +56,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
val multipartUpload = MultipartUpload("test-bucket", "testKey", "uploadId")
it should "initiate multipart upload when the region is us-east-1" in {
- implicit val settings = getSettings()
+ implicit val settings: S3Settings = getSettings()
val req =
HttpRequests.initiateMultipartUploadRequest(
@@ -74,7 +75,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "initiate multipart upload with other regions" in {
- implicit val settings = getSettings(s3Region = Region.US_EAST_2)
+ implicit val settings: S3Settings = getSettings(s3Region = Region.US_EAST_2)
val req =
HttpRequests.initiateMultipartUploadRequest(
@@ -93,21 +94,22 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "throw an error if path-style access is false and the bucket name contains non-LDH characters" in {
- implicit val settings = getSettings(s3Region = Region.EU_WEST_1)
+ implicit val settings: S3Settings = getSettings(s3Region = Region.EU_WEST_1)
assertThrows[IllegalUriException](
HttpRequests.getDownloadRequest(S3Location("invalid_bucket_name", "image-1024@2x")))
}
it should "throw an error if the key uses `..`" in {
- implicit val settings = getSettings(s3Region = Region.EU_WEST_1)
+ implicit val settings: S3Settings = getSettings(s3Region = Region.EU_WEST_1)
assertThrows[IllegalUriException](
HttpRequests.getDownloadRequest(S3Location("validbucket", "../other-bucket/image-1024@2x")))
}
it should "throw an error when using `..` with path-style access" in {
- implicit val settings = getSettings(s3Region = Region.EU_WEST_1).withAccessStyle(AccessStyle.PathAccessStyle)
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.EU_WEST_1).withAccessStyle(AccessStyle.PathAccessStyle)
assertThrows[IllegalUriException](
HttpRequests.getDownloadRequest(S3Location("invalid/../bucket_name", "image-1024@2x")))
@@ -120,7 +122,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "initiate multipart upload with path-style access in region us-east-1" in {
- implicit val settings = getSettings(s3Region = Region.US_EAST_1).withAccessStyle(AccessStyle.PathAccessStyle)
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.US_EAST_1).withAccessStyle(AccessStyle.PathAccessStyle)
val req =
HttpRequests.initiateMultipartUploadRequest(
@@ -133,7 +136,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "support download requests with path-style access in region us-east-1" in {
- implicit val settings = getSettings(s3Region = Region.US_EAST_1).withAccessStyle(AccessStyle.PathAccessStyle)
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.US_EAST_1).withAccessStyle(AccessStyle.PathAccessStyle)
val req = HttpRequests.getDownloadRequest(location)
@@ -143,7 +147,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "initiate multipart upload with path-style access in other regions" in {
- implicit val settings = getSettings(s3Region = Region.US_WEST_2).withAccessStyle(AccessStyle.PathAccessStyle)
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.US_WEST_2).withAccessStyle(AccessStyle.PathAccessStyle)
val req =
HttpRequests.initiateMultipartUploadRequest(
@@ -156,7 +161,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "support download requests with path-style access in other regions" in {
- implicit val settings = getSettings(s3Region = Region.EU_WEST_1).withAccessStyle(AccessStyle.PathAccessStyle)
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.EU_WEST_1).withAccessStyle(AccessStyle.PathAccessStyle)
val req = HttpRequests.getDownloadRequest(location)
@@ -166,7 +172,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "support download requests via configured `endpointUrl`" in {
- implicit val settings = getSettings(s3Region = Region.EU_WEST_1).withEndpointUrl("http://localhost:8080")
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.EU_WEST_1).withEndpointUrl("http://localhost:8080")
val req = HttpRequests.getDownloadRequest(location)
@@ -180,7 +187,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
// into an object at path /[empty string]/...
// added this test because of a tricky uri building issue
// in case of pathStyleAccess = false
- implicit val settings = getSettings()
+ implicit val settings: S3Settings = getSettings()
val location = S3Location("bucket", "/test/foo.txt")
@@ -193,7 +200,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
it should "support download requests with keys ending with /" in {
// object with a slash at the end of the filename should be accessible
- implicit val settings = getSettings()
+ implicit val settings: S3Settings = getSettings()
val location = S3Location("bucket", "/test//")
@@ -205,7 +212,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "support download requests with keys containing spaces" in {
- implicit val settings = getSettings()
+ implicit val settings: S3Settings = getSettings()
val location = S3Location("bucket", "test folder/test file.txt")
@@ -217,7 +224,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "support download requests with keys containing plus" in {
- implicit val settings = getSettings()
+ implicit val settings: S3Settings = getSettings()
val location = S3Location("bucket", "test folder/1 + 2 = 3")
val req = HttpRequests.getDownloadRequest(location)
@@ -227,7 +234,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "support download requests with keys containing spaces with path-style access in other regions" in {
- implicit val settings = getSettings(s3Region = Region.EU_WEST_1).withAccessStyle(AccessStyle.PathAccessStyle)
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.EU_WEST_1).withAccessStyle(AccessStyle.PathAccessStyle)
val location = S3Location("bucket", "test folder/test file.txt")
@@ -239,7 +247,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "add versionId query parameter when provided" in {
- implicit val settings = getSettings().withAccessStyle(AccessStyle.PathAccessStyle)
+ implicit val settings: S3Settings = getSettings().withAccessStyle(AccessStyle.PathAccessStyle)
val location = S3Location("bucket", "test/foo.txt")
val versionId = "123456"
@@ -253,7 +261,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "support multipart init upload requests via configured `endpointUrl`" in {
- implicit val settings = getSettings(s3Region = Region.EU_WEST_1).withEndpointUrl("http://localhost:8080")
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.EU_WEST_1).withEndpointUrl("http://localhost:8080")
val req =
HttpRequests.initiateMultipartUploadRequest(
@@ -267,7 +276,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "support multipart upload part requests via configured `endpointUrl`" in {
- implicit val settings = getSettings(s3Region = Region.EU_WEST_1).withEndpointUrl("http://localhost:8080")
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.EU_WEST_1).withEndpointUrl("http://localhost:8080")
val req =
HttpRequests.uploadPartRequest(multipartUpload, 1, MemoryChunk(ByteString.empty))
@@ -278,7 +288,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "properly multipart upload part request with customer keys server side encryption" in {
- implicit val settings = getSettings(s3Region = Region.EU_WEST_1).withAccessStyle(AccessStyle.PathAccessStyle)
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.EU_WEST_1).withAccessStyle(AccessStyle.PathAccessStyle)
val myKey = "my-key"
val md5Key = "md5-key"
val s3Headers = ServerSideEncryption.customerKeys(myKey).withMd5(md5Key).headersFor(UploadPart)
@@ -290,8 +301,9 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "support multipart upload complete requests via configured `endpointUrl`" in {
- implicit val settings = getSettings(s3Region = Region.EU_WEST_1).withEndpointUrl("http://localhost:8080")
- implicit val executionContext = scala.concurrent.ExecutionContext.global
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.EU_WEST_1).withEndpointUrl("http://localhost:8080")
+ implicit val executionContext: ExecutionContext = ExecutionContext.global
val req =
HttpRequests.completeMultipartUploadRequest(multipartUpload, (1, "part") :: Nil, Nil).futureValue
@@ -302,7 +314,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "initiate multipart upload with AES-256 server side encryption" in {
- implicit val settings = getSettings(s3Region = Region.US_EAST_2)
+ implicit val settings: S3Settings = getSettings(s3Region = Region.US_EAST_2)
val s3Headers = ServerSideEncryption.aes256().headersFor(InitiateMultipartUpload)
val req = HttpRequests.initiateMultipartUploadRequest(location, contentType, s3Headers)
@@ -310,7 +322,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "initiate multipart upload with aws:kms server side encryption" in {
- implicit val settings = getSettings(s3Region = Region.US_EAST_2)
+ implicit val settings: S3Settings = getSettings(s3Region = Region.US_EAST_2)
val testArn = "arn:aws:kms:my-region:my-account-id:key/my-key-id"
val s3Headers = ServerSideEncryption.kms(testArn).headersFor(InitiateMultipartUpload)
val req = HttpRequests.initiateMultipartUploadRequest(location, contentType, s3Headers)
@@ -320,7 +332,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "initiate multipart upload with customer keys encryption" in {
- implicit val settings = getSettings(s3Region = Region.US_EAST_2)
+ implicit val settings: S3Settings = getSettings(s3Region = Region.US_EAST_2)
val myKey = "my-key"
val md5Key = "md5-key"
val s3Headers = ServerSideEncryption.customerKeys(myKey).withMd5(md5Key).headersFor(InitiateMultipartUpload)
@@ -332,7 +344,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "initiate multipart upload with custom s3 storage class" in {
- implicit val settings = getSettings(s3Region = Region.US_EAST_2)
+ implicit val settings: S3Settings = getSettings(s3Region = Region.US_EAST_2)
val s3Headers = S3Headers().withStorageClass(StorageClass.ReducedRedundancy).headers
val req = HttpRequests.initiateMultipartUploadRequest(location, contentType, s3Headers)
@@ -340,7 +352,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "initiate multipart upload with custom s3 headers" in {
- implicit val settings = getSettings(s3Region = Region.US_EAST_2)
+ implicit val settings: S3Settings = getSettings(s3Region = Region.US_EAST_2)
val s3Headers = S3Headers().withCustomHeaders(Map("Cache-Control" -> "no-cache")).headers
val req = HttpRequests.initiateMultipartUploadRequest(location, contentType, s3Headers)
@@ -348,7 +360,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "properly construct the list bucket request with no prefix, continuation token or delimiter passed" in {
- implicit val settings = getSettings(s3Region = Region.US_EAST_2).withAccessStyle(AccessStyle.PathAccessStyle)
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.US_EAST_2).withAccessStyle(AccessStyle.PathAccessStyle)
val req =
HttpRequests.listBucket(location.bucket)
@@ -357,7 +370,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "properly construct the list bucket request with a prefix and token passed" in {
- implicit val settings = getSettings(s3Region = Region.US_EAST_2).withAccessStyle(AccessStyle.PathAccessStyle)
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.US_EAST_2).withAccessStyle(AccessStyle.PathAccessStyle)
val req =
HttpRequests.listBucket(location.bucket, Some("random/prefix"), Some("randomToken"))
@@ -368,7 +382,8 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "properly construct the list bucket request with a delimiter and token passed" in {
- implicit val settings = getSettings(s3Region = Region.US_EAST_2).withAccessStyle(AccessStyle.PathAccessStyle)
+ implicit val settings: S3Settings =
+ getSettings(s3Region = Region.US_EAST_2).withAccessStyle(AccessStyle.PathAccessStyle)
val req =
HttpRequests.listBucket(location.bucket, delimiter = Some("/"), continuationToken = Some("randomToken"))
@@ -377,7 +392,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "properly construct the list bucket request when using api version 1" in {
- implicit val settings =
+ implicit val settings: S3Settings =
getSettings(s3Region = Region.US_EAST_2, listBucketApiVersion = ApiVersion.ListBucketVersion1)
.withAccessStyle(AccessStyle.PathAccessStyle)
@@ -388,7 +403,7 @@ class HttpRequestsSpec extends AnyFlatSpec with Matchers with ScalaFutures with
}
it should "properly construct the list bucket request when using api version set to 1 and a continuation token" in {
- implicit val settings =
+ implicit val settings: S3Settings =
getSettings(s3Region = Region.US_EAST_2, listBucketApiVersion = ApiVersion.ListBucketVersion1)
.withAccessStyle(AccessStyle.PathAccessStyle)
diff --git a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/S3StreamSpec.scala b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/S3StreamSpec.scala
index 9e4415dfd..f7443656b 100644
--- a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/S3StreamSpec.scala
+++ b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/S3StreamSpec.scala
@@ -62,7 +62,7 @@ class S3StreamSpec(_system: ActorSystem)
}
val location = S3Location("test-bucket", "test-key")
- implicit val settings =
+ implicit val settings: S3Settings =
S3Settings(MemoryBufferType, credentialsProvider, regionProvider, ApiVersion.ListBucketVersion2)
val result: HttpRequest = S3Stream.invokePrivate(requestHeaders(getDownloadRequest(location), None))
@@ -86,7 +86,7 @@ class S3StreamSpec(_system: ActorSystem)
val location = S3Location("test-bucket", "test-key")
val range = ByteRange(1, 4)
- implicit val settings =
+ implicit val settings: S3Settings =
S3Settings(MemoryBufferType, credentialsProvider, regionProvider, ApiVersion.ListBucketVersion2)
val result: HttpRequest = S3Stream.invokePrivate(requestHeaders(getDownloadRequest(location), Some(range)))
diff --git a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/SignerSpec.scala b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/SignerSpec.scala
index 85adbdc36..4f16205d7 100644
--- a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/SignerSpec.scala
+++ b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/SignerSpec.scala
@@ -41,7 +41,7 @@ class SignerSpec(_system: ActorSystem)
with LogCapturing {
def this() = this(ActorSystem("SignerSpec"))
- implicit val defaultPatience =
+ implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = Span(2, Seconds), interval = Span(5, Millis))
val credentials = StaticCredentialsProvider.create(
diff --git a/slick/src/test/scala/docs/scaladsl/DocSnippets.scala b/slick/src/test/scala/docs/scaladsl/DocSnippets.scala
index b2a92cadf..265759d19 100644
--- a/slick/src/test/scala/docs/scaladsl/DocSnippets.scala
+++ b/slick/src/test/scala/docs/scaladsl/DocSnippets.scala
@@ -16,6 +16,8 @@ package docs.scaladsl
import org.apache.pekko
import pekko.Done
import pekko.actor.ActorSystem
+
+import scala.concurrent.ExecutionContext
//#important-imports
import org.apache.pekko
import pekko.stream.connectors.slick.scaladsl._
@@ -26,8 +28,8 @@ import slick.jdbc.GetResult
import scala.concurrent.Future
object SlickSourceWithPlainSQLQueryExample extends App {
- implicit val system = ActorSystem()
- implicit val ec = system.dispatcher
+ implicit val system: ActorSystem = ActorSystem()
+ implicit val ec: ExecutionContext = system.dispatcher
// #source-example
implicit val session = SlickSession.forConfig("slick-h2")
@@ -62,8 +64,8 @@ object SlickSourceWithPlainSQLQueryExample extends App {
}
object SlickSourceWithTypedQueryExample extends App {
- implicit val system = ActorSystem()
- implicit val ec = system.dispatcher
+ implicit val system: ActorSystem = ActorSystem()
+ implicit val ec: ExecutionContext = system.dispatcher
// #source-with-typed-query
implicit val session = SlickSession.forConfig("slick-h2")
@@ -94,8 +96,8 @@ object SlickSourceWithTypedQueryExample extends App {
}
object SlickSinkExample extends App {
- implicit val system = ActorSystem()
- implicit val ec = system.dispatcher
+ implicit val system: ActorSystem = ActorSystem()
+ implicit val ec: ExecutionContext = system.dispatcher
// #sink-example
implicit val session = SlickSession.forConfig("slick-h2")
@@ -125,8 +127,8 @@ object SlickSinkExample extends App {
}
object SlickFlowExample extends App {
- implicit val system = ActorSystem()
- implicit val ec = system.dispatcher
+ implicit val system: ActorSystem = ActorSystem()
+ implicit val ec: ExecutionContext = system.dispatcher
// #flow-example
implicit val session = SlickSession.forConfig("slick-h2")
@@ -168,8 +170,8 @@ object SlickFlowWithPassThroughExample extends App {
def map[B](f: A => B): KafkaMessage[B] = KafkaMessage(f(msg), offset)
}
- implicit val system = ActorSystem()
- implicit val ec = system.dispatcher
+ implicit val system: ActorSystem = ActorSystem()
+ implicit val ec: ExecutionContext = system.dispatcher
// #flowWithPassThrough-example
implicit val session = SlickSession.forConfig("slick-h2")
diff --git a/slick/src/test/scala/docs/scaladsl/SlickSpec.scala b/slick/src/test/scala/docs/scaladsl/SlickSpec.scala
index db1e4418b..f27879995 100644
--- a/slick/src/test/scala/docs/scaladsl/SlickSpec.scala
+++ b/slick/src/test/scala/docs/scaladsl/SlickSpec.scala
@@ -27,7 +27,7 @@ import slick.dbio.DBIOAction
import slick.jdbc.{ GetResult, JdbcProfile }
import scala.concurrent.duration._
-import scala.concurrent.{ Await, Future }
+import scala.concurrent.{ Await, ExecutionContext, Future }
import org.scalatest.matchers.must.Matchers
import org.scalatest.wordspec.AnyWordSpec
@@ -43,7 +43,7 @@ class SlickSpec
with Matchers
with LogCapturing {
// #init-mat
- implicit val system = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
// #init-mat
// #init-session
@@ -59,9 +59,9 @@ class SlickSpec
def * = (id, name)
}
- implicit val ec = system.dispatcher
- implicit val defaultPatience = PatienceConfig(timeout = 3.seconds, interval = 50.millis)
- implicit val getUserResult = GetResult(r => User(r.nextInt(), r.nextString()))
+ implicit val ec: ExecutionContext = system.dispatcher
+ implicit val defaultPatience: PatienceConfig = PatienceConfig(timeout = 3.seconds, interval = 50.millis)
+ implicit val getUserResult: GetResult[User] = GetResult(r => User(r.nextInt(), r.nextString()))
val users = (1 to 40).map(i => User(i, s"Name$i")).toSet
diff --git a/sns/src/test/scala/docs/scaladsl/SnsPublisherSpec.scala b/sns/src/test/scala/docs/scaladsl/SnsPublisherSpec.scala
index e55afcc86..db1c041c5 100644
--- a/sns/src/test/scala/docs/scaladsl/SnsPublisherSpec.scala
+++ b/sns/src/test/scala/docs/scaladsl/SnsPublisherSpec.scala
@@ -34,7 +34,7 @@ class SnsPublisherSpec
with IntegrationTestContext
with LogCapturing {
- implicit val defaultPatience =
+ implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = 15.seconds, interval = 100.millis)
"SNS Publisher sink" should "send string message" in {
diff --git a/solr/src/test/scala/docs/scaladsl/SolrSpec.scala b/solr/src/test/scala/docs/scaladsl/SolrSpec.scala
index 9f2f5e3ef..64c7645c3 100644
--- a/solr/src/test/scala/docs/scaladsl/SolrSpec.scala
+++ b/solr/src/test/scala/docs/scaladsl/SolrSpec.scala
@@ -90,7 +90,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
// #run-document
val copyCollection = SolrSource
.fromTupleStream(stream)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val book: Book = tupleToBook(tuple)
val doc: SolrInputDocument = bookToDoc(book)
WriteMessage.createUpsertMessage(doc)
@@ -138,7 +138,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
// #run-bean
val copyCollection = SolrSource
.fromTupleStream(stream)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val title = tuple.getString("title")
WriteMessage.createUpsertMessage(BookBean(title))
}
@@ -179,7 +179,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
// #run-typed
val copyCollection = SolrSource
.fromTupleStream(stream)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val book: Book = tupleToBook(tuple)
WriteMessage.createUpsertMessage(book)
}
@@ -223,7 +223,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
// #typeds-flow
val copyCollection = SolrSource
.fromTupleStream(stream)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val book: Book = tupleToBook(tuple)
WriteMessage.createUpsertMessage(book)
}
@@ -294,7 +294,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
// #kafka-example
// Note: This code mimics Pekko Connectors Kafka APIs
val copyCollection = kafkaConsumerSource
- .map { kafkaMessage: CommittableMessage =>
+ .map { (kafkaMessage: CommittableMessage) =>
val book = kafkaMessage.book
// Transform message so that we can write to solr
WriteMessage.createUpsertMessage(book).withPassThrough(kafkaMessage.committableOffset)
@@ -340,7 +340,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
val copyCollection = SolrSource
.fromTupleStream(stream)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val book: Book = tupleToBook(tuple)
val doc: SolrInputDocument = bookToDoc(book)
WriteMessage.createUpsertMessage(doc)
@@ -360,7 +360,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
// #delete-documents
val deleteDocuments = SolrSource
.fromTupleStream(stream2)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val id = tuple.fields.get("title").toString
WriteMessage.createDeleteMessage[SolrInputDocument](id)
}
@@ -393,7 +393,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
val upsertCollection = SolrSource
.fromTupleStream(stream)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val book: Book = tupleToBook(tuple)
.copy(comment = "Written by good authors.")
val doc: SolrInputDocument = bookToDoc(book)
@@ -414,7 +414,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
// #update-atomically-documents
val updateCollection = SolrSource
.fromTupleStream(stream2)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val id = tuple.fields.get("title").toString
val comment = tuple.fields.get("comment").toString
WriteMessage.createUpdateMessage[SolrInputDocument](
@@ -461,7 +461,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
val copyCollection = SolrSource
.fromTupleStream(stream)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val book: Book = tupleToBook(tuple)
WriteMessage.createUpsertMessage(book)
}
@@ -482,7 +482,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
val deleteElements = SolrSource
.fromTupleStream(stream2)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val title = tuple.fields.get("title").toString
WriteMessage.createDeleteMessage[Book](title)
}
@@ -512,7 +512,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
val copyCollection = SolrSource
.fromTupleStream(stream)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val book: Book =
tupleToBook(tuple).copy(comment = "Written by good authors.", routerOpt = Some("router-value"))
WriteMessage.createUpsertMessage(book)
@@ -534,7 +534,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
val updateCollection = SolrSource
.fromTupleStream(stream2)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
WriteMessage
.createUpdateMessage[Book](
idField = "title",
@@ -581,7 +581,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
val copyCollection = SolrSource
.fromTupleStream(stream)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val book: Book = tupleToBook(tuple)
val doc: SolrInputDocument = bookToDoc(book)
WriteMessage.createUpsertMessage(doc)
@@ -601,7 +601,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
// #delete-documents-query
val deleteByQuery = SolrSource
.fromTupleStream(stream2)
- .map { tuple: Tuple =>
+ .map { (tuple: Tuple) =>
val title = tuple.fields.get("title").toString
WriteMessage.createDeleteByQueryMessage[SolrInputDocument](
s"""title:"$title" """)
@@ -657,7 +657,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
// #kafka-example-PT
// Note: This code mimics Pekko Connectors Kafka APIs
val copyCollection = kafkaConsumerSource
- .map { offset: CommittableOffset =>
+ .map { (offset: CommittableOffset) =>
// Transform message so that we can write to solr
WriteMessage.createPassThrough(offset).withSource(new SolrInputDocument())
}
diff --git a/sqs/src/test/scala/docs/scaladsl/SqsAckSpec.scala b/sqs/src/test/scala/docs/scaladsl/SqsAckSpec.scala
index b8fe72edf..9f7c1c406 100644
--- a/sqs/src/test/scala/docs/scaladsl/SqsAckSpec.scala
+++ b/sqs/src/test/scala/docs/scaladsl/SqsAckSpec.scala
@@ -248,7 +248,7 @@ class SqsAckSpec extends AnyFlatSpec with Matchers with DefaultTestContext with
it should "fail if any of the messages in the batch request failed" in {
val messages = for (i <- 0 until 10) yield Message.builder().body(s"Message - $i").build()
- implicit val mockAwsSqsClient = mock[SqsAsyncClient]
+ implicit val mockAwsSqsClient: SqsAsyncClient = mock[SqsAsyncClient]
when(mockAwsSqsClient.deleteMessageBatch(any[DeleteMessageBatchRequest]))
.thenReturn(CompletableFuture.completedFuture {
@@ -269,7 +269,7 @@ class SqsAckSpec extends AnyFlatSpec with Matchers with DefaultTestContext with
it should "fail if the batch request failed" in {
val messages = for (i <- 0 until 10) yield Message.builder().body(s"Message - $i").build()
- implicit val mockAwsSqsClient = mock[SqsAsyncClient]
+ implicit val mockAwsSqsClient: SqsAsyncClient = mock[SqsAsyncClient]
when(mockAwsSqsClient.deleteMessageBatch(any[DeleteMessageBatchRequest]))
.thenReturn(
@@ -289,7 +289,7 @@ class SqsAckSpec extends AnyFlatSpec with Matchers with DefaultTestContext with
it should "fail if the client invocation failed" in {
val messages = for (i <- 0 until 10) yield Message.builder().body(s"Message - $i").build()
- implicit val mockAwsSqsClient = mock[SqsAsyncClient]
+ implicit val mockAwsSqsClient: SqsAsyncClient = mock[SqsAsyncClient]
when(
mockAwsSqsClient.deleteMessageBatch(any[DeleteMessageBatchRequest])).thenThrow(new RuntimeException("error"))
@@ -357,7 +357,7 @@ class SqsAckSpec extends AnyFlatSpec with Matchers with DefaultTestContext with
it should "ignore batch of messages" in {
val messages = for (i <- 0 until 10) yield Message.builder().body(s"Message - $i").build()
- implicit val mockAwsSqsClient = mock[SqsAsyncClient]
+ implicit val mockAwsSqsClient: SqsAsyncClient = mock[SqsAsyncClient]
val future =
// #batch-ignore
diff --git a/sqs/src/test/scala/docs/scaladsl/SqsSourceSpec.scala b/sqs/src/test/scala/docs/scaladsl/SqsSourceSpec.scala
index edc18d85a..abca1103d 100644
--- a/sqs/src/test/scala/docs/scaladsl/SqsSourceSpec.scala
+++ b/sqs/src/test/scala/docs/scaladsl/SqsSourceSpec.scala
@@ -256,7 +256,7 @@ class SqsSourceSpec extends AnyFlatSpec with ScalaFutures with Matchers with Def
val customClient: SdkAsyncHttpClient = PekkoHttpClient.builder().withActorSystem(system).build()
// #init-custom-client
- implicit val customSqsClient = SqsAsyncClient
+ implicit val customSqsClient: SqsAsyncClient = SqsAsyncClient
.builder()
.credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("x", "x")))
// #init-custom-client
diff --git a/sse/src/test/scala/docs/scaladsl/EventSourceSpec.scala b/sse/src/test/scala/docs/scaladsl/EventSourceSpec.scala
index c1e520cce..470ba08f6 100644
--- a/sse/src/test/scala/docs/scaladsl/EventSourceSpec.scala
+++ b/sse/src/test/scala/docs/scaladsl/EventSourceSpec.scala
@@ -32,7 +32,7 @@ import org.scalatest.BeforeAndAfterAll
import scala.collection.immutable
import scala.concurrent.duration.DurationInt
-import scala.concurrent.{ Await, Future }
+import scala.concurrent.{ Await, ExecutionContext, Future }
//#event-source
import org.apache.pekko
import pekko.http.scaladsl.Http
@@ -91,7 +91,7 @@ object EventSourceSpec {
import Server._
import context.dispatcher
- private implicit val sys = context.system
+ private implicit val sys: ActorSystem = context.system
context.system.scheduler.scheduleOnce(1.second, self, Bind)
@@ -147,8 +147,8 @@ object EventSourceSpec {
final class EventSourceSpec extends AsyncWordSpec with Matchers with BeforeAndAfterAll {
import EventSourceSpec._
- private implicit val system = ActorSystem()
- private implicit val ec = system.dispatcher
+ private implicit val system: ActorSystem = ActorSystem()
+ private implicit val ec: ExecutionContext = system.dispatcher
"EventSource" should {
"communicate correctly with an instable HTTP server" in {
diff --git a/udp/src/test/scala/docs/scaladsl/UdpSpec.scala b/udp/src/test/scala/docs/scaladsl/UdpSpec.scala
index 370ad284c..88ce915a7 100644
--- a/udp/src/test/scala/docs/scaladsl/UdpSpec.scala
+++ b/udp/src/test/scala/docs/scaladsl/UdpSpec.scala
@@ -41,8 +41,8 @@ class UdpSpec
with BeforeAndAfterAll
with LogCapturing {
- implicit val mat = Materializer(system)
- implicit val pat = PatienceConfig(3.seconds, 50.millis)
+ implicit val mat: Materializer = Materializer(system)
+ implicit val pat: PatienceConfig = PatienceConfig(3.seconds, 50.millis)
// #bind-address
val bindToLocal = new InetSocketAddress("localhost", 0)
diff --git a/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/scaladsl/UnixDomainSocket.scala b/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/scaladsl/UnixDomainSocket.scala
index 8527736c4..0f2ab1109 100644
--- a/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/scaladsl/UnixDomainSocket.scala
+++ b/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/scaladsl/UnixDomainSocket.scala
@@ -140,7 +140,7 @@ final class UnixDomainSocket(system: ExtendedActorSystem) extends UnixDomainSock
backlog: Int = 128,
halfClose: Boolean = false): Future[ServerBinding] =
bind(path, backlog, halfClose)
- .to(Sink.foreach { conn: IncomingConnection =>
+ .to(Sink.foreach { (conn: IncomingConnection) =>
conn.flow.join(handler).run()
})
.run()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
[incubator-pekko-connectors] 04/05: support scala3 (google-common) (#141)
Posted by fa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch scala3
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors.git
commit 99fd51ceb0c865af3ffc4a6783a3a5ca1d6fd6e4
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Thu Jun 8 15:59:33 2023 +0100
support scala3 (google-common) (#141)
* support scala3 (google-common)
* Update ResumableUpload.scala
* add huawei
* Update Dependencies.scala
* Update BigQuery.scala
* annotation
---
.../googlecloud/bigquery/BigQueryException.scala | 2 +-
.../googlecloud/bigquery/javadsl/BigQuery.scala | 29 ++++----
.../googlecloud/pubsub/impl/PubSubApi.scala | 6 +-
.../googlecloud/storage/impl/Formats.scala | 12 ++--
.../googlecloud/storage/impl/GCStorageStream.scala | 4 +-
.../storage/WithMaterializerGlobal.scala | 2 +-
.../impl/GCStorageStreamIntegrationSpec.scala | 2 +-
.../google/javadsl/XUploadContentType.java | 3 +-
.../stream/connectors/google/GoogleSettings.scala | 6 +-
.../stream/connectors/google/ResumableUpload.scala | 58 ++++++++--------
.../connectors/google/auth/Credentials.scala | 2 +-
.../connectors/google/auth/GoogleOAuth2.scala | 3 +-
.../google/auth/GoogleOAuth2Exception.scala | 3 +-
.../stream/connectors/google/http/GoogleHttp.scala | 10 +--
.../stream/connectors/google/javadsl/Google.scala | 4 +-
.../connectors/google/jwt/JwtSprayJson.scala | 81 ++++++++++++++++++++++
.../google/scaladsl/`X-Upload-Content-Type`.scala | 2 +-
.../connectors/google/ResumableUploadSpec.scala | 4 +-
.../connectors/google/auth/GoogleOAuth2Spec.scala | 2 +-
.../google/auth/OAuth2CredentialsSpec.scala | 2 +-
.../connectors/google/http/GoogleHttpSpec.scala | 5 +-
.../google/firebase/fcm/impl/FcmSender.scala | 2 +-
.../google/firebase/fcm/v1/impl/FcmSender.scala | 2 +-
.../firebase/fcm/v1/impl/FcmSenderSpec.scala | 2 +-
.../huawei/pushkit/impl/PushKitSenderSpec.scala | 2 +-
.../scala/docs/scaladsl/MqttActorSystemsSpec.scala | 6 +-
project/Dependencies.scala | 7 +-
27 files changed, 177 insertions(+), 86 deletions(-)
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/BigQueryException.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/BigQueryException.scala
index eb9e7e910..14bec8372 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/BigQueryException.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/BigQueryException.scala
@@ -39,7 +39,7 @@ object BigQueryException {
implicit val fromResponseUnmarshaller: FromResponseUnmarshaller[Throwable] =
Unmarshaller
- .withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
+ .withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
import SprayJsonSupport._
val HttpResponse(status, _, entity, _) = response: @nowarn("msg=match may not be exhaustive")
Unmarshaller
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/javadsl/BigQuery.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/javadsl/BigQuery.scala
index 7d8118ae3..22eea8cf9 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/javadsl/BigQuery.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/javadsl/BigQuery.scala
@@ -19,7 +19,7 @@ import pekko.annotation.ApiMayChange
import pekko.http.javadsl.marshalling.Marshaller
import pekko.http.javadsl.model.{ HttpEntity, RequestEntity }
import pekko.http.javadsl.unmarshalling.Unmarshaller
-import pekko.http.scaladsl.{ model => sm }
+import pekko.http.scaladsl.{ marshalling, model => sm, unmarshalling }
import pekko.japi.Pair
import pekko.stream.connectors.google.GoogleSettings
import pekko.stream.connectors.google.javadsl.Google
@@ -45,7 +45,6 @@ import pekko.util.OptionConverters._
import java.time.Duration
import java.util.concurrent.CompletionStage
import java.{ lang, util }
-
import scala.annotation.nowarn
import scala.concurrent.duration.{ FiniteDuration, MILLISECONDS }
@@ -220,7 +219,8 @@ object BigQuery extends Google {
selectedFields: util.List[String],
unmarshaller: Unmarshaller[HttpEntity, TableDataListResponse[Out]])
: Source[Out, CompletionStage[TableDataListResponse[Out]]] = {
- implicit val um = unmarshaller.asScalaCastInput[sm.HttpEntity]
+ implicit val um: unmarshalling.Unmarshaller[sm.HttpEntity, TableDataListResponse[Out]] =
+ unmarshaller.asScalaCastInput[sm.HttpEntity]
ScalaBigQuery
.tableData(datasetId, tableId, startIndex.toScala, maxResults.toScala, selectedFields.asScala.toList)
.mapMaterializedValue(_.asJava)
@@ -245,7 +245,8 @@ object BigQuery extends Google {
retryPolicy: InsertAllRetryPolicy,
templateSuffix: util.Optional[String],
marshaller: Marshaller[TableDataInsertAllRequest[In], RequestEntity]): Sink[util.List[In], NotUsed] = {
- implicit val m = marshaller.asScalaCastOutput[sm.RequestEntity]
+ implicit val m: marshalling.Marshaller[TableDataInsertAllRequest[In], sm.RequestEntity] =
+ marshaller.asScalaCastOutput[sm.RequestEntity]
ss.Flow[util.List[In]]
.map(_.asScala.toList)
.to(ScalaBigQuery.insertAll[In](datasetId, tableId, retryPolicy, templateSuffix.toScala))
@@ -269,7 +270,8 @@ object BigQuery extends Google {
retryFailedRequests: Boolean,
marshaller: Marshaller[TableDataInsertAllRequest[In], RequestEntity])
: Flow[TableDataInsertAllRequest[In], TableDataInsertAllResponse, NotUsed] = {
- implicit val m = marshaller.asScalaCastOutput[sm.RequestEntity]
+ implicit val m: marshalling.Marshaller[TableDataInsertAllRequest[In], sm.RequestEntity] =
+ marshaller.asScalaCastOutput[sm.RequestEntity]
ScalaBigQuery.insertAll[In](datasetId, tableId, retryFailedRequests).asJava
}
@@ -290,7 +292,8 @@ object BigQuery extends Google {
dryRun: Boolean,
useLegacySql: Boolean,
unmarshaller: Unmarshaller[HttpEntity, QueryResponse[Out]]): Source[Out, CompletionStage[QueryResponse[Out]]] = {
- implicit val um = unmarshaller.asScalaCastInput[sm.HttpEntity]
+ implicit val um: unmarshalling.Unmarshaller[sm.HttpEntity, QueryResponse[Out]] =
+ unmarshaller.asScalaCastInput[sm.HttpEntity]
ScalaBigQuery.query(query, dryRun, useLegacySql).mapMaterializedValue(_.asJava).asJava
}
@@ -309,7 +312,8 @@ object BigQuery extends Google {
query: QueryRequest,
unmarshaller: Unmarshaller[HttpEntity, QueryResponse[Out]])
: Source[Out, Pair[CompletionStage[JobReference], CompletionStage[QueryResponse[Out]]]] = {
- implicit val um = unmarshaller.asScalaCastInput[sm.HttpEntity]
+ implicit val um: unmarshalling.Unmarshaller[sm.HttpEntity, QueryResponse[Out]] =
+ unmarshaller.asScalaCastInput[sm.HttpEntity]
ScalaBigQuery
.query(query)
.mapMaterializedValue {
@@ -339,7 +343,8 @@ object BigQuery extends Google {
timeout: util.Optional[Duration],
location: util.Optional[String],
unmarshaller: Unmarshaller[HttpEntity, QueryResponse[Out]]): Source[Out, CompletionStage[QueryResponse[Out]]] = {
- implicit val um = unmarshaller.asScalaCastInput[sm.HttpEntity]
+ implicit val um: unmarshalling.Unmarshaller[sm.HttpEntity, QueryResponse[Out]] =
+ unmarshaller.asScalaCastInput[sm.HttpEntity]
ScalaBigQuery
.queryResults(jobId,
startIndex.toScala,
@@ -396,7 +401,7 @@ object BigQuery extends Google {
def insertAllAsync[In](datasetId: String,
tableId: String,
marshaller: Marshaller[In, RequestEntity]): Flow[In, Job, NotUsed] = {
- implicit val m = marshaller.asScalaCastOutput[sm.RequestEntity]
+ implicit val m: marshalling.Marshaller[In, sm.RequestEntity] = marshaller.asScalaCastOutput[sm.RequestEntity]
ScalaBigQuery.insertAllAsync[In](datasetId, tableId).asJava[In]
}
@@ -415,7 +420,7 @@ object BigQuery extends Google {
tableId: String,
labels: util.Optional[util.Map[String, String]],
marshaller: Marshaller[In, RequestEntity]): Flow[In, Job, NotUsed] = {
- implicit val m = marshaller.asScalaCastOutput[sm.RequestEntity]
+ implicit val m: marshalling.Marshaller[In, sm.RequestEntity] = marshaller.asScalaCastOutput[sm.RequestEntity]
ScalaBigQuery.insertAllAsync[In](datasetId, tableId, labels.toScala.map(_.asScala.toMap)).asJava[In]
}
@@ -436,8 +441,8 @@ object BigQuery extends Google {
job: Job,
marshaller: Marshaller[Job, RequestEntity],
unmarshaller: Unmarshaller[HttpEntity, Job]): Sink[ByteString, CompletionStage[Job]] = {
- implicit val m = marshaller.asScalaCastOutput[sm.RequestEntity]
- implicit val um = unmarshaller.asScalaCastInput[sm.HttpEntity]
+ implicit val m: marshalling.Marshaller[Job, sm.RequestEntity] = marshaller.asScalaCastOutput[sm.RequestEntity]
+ implicit val um: unmarshalling.Unmarshaller[sm.HttpEntity, Job] = unmarshaller.asScalaCastInput[sm.HttpEntity]
ScalaBigQuery.createLoadJob(job).mapMaterializedValue(_.asJava).asJava[ByteString]
}
diff --git a/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala b/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala
index 75b28878e..ccabe2828 100644
--- a/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala
+++ b/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/impl/PubSubApi.scala
@@ -173,7 +173,7 @@ private[pubsub] trait PubSubApi {
.mapMaterializedValue(_ => NotUsed)
private implicit val pullResponseUnmarshaller: FromResponseUnmarshaller[PullResponse] =
- Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
+ Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
response.status match {
case StatusCodes.Success(_) if response.entity.contentType == ContentTypes.`application/json` =>
Unmarshal(response.entity).to[PullResponse]
@@ -211,7 +211,7 @@ private[pubsub] trait PubSubApi {
.mapMaterializedValue(_ => NotUsed)
private implicit val acknowledgeResponseUnmarshaller: FromResponseUnmarshaller[Done] =
- Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
+ Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
response.status match {
case StatusCodes.Success(_) =>
response.discardEntityBytes().future
@@ -261,7 +261,7 @@ private[pubsub] trait PubSubApi {
publish(topic, parallelism, None)
private implicit val publishResponseUnmarshaller: FromResponseUnmarshaller[PublishResponse] =
- Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
+ Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
response.status match {
case StatusCodes.Success(_) if response.entity.contentType == ContentTypes.`application/json` =>
Unmarshal(response.entity).to[PublishResponse]
diff --git a/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/Formats.scala b/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/Formats.scala
index 003d0f244..9a36801ba 100644
--- a/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/Formats.scala
+++ b/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/Formats.scala
@@ -49,7 +49,8 @@ object Formats extends DefaultJsonProtocol {
domain: String,
projectTeam: ProjectTeam,
etag: String)
- private implicit val ObjectAccessControlsJsonFormat = jsonFormat13(ObjectAccessControls)
+ private implicit val ObjectAccessControlsJsonFormat: RootJsonFormat[ObjectAccessControls] =
+ jsonFormat13(ObjectAccessControls.apply)
/**
* Google API storage response object
@@ -79,7 +80,8 @@ object Formats extends DefaultJsonProtocol {
timeStorageClassUpdated: String,
updated: String)
- private implicit val storageObjectReadOnlyJson = jsonFormat18(StorageObjectReadOnlyJson)
+ private implicit val storageObjectReadOnlyJson: RootJsonFormat[StorageObjectReadOnlyJson] =
+ jsonFormat18(StorageObjectReadOnlyJson.apply)
// private sub class of StorageObjectJson used to workaround 22 field jsonFormat issue
private final case class StorageObjectWriteableJson(
@@ -98,7 +100,8 @@ object Formats extends DefaultJsonProtocol {
temporaryHold: Option[Boolean],
acl: Option[List[ObjectAccessControls]])
- private implicit val storageObjectWritableJson = jsonFormat14(StorageObjectWriteableJson)
+ private implicit val storageObjectWritableJson: RootJsonFormat[StorageObjectWriteableJson] =
+ jsonFormat14(StorageObjectWriteableJson.apply)
private implicit object StorageObjectJsonFormat extends RootJsonFormat[StorageObjectJson] {
override def read(value: JsValue): StorageObjectJson = {
@@ -175,7 +178,8 @@ object Formats extends DefaultJsonProtocol {
}
}
- private implicit val bucketListResultJsonReads = jsonFormat4(BucketListResultJson.apply)
+ private implicit val bucketListResultJsonReads: RootJsonFormat[BucketListResultJson] =
+ jsonFormat4(BucketListResultJson.apply)
implicit object RewriteResponseReads extends RootJsonReader[RewriteResponse] {
override def read(json: JsValue): RewriteResponse = {
diff --git a/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStream.scala b/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStream.scala
index c704d093b..4619acf7f 100644
--- a/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStream.scala
+++ b/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStream.scala
@@ -242,7 +242,7 @@ import scala.concurrent.{ ExecutionContext, Future }
getBucketPath(bucket) / "o" / objectName
implicit def unmarshaller[T: FromEntityUnmarshaller]: Unmarshaller[HttpResponse, T] =
- Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
+ Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
response match {
case HttpResponse(status, _, entity, _) if status.isSuccess() && !status.isRedirection() =>
Unmarshal(entity).to[T]
@@ -254,7 +254,7 @@ import scala.concurrent.{ ExecutionContext, Future }
}.withDefaultRetry
implicit def optionUnmarshaller[T: FromEntityUnmarshaller]: Unmarshaller[HttpResponse, Option[T]] =
- Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
+ Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
response match {
case HttpResponse(status, _, entity, _) if status.isSuccess() && !status.isRedirection() =>
Unmarshal(entity).to[T].map(Some(_))
diff --git a/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/WithMaterializerGlobal.scala b/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/WithMaterializerGlobal.scala
index 5edc3fdba..f6d2985e4 100644
--- a/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/WithMaterializerGlobal.scala
+++ b/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/WithMaterializerGlobal.scala
@@ -29,7 +29,7 @@ trait WithMaterializerGlobal
with ScalaFutures
with IntegrationPatience
with Matchers {
- implicit val actorSystem = ActorSystem("test")
+ implicit val actorSystem: ActorSystem = ActorSystem("test")
implicit val ec: ExecutionContext = actorSystem.dispatcher
override protected def afterAll(): Unit = {
diff --git a/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStreamIntegrationSpec.scala b/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStreamIntegrationSpec.scala
index d0ee0d8a5..e5d7c1520 100644
--- a/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStreamIntegrationSpec.scala
+++ b/google-cloud-storage/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/storage/impl/GCStorageStreamIntegrationSpec.scala
@@ -53,7 +53,7 @@ class GCStorageStreamIntegrationSpec
with ScalaFutures
with LogCapturing {
- private implicit val defaultPatience =
+ private implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = 60.seconds, interval = 60.millis)
var folderName: String = _
diff --git a/google-common/src/main/java/org/apache/pekko/stream/connectors/google/javadsl/XUploadContentType.java b/google-common/src/main/java/org/apache/pekko/stream/connectors/google/javadsl/XUploadContentType.java
index 03deb6968..30a8c29e9 100644
--- a/google-common/src/main/java/org/apache/pekko/stream/connectors/google/javadsl/XUploadContentType.java
+++ b/google-common/src/main/java/org/apache/pekko/stream/connectors/google/javadsl/XUploadContentType.java
@@ -22,7 +22,6 @@ public interface XUploadContentType {
ContentType getContentType();
static XUploadContentType create(ContentType contentType) {
- return X$minusUpload$minusContent$minusType$.MODULE$.apply(
- (org.apache.pekko.http.scaladsl.model.ContentType) contentType);
+ return X$minusUpload$minusContent$minusType$.MODULE$.apply(contentType.toString());
}
}
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/GoogleSettings.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/GoogleSettings.scala
index c0e98af1e..973cdb194 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/GoogleSettings.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/GoogleSettings.scala
@@ -90,7 +90,7 @@ object GoogleSettings {
}
-final case class GoogleSettings @InternalApi private (projectId: String,
+final case class GoogleSettings @InternalApi private[connectors] (projectId: String,
credentials: Credentials,
requestSettings: RequestSettings) {
def getProjectId = projectId
@@ -134,7 +134,7 @@ object RequestSettings {
apply(userIp.toScala, quotaUser.toScala, prettyPrint, chunkSize, retrySettings, forwardProxy.toScala)
}
-final case class RequestSettings @InternalApi private (
+final case class RequestSettings @InternalApi private[connectors] (
userIp: Option[String],
quotaUser: Option[String],
prettyPrint: Boolean,
@@ -252,7 +252,7 @@ object ForwardProxy {
credentials: Option[BasicHttpCredentials],
trustPem: Option[String])(implicit system: ClassicActorSystemProvider): ForwardProxy = {
ForwardProxy(
- trustPem.fold(Http(system).defaultClientHttpsContext)(ForwardProxyHttpsContext(_)),
+ trustPem.fold(Http(system.classicSystem).defaultClientHttpsContext)(ForwardProxyHttpsContext(_)),
ForwardProxyPoolSettings(scheme, host, port, credentials)(system.classicSystem))
}
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
index 630381f59..81f9e7db6 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/ResumableUpload.scala
@@ -56,7 +56,7 @@ private[connectors] object ResumableUpload {
Sink
.fromMaterializer { (mat, attr) =>
import mat.executionContext
- implicit val materializer = mat
+ implicit val materializer: Materializer = mat
implicit val settings: GoogleSettings = GoogleAttributes.resolveSettings(mat, attr)
val uploadChunkSize = settings.requestSettings.uploadChunkSize
@@ -96,25 +96,24 @@ private[connectors] object ResumableUpload {
private def initiateSession(request: HttpRequest)(implicit mat: Materializer,
settings: GoogleSettings): Future[Uri] = {
- implicit val system: ActorSystem = mat.system
import implicits._
- implicit val um = Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
- response.discardEntityBytes().future.map { _ =>
- response.header[Location].fold(throw InvalidResponseException(ErrorInfo("No Location header")))(_.uri)
- }
- }.withDefaultRetry
+ implicit val um: FromResponseUnmarshaller[Uri] =
+ Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
+ response.discardEntityBytes().future.map { _ =>
+ response.header[Location].fold(throw InvalidResponseException(ErrorInfo("No Location header")))(_.uri)
+ }
+ }.withDefaultRetry
- GoogleHttp().singleAuthenticatedRequest[Uri](request)
+ GoogleHttp(mat.system).singleAuthenticatedRequest[Uri](request)
}
private final case class DoNotRetry(ex: Throwable) extends Throwable(ex) with NoStackTrace
private def uploadChunk[T: FromResponseUnmarshaller](
request: HttpRequest)(implicit mat: Materializer): Flow[Either[T, MaybeLast[Chunk]], Try[Option[T]], NotUsed] = {
- implicit val system: ActorSystem = mat.system
- val um = Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
+ val um = Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
response.status match {
case PermanentRedirect =>
response.discardEntityBytes().future.map(_ => None)
@@ -127,7 +126,8 @@ private[connectors] object ResumableUpload {
val uri = request.uri
Flow[HttpRequest]
.map((_, ()))
- .via(GoogleHttp().cachedHostConnectionPoolWithContext(uri.authority.host.address, uri.effectivePort)(um))
+ .via(GoogleHttp(mat.system).cachedHostConnectionPoolWithContext(uri.authority.host.address, uri.effectivePort)(
+ um))
.map(_._1.recoverWith { case DoNotRetry(ex) => Failure(ex) })
}
@@ -147,30 +147,30 @@ private[connectors] object ResumableUpload {
request: HttpRequest,
chunk: Future[MaybeLast[Chunk]])(
implicit mat: Materializer, settings: GoogleSettings): Future[Either[T, MaybeLast[Chunk]]] = {
- implicit val system: ActorSystem = mat.system
import implicits._
- implicit val um = Unmarshaller.withMaterializer { implicit ec => implicit mat => response: HttpResponse =>
- response.status match {
- case OK | Created => Unmarshal(response).to[T].map(Left(_))
- case PermanentRedirect =>
- response.discardEntityBytes().future.map { _ =>
- Right(
- response
- .header[Range]
- .flatMap(_.ranges.headOption)
- .collect {
- case Slice(_, last) => last + 1
- }.getOrElse(0L))
- }
- case _ => throw InvalidResponseException(ErrorInfo(response.status.value, response.status.defaultMessage))
- }
- }.withDefaultRetry
+ implicit val um: FromResponseUnmarshaller[Either[T, Long]] =
+ Unmarshaller.withMaterializer { implicit ec => implicit mat => (response: HttpResponse) =>
+ response.status match {
+ case OK | Created => Unmarshal(response).to[T].map(Left(_))
+ case PermanentRedirect =>
+ response.discardEntityBytes().future.map { _ =>
+ Right(
+ response
+ .header[Range]
+ .flatMap(_.ranges.headOption)
+ .collect {
+ case Slice(_, last) => last + 1
+ }.getOrElse(0L))
+ }
+ case _ => throw InvalidResponseException(ErrorInfo(response.status.value, response.status.defaultMessage))
+ }
+ }.withDefaultRetry
import mat.executionContext
chunk.flatMap {
case maybeLast @ MaybeLast(Chunk(bytes, position)) =>
- GoogleHttp()
+ GoogleHttp(mat.system)
.singleAuthenticatedRequest[Either[T, Long]](request.addHeader(statusRequestHeader))
.map {
case Left(result) if maybeLast.isLast => Left(result)
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/Credentials.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/Credentials.scala
index 3d42f3adb..8f5afe6d8 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/Credentials.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/Credentials.scala
@@ -36,7 +36,7 @@ object Credentials {
*/
def apply(c: Config)(implicit system: ClassicActorSystemProvider): Credentials = c.getString("provider") match {
case "application-default" =>
- val log = Logging(system.classicSystem, getClass)
+ val log = Logging(system.classicSystem, classOf[Credentials])
try {
val creds = parseServiceAccount(c)
log.info("Using service account credentials")
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala
index 259770100..5456418b5 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala
@@ -21,9 +21,10 @@ import pekko.http.scaladsl.model.HttpMethods.POST
import pekko.http.scaladsl.model.{ FormData, HttpRequest }
import pekko.stream.Materializer
import pekko.stream.connectors.google.http.GoogleHttp
+import pekko.stream.connectors.google.jwt.JwtSprayJson
import pekko.stream.connectors.google.{ implicits, RequestSettings }
import pdi.jwt.JwtAlgorithm.RS256
-import pdi.jwt.{ JwtClaim, JwtSprayJson }
+import pdi.jwt.JwtClaim
import spray.json.DefaultJsonProtocol._
import spray.json.JsonFormat
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Exception.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Exception.scala
index ed392199a..464706f4b 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Exception.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Exception.scala
@@ -22,7 +22,8 @@ import pekko.stream.connectors.google.util.Retry
import spray.json.DefaultJsonProtocol._
import spray.json.RootJsonFormat
-final case class GoogleOAuth2Exception private (override val info: ErrorInfo) extends ExceptionWithErrorInfo(info)
+final case class GoogleOAuth2Exception private[google] (override val info: ErrorInfo)
+ extends ExceptionWithErrorInfo(info)
private[google] object GoogleOAuth2Exception {
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttp.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttp.scala
index ba364683e..933170855 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttp.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttp.scala
@@ -14,7 +14,7 @@
package org.apache.pekko.stream.connectors.google.http
import org.apache.pekko
-import pekko.actor.ClassicActorSystemProvider
+import pekko.actor.{ ClassicActorSystemProvider, ExtendedActorSystem, Scheduler }
import pekko.annotation.InternalApi
import pekko.dispatch.ExecutionContexts
import pekko.http.scaladsl.Http.HostConnectionPool
@@ -26,7 +26,7 @@ import pekko.stream.connectors.google.{ GoogleAttributes, GoogleSettings, Reques
import pekko.stream.connectors.google.util.Retry
import pekko.stream.scaladsl.{ Flow, FlowWithContext, Keep, RetryFlow }
-import scala.concurrent.Future
+import scala.concurrent.{ ExecutionContextExecutor, Future }
import scala.util.{ Failure, Success, Try }
@InternalApi
@@ -45,9 +45,9 @@ private[connectors] object GoogleHttp {
@InternalApi
private[connectors] final class GoogleHttp private (val http: HttpExt) extends AnyVal {
- private implicit def system = http.system
- private implicit def ec = system.dispatcher
- private implicit def scheduler = system.scheduler
+ private implicit def system: ExtendedActorSystem = http.system
+ private implicit def ec: ExecutionContextExecutor = system.dispatcher
+ private implicit def scheduler: Scheduler = system.scheduler
/**
* Sends a single [[HttpRequest]] and returns the raw [[HttpResponse]].
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/javadsl/Google.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/javadsl/Google.scala
index 8000dcd7d..4a9f84217 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/javadsl/Google.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/javadsl/Google.scala
@@ -18,7 +18,7 @@ import pekko.NotUsed
import pekko.actor.ClassicActorSystemProvider
import pekko.http.javadsl.model.{ HttpRequest, HttpResponse }
import pekko.http.javadsl.unmarshalling.Unmarshaller
-import pekko.http.scaladsl.{ model => sm }
+import pekko.http.scaladsl.{ model => sm, unmarshalling }
import pekko.stream.connectors.google.GoogleSettings
import pekko.stream.connectors.google.scaladsl.{ Google => ScalaGoogle }
import pekko.stream.javadsl.{ Sink, Source }
@@ -59,7 +59,7 @@ private[connectors] trait Google {
*/
final def paginatedRequest[Out <: Paginated](request: HttpRequest,
unmarshaller: Unmarshaller[HttpResponse, Out]): Source[Out, NotUsed] = {
- implicit val um = unmarshaller.asScala
+ implicit val um: unmarshalling.Unmarshaller[HttpResponse, Out] = unmarshaller.asScala
ScalaGoogle.paginatedRequest[Out](request).asJava
}
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/jwt/JwtSprayJson.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/jwt/JwtSprayJson.scala
new file mode 100644
index 000000000..8362fa88e
--- /dev/null
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/jwt/JwtSprayJson.scala
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, derived from Akka.
+ */
+
+package org.apache.pekko.stream.connectors.google.jwt
+
+import java.time.Clock
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pdi.jwt._
+import pdi.jwt.exceptions.JwtNonStringException
+import spray.json._
+
+/**
+ * Implementation of `JwtCore` using `JsObject` from spray-json.
+ */
+@InternalApi
+private[google] trait JwtSprayJsonParser[H, C] extends JwtJsonCommon[JsObject, H, C] {
+ protected def parse(value: String): JsObject = value.parseJson.asJsObject
+
+ protected def stringify(value: JsObject): String = value.compactPrint
+
+ protected def getAlgorithm(header: JsObject): Option[JwtAlgorithm] =
+ header.fields.get("alg").flatMap {
+ case JsString("none") => None
+ case JsString(algo) => Option(JwtAlgorithm.fromString(algo))
+ case JsNull => None
+ case _ => throw new JwtNonStringException("alg")
+ }
+
+}
+
+@InternalApi
+private[google] object JwtSprayJson extends JwtSprayJsonParser[JwtHeader, JwtClaim] {
+ import DefaultJsonProtocol._
+
+ def apply(clock: Clock): JwtSprayJson = new JwtSprayJson(clock)
+
+ override def parseHeader(header: String): JwtHeader = {
+ val jsObj = parse(header)
+ JwtHeader(
+ algorithm = getAlgorithm(jsObj),
+ typ = safeGetField[String](jsObj, "typ"),
+ contentType = safeGetField[String](jsObj, "cty"),
+ keyId = safeGetField[String](jsObj, "kid"))
+ }
+
+ override def parseClaim(claim: String): JwtClaim = {
+ val jsObj = parse(claim)
+ val content = JsObject(
+ jsObj.fields - "iss" - "sub" - "aud" - "exp" - "nbf" - "iat" - "jti")
+ JwtClaim(
+ content = stringify(content),
+ issuer = safeGetField[String](jsObj, "iss"),
+ subject = safeGetField[String](jsObj, "sub"),
+ audience = safeGetField[Set[String]](jsObj, "aud")
+ .orElse(safeGetField[String](jsObj, "aud").map(s => Set(s))),
+ expiration = safeGetField[Long](jsObj, "exp"),
+ notBefore = safeGetField[Long](jsObj, "nbf"),
+ issuedAt = safeGetField[Long](jsObj, "iat"),
+ jwtId = safeGetField[String](jsObj, "jti"))
+ }
+
+ private[this] def safeRead[A: JsonReader](js: JsValue) =
+ safeReader[A].read(js).fold(_ => None, a => Option(a))
+
+ private[this] def safeGetField[A: JsonReader](js: JsObject, name: String) =
+ js.fields.get(name).flatMap(safeRead[A])
+}
+
+@InternalApi
+private[google] class JwtSprayJson private (override val clock: Clock)
+ extends JwtSprayJsonParser[JwtHeader, JwtClaim] {
+ override def parseHeader(header: String): JwtHeader = JwtSprayJson.parseHeader(header)
+ override def parseClaim(header: String): JwtClaim = JwtSprayJson.parseClaim(header)
+}
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/scaladsl/`X-Upload-Content-Type`.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/scaladsl/`X-Upload-Content-Type`.scala
index f10fdf181..0dff2e87d 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/scaladsl/`X-Upload-Content-Type`.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/scaladsl/`X-Upload-Content-Type`.scala
@@ -34,7 +34,7 @@ object `X-Upload-Content-Type` extends ModeledCustomHeaderCompanion[`X-Upload-Co
contentType => Success(`X-Upload-Content-Type`(contentType)))
}
-final case class `X-Upload-Content-Type` private (contentType: ContentType)
+final case class `X-Upload-Content-Type` private[connectors] (contentType: ContentType)
extends ModeledCustomHeader[`X-Upload-Content-Type`]
with XUploadContentType {
override def value(): String = contentType.toString()
diff --git a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/ResumableUploadSpec.scala b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/ResumableUploadSpec.scala
index aa3559cab..9ae4806bb 100644
--- a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/ResumableUploadSpec.scala
+++ b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/ResumableUploadSpec.scala
@@ -18,7 +18,7 @@ import pekko.actor.ActorSystem
import pekko.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
import pekko.http.scaladsl.model.HttpMethods.POST
import pekko.http.scaladsl.model.{ ContentTypes, HttpRequest, Uri }
-import pekko.http.scaladsl.unmarshalling.Unmarshaller
+import pekko.http.scaladsl.unmarshalling.{ FromResponseUnmarshaller, Unmarshaller }
import pekko.stream.connectors.google.scaladsl.`X-Upload-Content-Type`
import pekko.stream.scaladsl.Source
import pekko.testkit.TestKit
@@ -79,7 +79,7 @@ class ResumableUploadSpec
.willReturn(created().header("Content-Type", "application/json").body("{}"))))
import implicits._
- implicit val um =
+ implicit val um: FromResponseUnmarshaller[JsValue] =
Unmarshaller.messageUnmarshallerFromEntityUnmarshaller(sprayJsValueUnmarshaller).withDefaultRetry
val result = Source
diff --git a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Spec.scala b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Spec.scala
index 6c20d12a9..bc5613120 100644
--- a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Spec.scala
+++ b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2Spec.scala
@@ -46,7 +46,7 @@ class GoogleOAuth2Spec
implicit val executionContext: ExecutionContext = system.dispatcher
implicit val settings: GoogleSettings = GoogleSettings(system)
- implicit val clock = Clock.systemUTC()
+ implicit val clock: Clock = Clock.systemUTC()
lazy val privateKey = {
val inputStream = getClass.getClassLoader.getResourceAsStream("private_pcks8.pem")
diff --git a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala
index 967a067e0..c649211c7 100644
--- a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala
+++ b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/auth/OAuth2CredentialsSpec.scala
@@ -45,7 +45,7 @@ class OAuth2CredentialsSpec
import system.dispatcher
implicit val settings: RequestSettings = GoogleSettings().requestSettings
- implicit val clock = Clock.systemUTC()
+ implicit val clock: Clock = Clock.systemUTC()
final object AccessTokenProvider {
@volatile var accessTokenPromise: Promise[AccessToken] = Promise.failed(new RuntimeException)
diff --git a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttpSpec.scala b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttpSpec.scala
index 125d406cf..f6d1204fa 100644
--- a/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttpSpec.scala
+++ b/google-common/src/test/scala/org/apache/pekko/stream/connectors/google/http/GoogleHttpSpec.scala
@@ -67,10 +67,11 @@ class GoogleHttpSpec
anyInt,
any[HttpsConnectionContext],
any[ConnectionPoolSettings],
- any[LoggingAdapter])).thenReturn(Flow[Any]
+ any[LoggingAdapter])).thenReturn(
+ Flow[Any]
.zipWith(response)(Keep.right)
.map(Try(_))
- .map((_, mock[Nothing]))
+ .map((_, mock[Nothing](scala.reflect.ClassTag.Nothing)))
.mapMaterializedValue(_ => mock[HostConnectionPool]),
Nil: _*): @nowarn("msg=dead code")
http
diff --git a/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmSender.scala b/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmSender.scala
index 7822968bf..dfa154fcc 100644
--- a/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmSender.scala
+++ b/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/impl/FcmSender.scala
@@ -55,7 +55,7 @@ private[fcm] class FcmSender {
}
implicit private val unmarshaller: FromResponseUnmarshaller[FcmSuccessResponse] = Unmarshaller.withMaterializer {
- implicit ec => implicit mat => response: HttpResponse =>
+ implicit ec => implicit mat => (response: HttpResponse) =>
if (response.status.isSuccess) {
Unmarshal(response.entity).to[FcmSuccessResponse]
} else {
diff --git a/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSender.scala b/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSender.scala
index 5c5a26147..8532054d1 100644
--- a/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSender.scala
+++ b/google-fcm/src/main/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSender.scala
@@ -50,7 +50,7 @@ private[fcm] class FcmSender {
}
implicit private val unmarshaller: FromResponseUnmarshaller[FcmSuccessResponse] = Unmarshaller.withMaterializer {
- implicit ec => implicit mat => response: HttpResponse =>
+ implicit ec => implicit mat => (response: HttpResponse) =>
if (response.status.isSuccess) {
Unmarshal(response.entity).to[FcmSuccessResponse]
} else {
diff --git a/google-fcm/src/test/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSenderSpec.scala b/google-fcm/src/test/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSenderSpec.scala
index c37111b3e..76d6b2259 100644
--- a/google-fcm/src/test/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSenderSpec.scala
+++ b/google-fcm/src/test/scala/org/apache/pekko/stream/connectors/google/firebase/fcm/v1/impl/FcmSenderSpec.scala
@@ -57,7 +57,7 @@ class FcmSenderSpec
implicit val executionContext: ExecutionContext = system.dispatcher
- implicit val conf = FcmSettings()
+ implicit val conf: FcmSettings = FcmSettings()
implicit val settings: GoogleSettings = GoogleSettings().copy(projectId = "projectId")
"FcmSender" should {
diff --git a/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSenderSpec.scala b/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSenderSpec.scala
index 8f739a32d..67cdcccbe 100644
--- a/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSenderSpec.scala
+++ b/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSenderSpec.scala
@@ -55,7 +55,7 @@ class PushKitSenderSpec
implicit val executionContext: ExecutionContext = system.dispatcher
- implicit val config = HmsSettings()
+ implicit val config: HmsSettings = HmsSettings()
"HmsSender" should {
diff --git a/mqtt-streaming/src/test/scala/docs/scaladsl/MqttActorSystemsSpec.scala b/mqtt-streaming/src/test/scala/docs/scaladsl/MqttActorSystemsSpec.scala
index b941e4ba0..47f69b275 100644
--- a/mqtt-streaming/src/test/scala/docs/scaladsl/MqttActorSystemsSpec.scala
+++ b/mqtt-streaming/src/test/scala/docs/scaladsl/MqttActorSystemsSpec.scala
@@ -14,6 +14,7 @@
package docs.scaladsl
import org.apache.pekko
+import pekko.actor
import pekko.actor.typed.scaladsl.Behaviors
import pekko.stream.connectors.mqtt.streaming.MqttSessionSettings
import pekko.stream.connectors.mqtt.streaming.scaladsl.{ ActorMqttClientSession, ActorMqttServerSession }
@@ -21,7 +22,8 @@ import org.scalatest.wordspec.AnyWordSpec
class MqttTypedActorSystemSpec extends AnyWordSpec {
- implicit val actorSystem = pekko.actor.typed.ActorSystem(Behaviors.ignore, "MqttTypedActorSystemSpec")
+ implicit val actorSystem: actor.typed.ActorSystem[Nothing] =
+ actor.typed.ActorSystem(Behaviors.ignore, "MqttTypedActorSystemSpec")
"A typed actor system" should {
"allow client creation" in {
@@ -41,7 +43,7 @@ class MqttTypedActorSystemSpec extends AnyWordSpec {
class MqttClassicActorSystemSpec extends AnyWordSpec {
- implicit val actorSystem = pekko.actor.ActorSystem("MqttClassicActorSystemSpec")
+ implicit val actorSystem: actor.ActorSystem = actor.ActorSystem("MqttClassicActorSystemSpec")
"A typed actor system" should {
"allow client creation" in {
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index 21f10c3f4..6f7301efb 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -197,11 +197,10 @@ object Dependencies {
"org.apache.logging.log4j" % "log4j-to-slf4j" % "2.17.1" % Test) ++ JacksonDatabindDependencies)
val GoogleCommon = Seq(
- crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion,
- "com.github.jwt-scala" %% "jwt-spray-json" % "7.1.4", // ApacheV2
+ "com.github.jwt-scala" %% "jwt-json-common" % "7.1.5", // ApacheV2
"com.google.auth" % "google-auth-library-credentials" % "0.24.1", // BSD 3-clause
"io.specto" % "hoverfly-java" % hoverflyVersion % Test // ApacheV2
) ++ Mockito)
@@ -253,7 +252,6 @@ object Dependencies {
"org.apache.pekko" %% "pekko-discovery" % PekkoVersion))
val GoogleFcm = Seq(
- crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion) ++ Mockito)
@@ -300,11 +298,10 @@ object Dependencies {
))
val HuaweiPushKit = Seq(
- crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"org.apache.pekko" %% "pekko-http" % PekkoHttpVersion,
"org.apache.pekko" %% "pekko-http-spray-json" % PekkoHttpVersion,
- "com.github.jwt-scala" %% "jwt-spray-json" % "7.1.4" // ApacheV2
+ "com.github.jwt-scala" %% "jwt-json-common" % "7.1.5" // ApacheV2
) ++ Mockito)
val InfluxDB = Seq(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org