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 2025/10/27 13:47:11 UTC
(pekko-connectors) branch main updated (ec74c930b -> edd3227a1)
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
from ec74c930b Update specs2-core to 4.23.0 (#1272)
new ad9891106 Update scalafmt-core to 3.9.10
new 52d9d26f9 Reformat with scalafmt 3.9.10
new edd3227a1 Add 'Reformat with scalafmt 3.9.10' to .git-blame-ignore-revs
The 3 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:
.git-blame-ignore-revs | 3 +++
.scalafmt.conf | 2 +-
.../connectors/amqp/AmqpConnectionProvider.scala | 8 ++++----
.../stream/connectors/csv/impl/CsvParsingStage.scala | 2 +-
.../ElasticsearchConnectionSettings.scala | 2 +-
.../connectors/elasticsearch/impl/RestBulkApiV5.scala | 2 +-
.../connectors/elasticsearch/impl/RestBulkApiV7.scala | 2 +-
.../connectors/file/impl/archive/ZipReaderSource.scala | 6 +++---
.../pubsub/grpc/impl/PekkoGrpcSettings.scala | 2 +-
.../stream/connectors/google/ResumableUpload.scala | 4 ++--
.../stream/connectors/google/util/EitherFlow.scala | 2 +-
.../connectors/influxdb/impl/InfluxDbSourceStage.scala | 4 ++--
.../stream/connectors/ironmq/impl/IronMqClient.scala | 2 +-
.../stream/connectors/jakartams/AcknowledgeMode.scala | 2 +-
.../connectors/jakartams/impl/JmsConnector.scala | 6 +++---
.../connectors/jakartams/impl/JmsProducerStage.scala | 2 +-
.../connectors/jakartams/impl/SoftReferenceCache.scala | 2 +-
.../connectors/jakartams/scaladsl/JmsConsumer.scala | 6 +++---
.../connectors/jakartams/scaladsl/JmsProducer.scala | 4 ++--
.../pekko/stream/connectors/jms/AcknowledgeMode.scala | 2 +-
.../stream/connectors/jms/impl/JmsConnector.scala | 6 +++---
.../stream/connectors/jms/impl/JmsProducerStage.scala | 2 +-
.../connectors/jms/impl/SoftReferenceCache.scala | 2 +-
.../stream/connectors/jms/scaladsl/JmsConsumer.scala | 6 +++---
.../stream/connectors/jms/scaladsl/JmsProducer.scala | 4 ++--
.../connectors/kinesis/impl/KinesisSourceStage.scala | 6 +++---
.../connectors/kinesis/scaladsl/KinesisSource.scala | 6 +++---
.../mqtt/streaming/scaladsl/MqttSession.scala | 12 ++++++------
project/TestChanged.scala | 2 +-
.../pekko/stream/connectors/s3/impl/Marshalling.scala | 18 +++++++++---------
.../pekko/stream/connectors/s3/impl/S3Stream.scala | 2 +-
.../apache/pekko/stream/connectors/s3/settings.scala | 4 ++--
.../stream/connectors/s3/scaladsl/S3WireMockBase.scala | 2 +-
.../stream/connectors/testkit/CapturingAppender.scala | 2 +-
.../pekko/stream/connectors/testkit/LogbackUtil.scala | 2 +-
.../unixdomainsocket/impl/UnixDomainSocketImpl.scala | 10 +++++-----
.../connectors/xml/impl/StreamingXmlWriter.scala | 2 +-
37 files changed, 78 insertions(+), 75 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
(pekko-connectors) 02/03: Reformat with scalafmt 3.9.10
Posted by fa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
commit 52d9d26f9e41562ac0dbbcc9e6dfb5db0fab8d5e
Author: scala-steward-asf[bot] <14...@users.noreply.github.com>
AuthorDate: Mon Oct 27 11:53:55 2025 +0000
Reformat with scalafmt 3.9.10
Executed command: scalafmt --non-interactive
---
.../connectors/amqp/AmqpConnectionProvider.scala | 8 ++++----
.../stream/connectors/csv/impl/CsvParsingStage.scala | 2 +-
.../ElasticsearchConnectionSettings.scala | 2 +-
.../connectors/elasticsearch/impl/RestBulkApiV5.scala | 2 +-
.../connectors/elasticsearch/impl/RestBulkApiV7.scala | 2 +-
.../connectors/file/impl/archive/ZipReaderSource.scala | 6 +++---
.../pubsub/grpc/impl/PekkoGrpcSettings.scala | 2 +-
.../stream/connectors/google/ResumableUpload.scala | 4 ++--
.../stream/connectors/google/util/EitherFlow.scala | 2 +-
.../connectors/influxdb/impl/InfluxDbSourceStage.scala | 4 ++--
.../stream/connectors/ironmq/impl/IronMqClient.scala | 2 +-
.../stream/connectors/jakartams/AcknowledgeMode.scala | 2 +-
.../connectors/jakartams/impl/JmsConnector.scala | 6 +++---
.../connectors/jakartams/impl/JmsProducerStage.scala | 2 +-
.../connectors/jakartams/impl/SoftReferenceCache.scala | 2 +-
.../connectors/jakartams/scaladsl/JmsConsumer.scala | 6 +++---
.../connectors/jakartams/scaladsl/JmsProducer.scala | 4 ++--
.../pekko/stream/connectors/jms/AcknowledgeMode.scala | 2 +-
.../stream/connectors/jms/impl/JmsConnector.scala | 6 +++---
.../stream/connectors/jms/impl/JmsProducerStage.scala | 2 +-
.../connectors/jms/impl/SoftReferenceCache.scala | 2 +-
.../stream/connectors/jms/scaladsl/JmsConsumer.scala | 6 +++---
.../stream/connectors/jms/scaladsl/JmsProducer.scala | 4 ++--
.../connectors/kinesis/impl/KinesisSourceStage.scala | 6 +++---
.../connectors/kinesis/scaladsl/KinesisSource.scala | 6 +++---
.../mqtt/streaming/scaladsl/MqttSession.scala | 12 ++++++------
project/TestChanged.scala | 2 +-
.../pekko/stream/connectors/s3/impl/Marshalling.scala | 18 +++++++++---------
.../pekko/stream/connectors/s3/impl/S3Stream.scala | 2 +-
.../apache/pekko/stream/connectors/s3/settings.scala | 4 ++--
.../stream/connectors/s3/scaladsl/S3WireMockBase.scala | 2 +-
.../stream/connectors/testkit/CapturingAppender.scala | 2 +-
.../pekko/stream/connectors/testkit/LogbackUtil.scala | 2 +-
.../unixdomainsocket/impl/UnixDomainSocketImpl.scala | 10 +++++-----
.../connectors/xml/impl/StreamingXmlWriter.scala | 2 +-
35 files changed, 74 insertions(+), 74 deletions(-)
diff --git a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectionProvider.scala b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectionProvider.scala
index c93068824..6d265e8dd 100644
--- a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectionProvider.scala
+++ b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectionProvider.scala
@@ -387,12 +387,12 @@ final class AmqpCachedConnectionProvider private (val provider: AmqpConnectionPr
connection
} catch {
case e: ConcurrentModificationException => throw e
- case e: Throwable =>
+ case e: Throwable =>
state.compareAndSet(Connecting, Empty)
throw e
}
} else getRecursive(amqpConnectionProvider)
- case Connecting => getRecursive(amqpConnectionProvider)
+ case Connecting => getRecursive(amqpConnectionProvider)
case c @ Connected(connection, clients) =>
if (state.compareAndSet(c, Connected(connection, clients + 1))) connection
else getRecursive(amqpConnectionProvider)
@@ -404,8 +404,8 @@ final class AmqpCachedConnectionProvider private (val provider: AmqpConnectionPr
@tailrec
private def releaseRecursive(amqpConnectionProvider: AmqpConnectionProvider, connection: Connection): Unit =
state.get match {
- case Empty => throw new IllegalStateException("There is no connection to release.")
- case Connecting => releaseRecursive(amqpConnectionProvider, connection)
+ case Empty => throw new IllegalStateException("There is no connection to release.")
+ case Connecting => releaseRecursive(amqpConnectionProvider, connection)
case c @ Connected(cachedConnection, clients) =>
if (cachedConnection != connection)
throw new IllegalArgumentException("Can't release a connection that's not owned by this provider")
diff --git a/csv/src/main/scala/org/apache/pekko/stream/connectors/csv/impl/CsvParsingStage.scala b/csv/src/main/scala/org/apache/pekko/stream/connectors/csv/impl/CsvParsingStage.scala
index 8b2f89e83..09f3bd8e0 100644
--- a/csv/src/main/scala/org/apache/pekko/stream/connectors/csv/impl/CsvParsingStage.scala
+++ b/csv/src/main/scala/org/apache/pekko/stream/connectors/csv/impl/CsvParsingStage.scala
@@ -60,7 +60,7 @@ import scala.util.control.NonFatal
private def tryPollBuffer() =
try buffer.poll(requireLineEnd = true) match {
case Some(csvLine) => push(out, csvLine)
- case _ =>
+ case _ =>
if (isClosed(in)) {
emitRemaining()
completeStage()
diff --git a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/ElasticsearchConnectionSettings.scala b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/ElasticsearchConnectionSettings.scala
index 82743c9df..3540ef358 100644
--- a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/ElasticsearchConnectionSettings.scala
+++ b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/ElasticsearchConnectionSettings.scala
@@ -47,7 +47,7 @@ final class ElasticsearchConnectionSettings private (
.map(x => {
HttpHeader.parse(x.name(), x.value()) match {
case ParsingResult.Ok(header, _) => header
- case ParsingResult.Error(error) =>
+ case ParsingResult.Error(error) =>
throw new Exception(s"Unable to convert java HttpHeader to scala HttpHeader: ${error.summary}")
}
})
diff --git a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApiV5.scala b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApiV5.scala
index e4e5f2069..30cc12a5b 100644
--- a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApiV5.scala
+++ b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApiV5.scala
@@ -48,7 +48,7 @@ private[impl] final class RestBulkApiV5[T, C](indexName: String,
optionalString("version_type", versionType),
optionalString("_id", message.id)).flatten
"index" -> JsObject(sharedFields ++ fields: _*)
- case Create => "create" -> JsObject(sharedFields ++ optionalString("_id", message.id): _*)
+ case Create => "create" -> JsObject(sharedFields ++ optionalString("_id", message.id): _*)
case Update | Upsert =>
val fields =
("_id" -> JsString(message.id.get)) +: Seq(
diff --git a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApiV7.scala b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApiV7.scala
index 5e0c896c2..caa790d51 100644
--- a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApiV7.scala
+++ b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApiV7.scala
@@ -47,7 +47,7 @@ private[impl] final class RestBulkApiV7[T, C](indexName: String,
"index" -> JsObject(sharedFields ++ fields: _*)
case Create => "create" -> JsObject(sharedFields ++ optionalString("_id", message.id): _*)
case Update | Upsert => "update" -> JsObject(sharedFields :+ ("_id" -> JsString(message.id.get)): _*)
- case Delete =>
+ case Delete =>
val fields =
("_id" -> JsString(message.id.get)) +: Seq(
optionalNumber("version", message.version),
diff --git a/file/src/main/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipReaderSource.scala b/file/src/main/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipReaderSource.scala
index 3c2bfb9ba..ece284b43 100644
--- a/file/src/main/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipReaderSource.scala
+++ b/file/src/main/scala/org/apache/pekko/stream/connectors/file/impl/archive/ZipReaderSource.scala
@@ -40,9 +40,9 @@ import java.util.zip.{ ZipEntry, ZipInputStream }
def seek() = {
while ({
- entry = zis.getNextEntry()
- entry != null && entry.getName != n.name
- }) {
+ entry = zis.getNextEntry()
+ entry != null && entry.getName != n.name
+ }) {
zis.closeEntry()
}
}
diff --git a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/PekkoGrpcSettings.scala b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/PekkoGrpcSettings.scala
index ec0a686a3..fbe992805 100644
--- a/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/PekkoGrpcSettings.scala
+++ b/google-cloud-pub-sub-grpc/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/impl/PekkoGrpcSettings.scala
@@ -43,7 +43,7 @@ import scala.annotation.nowarn
.withFallback(sys.settings.config.getConfig("pekko.grpc.client.\"*\"")))
(config.callCredentials: @nowarn("msg=deprecated")) match {
- case None => settings
+ case None => settings
case Some(DeprecatedCredentials(_)) => // Deprecated credentials were loaded from config so override them
sys.log.warning(
"Config path pekko.connectors.google.cloud.pubsub.grpc.callCredentials is deprecated, use pekko.connectors.google.credentials")
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 04dd782cf..85978f377 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
@@ -162,7 +162,7 @@ private[connectors] object ResumableUpload {
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 OK | Created => Unmarshal(response).to[T].map(Left(_))
case PermanentRedirect =>
response.discardEntityBytes().future.map { _ =>
Right(
@@ -183,7 +183,7 @@ private[connectors] object ResumableUpload {
GoogleHttp(mat.system)
.singleAuthenticatedRequest[Either[T, Long]](request.addHeader(statusRequestHeader))
.map {
- case Left(result) if maybeLast.isLast => Left(result)
+ case Left(result) if maybeLast.isLast => Left(result)
case Right(newPosition) if newPosition >= position =>
Right(maybeLast.map { _ =>
Chunk(bytes.drop(Math.toIntExact(newPosition - position)), newPosition)
diff --git a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/util/EitherFlow.scala b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/util/EitherFlow.scala
index 10f6682da..3904db2cc 100644
--- a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/util/EitherFlow.scala
+++ b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/util/EitherFlow.scala
@@ -30,7 +30,7 @@ private[google] object EitherFlow {
import GraphDSL.Implicits._
val in = b.add(Partition[Either[LeftIn, RightIn]](2, x => if (x.isRight) 1 else 0))
val out = b.add(Merge[Either[LeftOut, RightOut]](2))
- in ~> Flow[Either[LeftIn, RightIn]].map(_.swap.toOption.get) ~> leftFlow ~> Flow[LeftOut].map(Left(_)) ~> out
+ in ~> Flow[Either[LeftIn, RightIn]].map(_.swap.toOption.get) ~> leftFlow ~> Flow[LeftOut].map(Left(_)) ~> out
in ~> Flow[Either[LeftIn, RightIn]].map(_.toOption.get) ~> rightFlow ~> Flow[RightOut]
.map(
Right(_)) ~> out
diff --git a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/InfluxDbSourceStage.scala b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/InfluxDbSourceStage.scala
index b2bd5a789..9738b29a1 100644
--- a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/InfluxDbSourceStage.scala
+++ b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/InfluxDbSourceStage.scala
@@ -65,7 +65,7 @@ private[influxdb] final class InfluxDbSourceLogic[T](clazz: Class[T],
}
override def onPull(): Unit =
this.dataRetrieved match {
- case None => completeStage()
+ case None => completeStage()
case Some(queryResult) => {
for (result <- queryResult.getResults.asScala) {
if (result.hasError) {
@@ -112,7 +112,7 @@ private[influxdb] final class InfluxDbSourceRawLogic(query: Query,
override def onPull(): Unit =
dataRetrieved match {
- case None => completeStage()
+ case None => completeStage()
case Some(queryResult) => {
emit(outlet, queryResult)
dataRetrieved = None
diff --git a/ironmq/src/main/scala/org/apache/pekko/stream/connectors/ironmq/impl/IronMqClient.scala b/ironmq/src/main/scala/org/apache/pekko/stream/connectors/ironmq/impl/IronMqClient.scala
index f6143a2fc..e2d07e8f0 100644
--- a/ironmq/src/main/scala/org/apache/pekko/stream/connectors/ironmq/impl/IronMqClient.scala
+++ b/ironmq/src/main/scala/org/apache/pekko/stream/connectors/ironmq/impl/IronMqClient.scala
@@ -57,7 +57,7 @@ private[ironmq] final class IronMqClient(settings: IronMqSettings)(implicit acto
case "https" =>
http.cachedHostConnectionPoolHttps[NotUsed](endpoint.authority.host.address(), endpoint.authority.port)
case "http" => http.cachedHostConnectionPool[NotUsed](endpoint.authority.host.address(), endpoint.authority.port)
- case other =>
+ case other =>
throw new IllegalArgumentException(s"Endpoint $endpoint contains an invalid HTTP/HTTPS scheme $other")
}
}
diff --git a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/AcknowledgeMode.scala b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/AcknowledgeMode.scala
index e08490c2b..93b2ee679 100644
--- a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/AcknowledgeMode.scala
+++ b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/AcknowledgeMode.scala
@@ -42,7 +42,7 @@ object AcknowledgeMode {
case "client" => ClientAcknowledge
case "duplicates-ok" => DupsOkAcknowledge
case "session" => SessionTransacted
- case other =>
+ case other =>
try {
val mode = other.toInt
new AcknowledgeMode(mode)
diff --git a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/JmsConnector.scala b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/JmsConnector.scala
index 25d5a05b0..a70e208f4 100644
--- a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/JmsConnector.scala
+++ b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/JmsConnector.scala
@@ -81,7 +81,7 @@ private[jakartams] trait JmsConnector[S <: JmsSession] {
val update: InternalConnectionState => InternalConnectionState = {
case JmsConnectorStopping(completion) => JmsConnectorStopped(completion)
case stopped: JmsConnectorStopped => stopped
- case current =>
+ case current =>
JmsConnectorStopped(
Failure(new IllegalStateException(s"Completing stage stop in unexpected state ${current.getClass}")))
}
@@ -136,7 +136,7 @@ private[jakartams] trait JmsConnector[S <: JmsSession] {
case _ =>
connectionState match {
case _: JmsConnectorStopping | _: JmsConnectorStopped => logStoppingException(ex)
- case _ =>
+ case _ =>
log.error(ex, "{} connection failed for destination[{}]", attributes.nameLifted.mkString, destination.name)
publishAndFailStage(ex)
}
@@ -150,7 +150,7 @@ private[jakartams] trait JmsConnector[S <: JmsSession] {
case JmsConnectorConnected(_) | JmsConnectorDisconnected =>
maybeReconnect(ex, 0, backoffMaxed = false)
case _: JmsConnectorStopping | _: JmsConnectorStopped => logStoppingException(ex)
- case other =>
+ case other =>
log.warning("received [{}] in connectionState={}", ex, connectionState)
}
}
diff --git a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/JmsProducerStage.scala b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/JmsProducerStage.scala
index 8d6122f85..a7b38643d 100644
--- a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/JmsProducerStage.scala
+++ b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/JmsProducerStage.scala
@@ -283,7 +283,7 @@ private[jakartams] object JmsProducerStage {
def supervisionDirectiveFor(decider: Supervision.Decider, ex: Throwable): Supervision.Directive =
cachedSupervisionDirective match {
case OptionVal.Some(d) => d
- case OptionVal.None =>
+ case OptionVal.None =>
val d = decider(ex)
cachedSupervisionDirective = OptionVal.Some(d)
d
diff --git a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/SoftReferenceCache.scala b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/SoftReferenceCache.scala
index 6de89c2b2..d6d04b7a7 100644
--- a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/SoftReferenceCache.scala
+++ b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/impl/SoftReferenceCache.scala
@@ -32,7 +32,7 @@ private final class SoftReferenceCache[K, V <: AnyRef] {
case Some(ref) =>
ref.get match {
case Some(value) => value
- case None =>
+ case None =>
purgeCache() // facing a garbage collected soft reference, purge other entries.
update(key, default)
}
diff --git a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/scaladsl/JmsConsumer.scala b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/scaladsl/JmsConsumer.scala
index 371c6d4d6..68cb68f16 100644
--- a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/scaladsl/JmsConsumer.scala
+++ b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/scaladsl/JmsConsumer.scala
@@ -33,7 +33,7 @@ object JmsConsumer {
*/
def apply(settings: JmsConsumerSettings): Source[jakarta.jms.Message, JmsConsumerControl] =
settings.destination match {
- case None => throw new IllegalArgumentException(noConsumerDestination(settings))
+ case None => throw new IllegalArgumentException(noConsumerDestination(settings))
case Some(destination) =>
Source.fromGraph(new JmsConsumerStage(settings, destination)).mapMaterializedValue(toConsumerControl)
}
@@ -85,7 +85,7 @@ object JmsConsumer {
* It requires explicit acknowledgements on the envelopes. The acknowledgements must be called on the envelope and not on the message inside.
*/
def ackSource(settings: JmsConsumerSettings): Source[AckEnvelope, JmsConsumerControl] = settings.destination match {
- case None => throw new IllegalArgumentException(noConsumerDestination(settings))
+ case None => throw new IllegalArgumentException(noConsumerDestination(settings))
case Some(destination) =>
Source.fromGraph(new JmsAckSourceStage(settings, destination)).mapMaterializedValue(toConsumerControl)
}
@@ -96,7 +96,7 @@ object JmsConsumer {
* It requires explicit committing or rollback on the envelopes.
*/
def txSource(settings: JmsConsumerSettings): Source[TxEnvelope, JmsConsumerControl] = settings.destination match {
- case None => throw new IllegalArgumentException(noConsumerDestination(settings))
+ case None => throw new IllegalArgumentException(noConsumerDestination(settings))
case Some(destination) =>
Source.fromGraph(new JmsTxSourceStage(settings, destination)).mapMaterializedValue(toConsumerControl)
}
diff --git a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/scaladsl/JmsProducer.scala b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/scaladsl/JmsProducer.scala
index 17a2976a7..724f41746 100644
--- a/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/scaladsl/JmsProducer.scala
+++ b/jakartams/src/main/scala/org/apache/pekko/stream/connectors/jakartams/scaladsl/JmsProducer.scala
@@ -33,7 +33,7 @@ object JmsProducer {
*/
def flow[T <: JmsMessage](settings: JmsProducerSettings): Flow[T, T, JmsProducerStatus] =
settings.destination match {
- case None => throw new IllegalArgumentException(noProducerDestination(settings))
+ case None => throw new IllegalArgumentException(noProducerDestination(settings))
case Some(destination) =>
Flow[T]
.viaMat(Flow.fromGraph(new JmsProducerStage[T, NotUsed](settings, destination)))(Keep.right)
@@ -47,7 +47,7 @@ object JmsProducer {
def flexiFlow[PassThrough](
settings: JmsProducerSettings): Flow[JmsEnvelope[PassThrough], JmsEnvelope[PassThrough], JmsProducerStatus] =
settings.destination match {
- case None => throw new IllegalArgumentException(noProducerDestination(settings))
+ case None => throw new IllegalArgumentException(noProducerDestination(settings))
case Some(destination) =>
Flow
.fromGraph(new JmsProducerStage[JmsEnvelope[PassThrough], PassThrough](settings, destination))
diff --git a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/AcknowledgeMode.scala b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/AcknowledgeMode.scala
index a456f1f46..46918a24d 100644
--- a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/AcknowledgeMode.scala
+++ b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/AcknowledgeMode.scala
@@ -42,7 +42,7 @@ object AcknowledgeMode {
case "client" => ClientAcknowledge
case "duplicates-ok" => DupsOkAcknowledge
case "session" => SessionTransacted
- case other =>
+ case other =>
try {
val mode = other.toInt
new AcknowledgeMode(mode)
diff --git a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsConnector.scala b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsConnector.scala
index 9cc6ed10b..b2f3706b9 100644
--- a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsConnector.scala
+++ b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsConnector.scala
@@ -82,7 +82,7 @@ private[jms] trait JmsConnector[S <: JmsSession] {
val update: InternalConnectionState => InternalConnectionState = {
case JmsConnectorStopping(completion) => JmsConnectorStopped(completion)
case stopped: JmsConnectorStopped => stopped
- case current =>
+ case current =>
JmsConnectorStopped(
Failure(new IllegalStateException(s"Completing stage stop in unexpected state ${current.getClass}")))
}
@@ -137,7 +137,7 @@ private[jms] trait JmsConnector[S <: JmsSession] {
case _ =>
connectionState match {
case _: JmsConnectorStopping | _: JmsConnectorStopped => logStoppingException(ex)
- case _ =>
+ case _ =>
log.error(ex, "{} connection failed for destination[{}]", attributes.nameLifted.mkString, destination.name)
publishAndFailStage(ex)
}
@@ -151,7 +151,7 @@ private[jms] trait JmsConnector[S <: JmsSession] {
case JmsConnectorConnected(_) | JmsConnectorDisconnected =>
maybeReconnect(ex, 0, backoffMaxed = false)
case _: JmsConnectorStopping | _: JmsConnectorStopped => logStoppingException(ex)
- case other =>
+ case other =>
log.warning("received [{}] in connectionState={}", ex, connectionState)
}
}
diff --git a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsProducerStage.scala b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsProducerStage.scala
index b0bcb5daa..aae31aa04 100644
--- a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsProducerStage.scala
+++ b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsProducerStage.scala
@@ -282,7 +282,7 @@ private[jms] object JmsProducerStage {
def supervisionDirectiveFor(decider: Supervision.Decider, ex: Throwable): Supervision.Directive =
cachedSupervisionDirective match {
case OptionVal.Some(d) => d
- case OptionVal.None =>
+ case OptionVal.None =>
val d = decider(ex)
cachedSupervisionDirective = OptionVal.Some(d)
d
diff --git a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/SoftReferenceCache.scala b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/SoftReferenceCache.scala
index c6c4c6ec6..c9a1e1d2c 100644
--- a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/SoftReferenceCache.scala
+++ b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/SoftReferenceCache.scala
@@ -32,7 +32,7 @@ private final class SoftReferenceCache[K, V <: AnyRef] {
case Some(ref) =>
ref.get match {
case Some(value) => value
- case None =>
+ case None =>
purgeCache() // facing a garbage collected soft reference, purge other entries.
update(key, default)
}
diff --git a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsConsumer.scala b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsConsumer.scala
index fd5000d1a..526af8691 100644
--- a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsConsumer.scala
+++ b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsConsumer.scala
@@ -32,7 +32,7 @@ object JmsConsumer {
* control instance to shut down the consumer.
*/
def apply(settings: JmsConsumerSettings): Source[javax.jms.Message, JmsConsumerControl] = settings.destination match {
- case None => throw new IllegalArgumentException(noConsumerDestination(settings))
+ case None => throw new IllegalArgumentException(noConsumerDestination(settings))
case Some(destination) =>
Source.fromGraph(new JmsConsumerStage(settings, destination)).mapMaterializedValue(toConsumerControl)
}
@@ -84,7 +84,7 @@ object JmsConsumer {
* It requires explicit acknowledgements on the envelopes. The acknowledgements must be called on the envelope and not on the message inside.
*/
def ackSource(settings: JmsConsumerSettings): Source[AckEnvelope, JmsConsumerControl] = settings.destination match {
- case None => throw new IllegalArgumentException(noConsumerDestination(settings))
+ case None => throw new IllegalArgumentException(noConsumerDestination(settings))
case Some(destination) =>
Source.fromGraph(new JmsAckSourceStage(settings, destination)).mapMaterializedValue(toConsumerControl)
}
@@ -95,7 +95,7 @@ object JmsConsumer {
* It requires explicit committing or rollback on the envelopes.
*/
def txSource(settings: JmsConsumerSettings): Source[TxEnvelope, JmsConsumerControl] = settings.destination match {
- case None => throw new IllegalArgumentException(noConsumerDestination(settings))
+ case None => throw new IllegalArgumentException(noConsumerDestination(settings))
case Some(destination) =>
Source.fromGraph(new JmsTxSourceStage(settings, destination)).mapMaterializedValue(toConsumerControl)
}
diff --git a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsProducer.scala b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsProducer.scala
index 15b28697e..93c77c361 100644
--- a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsProducer.scala
+++ b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/scaladsl/JmsProducer.scala
@@ -33,7 +33,7 @@ object JmsProducer {
*/
def flow[T <: JmsMessage](settings: JmsProducerSettings): Flow[T, T, JmsProducerStatus] =
settings.destination match {
- case None => throw new IllegalArgumentException(noProducerDestination(settings))
+ case None => throw new IllegalArgumentException(noProducerDestination(settings))
case Some(destination) =>
Flow[T]
.viaMat(Flow.fromGraph(new JmsProducerStage[T, NotUsed](settings, destination)))(Keep.right)
@@ -47,7 +47,7 @@ object JmsProducer {
def flexiFlow[PassThrough](
settings: JmsProducerSettings): Flow[JmsEnvelope[PassThrough], JmsEnvelope[PassThrough], JmsProducerStatus] =
settings.destination match {
- case None => throw new IllegalArgumentException(noProducerDestination(settings))
+ case None => throw new IllegalArgumentException(noProducerDestination(settings))
case Some(destination) =>
Flow
.fromGraph(new JmsProducerStage[JmsEnvelope[PassThrough], PassThrough](settings, destination))
diff --git a/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/impl/KinesisSourceStage.scala b/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/impl/KinesisSourceStage.scala
index 538774147..b8db4c5c2 100644
--- a/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/impl/KinesisSourceStage.scala
+++ b/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/impl/KinesisSourceStage.scala
@@ -92,7 +92,7 @@ private[kinesis] class KinesisSourceStage(shardSettings: ShardSettings, amazonKi
failStage(error)
case (_, Pump) =>
- case (_, msg) =>
+ case (_, msg) =>
throw new IllegalArgumentException(s"unexpected message $msg in state `ready`")
}
@@ -119,7 +119,7 @@ private[kinesis] class KinesisSourceStage(shardSettings: ShardSettings, amazonKi
failStage(error)
case (_, Pump) =>
- case (_, msg) =>
+ case (_, msg) =>
throw new IllegalArgumentException(s"unexpected message $msg in state `ready`")
}
@@ -140,7 +140,7 @@ private[kinesis] class KinesisSourceStage(shardSettings: ShardSettings, amazonKi
override protected def onTimer(timerKey: Any): Unit = timerKey match {
case GetRecords => requestRecords()
- case other =>
+ case other =>
log.warning("unexpected timer [{}]", other)
}
diff --git a/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/scaladsl/KinesisSource.scala b/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/scaladsl/KinesisSource.scala
index ba972ba14..8fb6ee634 100644
--- a/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/scaladsl/KinesisSource.scala
+++ b/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/scaladsl/KinesisSource.scala
@@ -40,9 +40,9 @@ object KinesisSource {
require(shardSettings.nonEmpty, "shard settings need to be specified")
val create: ShardSettings => Source[Record, NotUsed] = basic(_, amazonKinesisAsync)
shardSettings match {
- case Nil => Source.failed(NoShardsError)
- case first :: Nil => create(first)
- case first :: second :: Nil => Source.combine(create(first), create(second))(Merge(_))
+ case Nil => Source.failed(NoShardsError)
+ case first :: Nil => create(first)
+ case first :: second :: Nil => Source.combine(create(first), create(second))(Merge(_))
case first :: second :: rest =>
Source.combine(create(first), create(second), rest.map(create(_)): _*)(Merge(_))
}
diff --git a/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/scaladsl/MqttSession.scala b/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/scaladsl/MqttSession.scala
index 111eb7f42..eb66ea3a4 100644
--- a/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/scaladsl/MqttSession.scala
+++ b/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/scaladsl/MqttSession.scala
@@ -199,7 +199,7 @@ final class ActorMqttClientSession(settings: MqttSessionSettings)(implicit syste
case (_, terminated) =>
terminated.onComplete {
case Failure(_: WatchedActorTerminatedException) =>
- case _ =>
+ case _ =>
clientConnector ! ClientConnector.ConnectionLost(connectionId)
}
NotUsed
@@ -213,8 +213,8 @@ final class ActorMqttClientSession(settings: MqttSessionSettings)(implicit syste
clientConnector ! ClientConnector.ConnectReceivedLocally(connectionId, cp, carry, reply)
Source.futureSource(
reply.future.map(_.map {
- case ClientConnector.ForwardConnect => cp.encode(ByteString.newBuilder).result()
- case ClientConnector.ForwardPingReq => pingReqBytes
+ case ClientConnector.ForwardConnect => cp.encode(ByteString.newBuilder).result()
+ case ClientConnector.ForwardPingReq => pingReqBytes
case ClientConnector.ForwardPublish(publish, packetId) =>
publish.encode(ByteString.newBuilder, packetId).result()
case ClientConnector.ForwardPubRel(packetId) =>
@@ -307,7 +307,7 @@ final class ActorMqttClientSession(settings: MqttSessionSettings)(implicit syste
case (_, terminated) =>
terminated.onComplete {
case Failure(_: WatchedActorTerminatedException) =>
- case _ =>
+ case _ =>
clientConnector ! ClientConnector.ConnectionLost(connectionId)
}
NotUsed
@@ -535,7 +535,7 @@ final class ActorMqttServerSession(settings: MqttSessionSettings)(implicit syste
case (_, terminated) =>
terminated.onComplete {
case Failure(_: WatchedActorTerminatedException) =>
- case _ =>
+ case _ =>
serverConnector ! ServerConnector.ConnectionLost(connectionId)
}
NotUsed
@@ -660,7 +660,7 @@ final class ActorMqttServerSession(settings: MqttSessionSettings)(implicit syste
case (_, terminated) =>
terminated.onComplete {
case Failure(_: WatchedActorTerminatedException) =>
- case _ =>
+ case _ =>
serverConnector ! ServerConnector.ConnectionLost(connectionId)
}
NotUsed
diff --git a/project/TestChanged.scala b/project/TestChanged.scala
index 55c80969c..4a7d6f879 100644
--- a/project/TestChanged.scala
+++ b/project/TestChanged.scala
@@ -51,7 +51,7 @@ object TestChanged extends AutoPlugin {
}
private def shouldBuild(projectName: String, changedDirectories: Set[String]) = projectName match {
- case "connectors" => false
+ case "connectors" => false
case re"pekko-connectors-(.+)$subproject" =>
changedDirectories.contains(subproject) || changedDirectories.contains("") || changedDirectories.contains(
"project")
diff --git a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/Marshalling.scala b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/Marshalling.scala
index 4e9cacef5..2581f60d9 100644
--- a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/Marshalling.scala
+++ b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/Marshalling.scala
@@ -35,7 +35,7 @@ import scala.xml.NodeSeq
implicit val bucketVersioningUnmarshaller: FromEntityUnmarshaller[BucketVersioningResult] = {
nodeSeqUnmarshaller(MediaTypes.`application/xml`, ContentTypes.`application/octet-stream`).map {
case NodeSeq.Empty => throw Unmarshaller.NoContentException
- case x =>
+ case x =>
val status = (x \ "Status").headOption.map(_.text match {
case "Enabled" => BucketVersioningStatus.Enabled
case "Suspended" => BucketVersioningStatus.Suspended
@@ -48,7 +48,7 @@ import scala.xml.NodeSeq
implicit val multipartUploadUnmarshaller: FromEntityUnmarshaller[MultipartUpload] = {
nodeSeqUnmarshaller(MediaTypes.`application/xml`, ContentTypes.`application/octet-stream`).map {
case NodeSeq.Empty => throw Unmarshaller.NoContentException
- case x =>
+ case x =>
MultipartUpload((x \ "Bucket").text, (x \ "Key").text, (x \ "UploadId").text)
}
}
@@ -57,7 +57,7 @@ import scala.xml.NodeSeq
nodeSeqUnmarshaller(MediaTypes.`application/xml`.withCharset(HttpCharsets.`UTF-8`),
MediaTypes.`text/event-stream`).map {
case NodeSeq.Empty => throw Unmarshaller.NoContentException
- case x =>
+ case x =>
CompleteMultipartUploadResult(
Try(Uri((x \ "Location").text))
.getOrElse(Uri((x \ "Location").text.split("/").map(s => URLEncoder.encode(s, "utf-8")).mkString("/"))),
@@ -73,7 +73,7 @@ import scala.xml.NodeSeq
implicit val listBucketResultUnmarshaller: FromEntityUnmarshaller[ListBucketResult] = {
nodeSeqUnmarshaller(MediaTypes.`application/xml`.withCharset(HttpCharsets.`UTF-8`)).map {
case NodeSeq.Empty => throw Unmarshaller.NoContentException
- case x =>
+ case x =>
val truncated = (x \ isTruncated).text == "true"
val continuation = if (truncated) {
Some(x \ apiV2ContinuationToken)
@@ -104,7 +104,7 @@ import scala.xml.NodeSeq
implicit val copyPartResultUnmarshaller: FromEntityUnmarshaller[CopyPartResult] = {
nodeSeqUnmarshaller(MediaTypes.`application/xml`, ContentTypes.`application/octet-stream`).map {
case NodeSeq.Empty => throw Unmarshaller.NoContentException
- case x =>
+ case x =>
val lastModified = Instant.parse((x \ "LastModified").text)
val eTag = (x \ "ETag").text
CopyPartResult(lastModified, Utils.removeQuotes(eTag))
@@ -114,7 +114,7 @@ import scala.xml.NodeSeq
implicit val listBucketsResultUnmarshaller: FromEntityUnmarshaller[ListBucketsResult] = {
nodeSeqUnmarshaller(MediaTypes.`application/xml`.withCharset(HttpCharsets.`UTF-8`)).map {
case NodeSeq.Empty => throw Unmarshaller.NoContentException
- case x =>
+ case x =>
val bucketsRoot = (x \\ "Buckets").headOption
.map { br =>
(br \\ "Bucket").map { u =>
@@ -134,7 +134,7 @@ import scala.xml.NodeSeq
implicit val listMultipartUploadsResultUnmarshaller: FromEntityUnmarshaller[ListMultipartUploadsResult] = {
nodeSeqUnmarshaller(MediaTypes.`application/xml`.withCharset(HttpCharsets.`UTF-8`)).map {
case NodeSeq.Empty => throw Unmarshaller.NoContentException
- case x =>
+ case x =>
val bucket = (x \ "Bucket").text
val keyMarker = (x \ "KeyMarker").headOption.flatMap(x => Utils.emptyStringToOption(x.text))
val uploadIdMarker = (x \ "UploadIdMarker").headOption.flatMap(x => Utils.emptyStringToOption(x.text))
@@ -185,7 +185,7 @@ import scala.xml.NodeSeq
implicit val listPartsResultUnmarshaller: FromEntityUnmarshaller[ListPartsResult] = {
nodeSeqUnmarshaller(MediaTypes.`application/xml`.withCharset(HttpCharsets.`UTF-8`)).map {
case NodeSeq.Empty => throw Unmarshaller.NoContentException
- case x =>
+ case x =>
val bucket = (x \ "Bucket").text
val key = (x \ "Key").text
val uploadId = (x \ "UploadId").text
@@ -238,7 +238,7 @@ import scala.xml.NodeSeq
implicit val listObjectVersionsResultUnmarshaller: FromEntityUnmarshaller[ListObjectVersionsResult] = {
nodeSeqUnmarshaller(MediaTypes.`application/xml`.withCharset(HttpCharsets.`UTF-8`)).map {
case NodeSeq.Empty => throw Unmarshaller.NoContentException
- case x =>
+ case x =>
val bucket = (x \ "Bucket").text
val name = (x \ "Name").text
val prefix =
diff --git a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/S3Stream.scala b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/S3Stream.scala
index d782f214b..b715288e3 100644
--- a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/S3Stream.scala
+++ b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/S3Stream.scala
@@ -730,7 +730,7 @@ import scala.util.{ Failure, Success, Try }
// see https://docs.aws.amazon.com/AmazonS3/latest/API/API_CreateBucket.html
val maybeRegionPayload = region match {
case Region.US_EAST_1 => None
- case region =>
+ case region =>
Some(HttpRequests.createBucketRegionPayload(region)(ExecutionContext.parasitic))
}
diff --git a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/settings.scala b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/settings.scala
index 9431ee285..dcf72c3a6 100644
--- a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/settings.scala
+++ b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/settings.scala
@@ -535,12 +535,12 @@ object S3Settings {
c.getString("access-style") match {
case "virtual" => VirtualHostAccessStyle
case "path" => PathAccessStyle
- case other =>
+ case other =>
throw new IllegalArgumentException(s"'access-style' must be 'virtual' or 'path'. Got: [$other]")
}
case Some("true") | Some("force") => PathAccessStyle
case Some("false") => VirtualHostAccessStyle
- case Some(other) =>
+ case Some(other) =>
throw new IllegalArgumentException(
s"'path-style-access' must be 'false', 'true' or 'force'. Got: [$other]. Prefer using access-style instead.")
}
diff --git a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/scaladsl/S3WireMockBase.scala b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/scaladsl/S3WireMockBase.scala
index 7a6bb91f1..e77ff00c5 100644
--- a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/scaladsl/S3WireMockBase.scala
+++ b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/scaladsl/S3WireMockBase.scala
@@ -351,7 +351,7 @@ abstract class S3WireMockBase(_system: ActorSystem, val _wireMockServer: WireMoc
def mockMultipartUploadInitiationWithTransientError(expectedBody: String, faultOrStatus: Either[Fault, Int]): Unit = {
val scenarioName = "UploadWithTransientErrors"
val responseBuilder = faultOrStatus match {
- case Left(fault) => aResponse().withFault(fault)
+ case Left(fault) => aResponse().withFault(fault)
case Right(status) =>
aResponse()
.withStatus(status)
diff --git a/testkit/src/main/scala/org/apache/pekko/stream/connectors/testkit/CapturingAppender.scala b/testkit/src/main/scala/org/apache/pekko/stream/connectors/testkit/CapturingAppender.scala
index 76edf8428..bfb16c201 100644
--- a/testkit/src/main/scala/org/apache/pekko/stream/connectors/testkit/CapturingAppender.scala
+++ b/testkit/src/main/scala/org/apache/pekko/stream/connectors/testkit/CapturingAppender.scala
@@ -40,7 +40,7 @@ import org.slf4j.LoggerFactory
throw new IllegalStateException(
s"$CapturingAppenderName not defined for [${loggerNameOrRoot(loggerName)}] in logback-test.xml")
case appender: CapturingAppender => appender
- case other =>
+ case other =>
throw new IllegalStateException(s"Unexpected $CapturingAppender: $other")
}
}
diff --git a/testkit/src/main/scala/org/apache/pekko/stream/connectors/testkit/LogbackUtil.scala b/testkit/src/main/scala/org/apache/pekko/stream/connectors/testkit/LogbackUtil.scala
index 5aa368fb6..127b29462 100644
--- a/testkit/src/main/scala/org/apache/pekko/stream/connectors/testkit/LogbackUtil.scala
+++ b/testkit/src/main/scala/org/apache/pekko/stream/connectors/testkit/LogbackUtil.scala
@@ -54,7 +54,7 @@ import scala.annotation.tailrec
case ch.qos.logback.classic.Level.INFO_INT => Level.INFO
case ch.qos.logback.classic.Level.WARN_INT => Level.WARN
case ch.qos.logback.classic.Level.ERROR_INT => Level.ERROR
- case _ =>
+ case _ =>
throw new IllegalArgumentException("Level " + level.levelStr + ", " + level.levelInt + " is unknown.")
}
}
diff --git a/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala b/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala
index ab8b1dbda..b5cbf5bfe 100644
--- a/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala
+++ b/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala
@@ -119,7 +119,7 @@ private[unixdomainsocket] object UnixDomainSocketImpl {
newConnectionOp(sel, key)
}
key.attachment match {
- case null =>
+ case null =>
case sendReceiveContext: SendReceiveContext =>
sendReceiveContext.send match {
case SendRequested(buffer, sent) if keySelectable && key.isWritable =>
@@ -149,7 +149,7 @@ private[unixdomainsocket] object UnixDomainSocketImpl {
}
case _: SendRequested =>
key.interestOps(key.interestOps() | SelectionKey.OP_WRITE)
- case _: SendAvailable =>
+ case _: SendAvailable =>
case ShutdownRequested if key.isValid && !sendReceiveContext.isOutputShutdown =>
try {
if (sendReceiveContext.isInputShutdown) {
@@ -167,7 +167,7 @@ private[unixdomainsocket] object UnixDomainSocketImpl {
case _: IOException =>
}
case ShutdownRequested =>
- case CloseRequested =>
+ case CloseRequested =>
log.debug("Write-side is shutting down unconditionally")
key.cancel()
try {
@@ -217,7 +217,7 @@ private[unixdomainsocket] object UnixDomainSocketImpl {
case _: IOException =>
}
}
- case _: ReceiveAvailable =>
+ case _: ReceiveAvailable =>
case PendingReceiveAck(receiveQueue, receiveBuffer, pendingResult) if pendingResult.isCompleted =>
pendingResult.value.get match {
case Success(QueueOfferResult.Enqueued) =>
@@ -234,7 +234,7 @@ private[unixdomainsocket] object UnixDomainSocketImpl {
case _: PendingReceiveAck =>
}
case _: ((Selector, SelectionKey) => Unit) @unchecked =>
- case other =>
+ case other =>
log.warning("unexpected receive: [{}]", other)
}
}
diff --git a/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/StreamingXmlWriter.scala b/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/StreamingXmlWriter.scala
index 8a94c0b57..d20f8cf9d 100644
--- a/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/StreamingXmlWriter.scala
+++ b/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/StreamingXmlWriter.scala
@@ -98,7 +98,7 @@ import javax.xml.stream.XMLOutputFactory
case ProcessingInstruction(None, Some(data)) =>
output.writeProcessingInstruction(None.orNull, data)
case ProcessingInstruction(None, None) =>
- case Comment(text) =>
+ case Comment(text) =>
output.writeComment(text)
case CData(text) =>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
(pekko-connectors) 03/03: Add 'Reformat with scalafmt 3.9.10' to .git-blame-ignore-revs
Posted by fa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
commit edd3227a1337c3ed866f1c4a94e5e726853280de
Author: scala-steward-asf[bot] <14...@users.noreply.github.com>
AuthorDate: Mon Oct 27 11:53:55 2025 +0000
Add 'Reformat with scalafmt 3.9.10' to .git-blame-ignore-revs
---
.git-blame-ignore-revs | 3 +++
1 file changed, 3 insertions(+)
diff --git a/.git-blame-ignore-revs b/.git-blame-ignore-revs
index 7035888e3..097cee605 100644
--- a/.git-blame-ignore-revs
+++ b/.git-blame-ignore-revs
@@ -14,3 +14,6 @@ edba0bb6310dfbaf3830bcfb1dc37366b320d0fd
# Scala Steward: Reformat with scalafmt 3.9.3
f777a66544eb2a5230be19990ff3a9befa2efbe4
+
+# Scala Steward: Reformat with scalafmt 3.9.10
+39511fd0f2caf65baa36b7accfbefaec4be9951d
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
(pekko-connectors) 01/03: Update scalafmt-core to 3.9.10
Posted by fa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
commit ad9891106850d7361670656009ad37fd7485527e
Author: scala-steward-asf[bot] <14...@users.noreply.github.com>
AuthorDate: Mon Oct 27 11:53:38 2025 +0000
Update scalafmt-core to 3.9.10
---
.scalafmt.conf | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/.scalafmt.conf b/.scalafmt.conf
index e25ad0b52..d8a448658 100644
--- a/.scalafmt.conf
+++ b/.scalafmt.conf
@@ -1,4 +1,4 @@
-version = 3.9.3
+version = 3.9.10
runner.dialect = scala213
project.git = true
style = defaultWithAlign
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org