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 2026/09/13 13:59:44 UTC
(pekko-connectors) branch main updated: Chore: remove leftover Scala 2.12 immutable.Seq idioms (#1941)
This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 2c2837b4b Chore: remove leftover Scala 2.12 immutable.Seq idioms (#1941)
2c2837b4b is described below
commit 2c2837b4b8e7c0f0f3791dfec3780fd767ff59f6
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Sun Sep 13 14:59:37 2026 +0100
Chore: remove leftover Scala 2.12 immutable.Seq idioms (#1941)
Motivation:
The codebase still qualified Seq as immutable.Seq throughout, a Scala
2.12-era habit. Since 2.13 (and on Scala 3) scala.Seq is an alias for
scala.collection.immutable.Seq, so the qualifier and the supporting
scala.collection.immutable imports are redundant. Mirrors apache/pekko#3539.
Modification:
- Replace immutable.Seq with Seq across 120 Scala files.
- Drop the now-unused scala.collection.immutable and
scala.collection.immutable.Seq imports; reduce
scala.collection.{ immutable, mutable } to scala.collection.mutable
where only mutable remains in use.
- Keep scala.collection.immutable in files that still use
immutable.Iterable, immutable.Map etc. immutable.Iterable is left
unchanged because bare Iterable is scala.collection.Iterable.
- csv.md prose updated from immutable.Seq[String] to Seq[String].
Result:
No remaining immutable.Seq qualifiers in main, test or doc sources.
Types are identical (same erasure), so no API or binary change.
Tests:
- native scalafmt --mode diff-ref=origin/main
- sbt Test/compile (Scala 2.13.18) and sbt "++3.3.8" Test/compile
- sbt mimaReportBinaryIssues (all modules) passes, no filters needed
- sbt csv/test xml/test file/test text/test simple-codecs/test reference/test
- sbt docs/paradox
References:
None - follows apache/pekko#3539
---
.../connectors/amqp/AmqpConnectionProvider.scala | 25 +++++----
.../connectors/amqp/AmqpConnectorSettings.scala | 27 +++++-----
.../test/scala/docs/scaladsl/AmqpDocsSpec.scala | 9 ++--
.../amqp/scaladsl/AmqpConnectorsSpec.scala | 7 ++-
.../stream/connectors/awsspi/PekkoHttpClient.scala | 3 +-
.../connectors/cassandra/CqlSessionProvider.scala | 3 +-
.../cassandra/PekkoDiscoverySessionProvider.scala | 5 +-
.../cassandra/scaladsl/CassandraSession.scala | 5 +-
.../scala/docs/scaladsl/CassandraFlowSpec.scala | 9 ++--
.../scala/docs/scaladsl/CassandraSourceSpec.scala | 5 +-
.../cassandra/javadsl/CassandraSessionSpec.scala | 3 +-
.../cassandra/scaladsl/CassandraLifecycle.scala | 9 ++--
.../pekko/stream/connectors/couchbase/model.scala | 7 ++-
.../couchbase/scaladsl/DiscoverySupport.scala | 5 +-
.../scala/docs/scaladsl/CouchbaseFlowSpec.scala | 34 ++++++------
.../scala/docs/scaladsl/CouchbaseSourceSpec.scala | 1 -
.../couchbase/testing/CouchbaseSupport.scala | 1 -
.../docs/scaladsl/CouchbaseTestFlowSpec.scala | 1 -
.../stream/connectors/csv/impl/CsvToMapStage.scala | 22 ++++----
.../scala/docs/scaladsl/CsvFormattingSpec.scala | 4 +-
.../test/scala/docs/scaladsl/CsvParsingSpec.scala | 1 -
docs/src/main/paradox/data-transformations/csv.md | 2 +-
.../impl/ElasticsearchSimpleFlowStage.scala | 17 +++---
.../elasticsearch/impl/RestBulkApi.scala | 8 +--
.../elasticsearch/impl/RestBulkApiV5.scala | 4 +-
.../elasticsearch/impl/RestBulkApiV7.scala | 4 +-
.../elasticsearch/scaladsl/ElasticsearchFlow.scala | 41 +++++++-------
.../scaladsl/ElasticsearchConnectorBehaviour.scala | 21 ++++----
.../docs/scaladsl/ElasticsearchSpecUtils.scala | 3 +-
.../scala/docs/scaladsl/ElasticsearchV5Spec.scala | 5 +-
.../scala/docs/scaladsl/ElasticsearchV7Spec.scala | 5 +-
.../scaladsl/OpensearchConnectorBehaviour.scala | 21 ++++----
.../scala/docs/scaladsl/OpensearchV1Spec.scala | 5 +-
.../impl/ElasticsearchSimpleFlowStageTest.scala | 11 ++--
.../connectors/file/scaladsl/Directory.scala | 4 +-
.../connectors/file/scaladsl/LogRotatorSink.scala | 3 +-
.../test/scala/docs/scaladsl/DirectorySpec.scala | 9 ++--
.../scala/docs/scaladsl/LogRotatorSinkSpec.scala | 9 ++--
.../test/scala/docs/scaladsl/TarArchiveSpec.scala | 9 ++--
.../connectors/ftp/impl/CommonFtpOperations.scala | 5 +-
.../pekko/stream/connectors/ftp/impl/FtpLike.scala | 5 +-
.../connectors/ftp/impl/SftpOperations.scala | 5 +-
.../stream/connectors/ftp/CommonFtpStageSpec.scala | 7 ++-
.../bigquery/model/DatasetJsonProtocol.scala | 1 -
.../bigquery/model/JobJsonProtocol.scala | 1 -
.../bigquery/model/QueryJsonProtocol.scala | 1 -
.../bigquery/model/TableDataJsonProtocol.scala | 1 -
.../bigquery/model/TableJsonProtocol.scala | 1 -
.../bigquery/scaladsl/BigQueryTableData.scala | 1 -
.../bigquery/scaladsl/schema/ProductSchemas.scala | 1 -
.../src/test/scala/docs/scaladsl/BigQueryDoc.scala | 1 -
.../googlecloud/pubsub/impl/PubSubApi.scala | 8 +--
.../connectors/googlecloud/pubsub/model.scala | 14 ++---
.../googlecloud/pubsub/scaladsl/GooglePubSub.scala | 17 +++---
.../test/scala/docs/scaladsl/ExampleUsage.scala | 1 -
.../test/scala/docs/scaladsl/IntegrationSpec.scala | 1 -
.../google/AuthorizationHeaderSpec.scala | 1 -
.../googlecloud/pubsub/GooglePubSubSpec.scala | 1 -
.../connectors/googlecloud/pubsub/ModelSpec.scala | 1 -
.../googlecloud/pubsub/impl/PubSubApiSpec.scala | 1 -
.../googlecloud/storage/FailedUpload.scala | 1 -
.../src/test/scala/docs/scaladsl/FcmExamples.scala | 5 +-
.../stream/connectors/hbase/HTableSettings.scala | 17 +++---
.../test/scala/docs/scaladsl/HBaseStageSpec.scala | 13 +++--
.../huawei/pushkit/impl/PushKitSender.scala | 5 +-
.../test/scala/docs/scaladsl/PushKitExamples.scala | 3 +-
.../influxdb/impl/InfluxDbFlowStage.scala | 39 +++++++-------
.../influxdb/scaladsl/InfluxDbFlow.scala | 12 ++---
.../influxdb/scaladsl/InfluxDbSink.scala | 5 +-
.../connectors/ironmq/impl/IronMqClient.scala | 4 +-
.../test/scala/docs/scaladsl/IronMqDocsSpec.scala | 21 ++++----
.../scaladsl/JmsBufferedAckConnectorsSpec.scala | 4 +-
.../scala/docs/scaladsl/JmsConnectorsSpec.scala | 22 ++++----
.../docs/scaladsl/JmsIbmmqConnectorsSpec.scala | 5 +-
.../scala/docs/scaladsl/JmsTxConnectorsSpec.scala | 4 +-
.../scaladsl/JmsBufferedAckConnectorsSpec.scala | 4 +-
.../scala/docs/scaladsl/JmsConnectorsSpec.scala | 21 ++++----
.../docs/scaladsl/JmsIbmmqConnectorsSpec.scala | 5 +-
.../scala/docs/scaladsl/JmsTxConnectorsSpec.scala | 4 +-
.../kinesis/scaladsl/KinesisSchedulerSource.scala | 3 +-
.../test/scala/docs/scaladsl/MongoSourceSpec.scala | 1 -
.../mqtt/streaming/impl/BehaviorRunner.scala | 2 -
.../mqtt/streaming/impl/ClientState.scala | 1 -
.../mqtt/streaming/impl/QueueOfferState.scala | 1 -
.../mqtt/streaming/impl/ServerState.scala | 1 -
.../pekko/stream/connectors/mqtt/settings.scala | 11 ++--
.../test/scala/docs/scaladsl/MqttSourceSpec.scala | 1 -
.../pekko/stream/connectors/mqttv5/model.scala | 3 +-
.../pekko/stream/connectors/mqttv5/settings.scala | 3 +-
.../orientdb/impl/OrientDbFlowStage.scala | 15 +++---
.../orientdb/scaladsl/OrientDbFlow.scala | 15 +++---
.../orientdb/scaladsl/OrientDbSink.scala | 5 +-
.../test/scala/docs/scaladsl/OrientDbSpec.scala | 3 +-
.../reference/impl/ReferenceSourceStage.scala | 3 +-
.../pekko/stream/connectors/reference/model.scala | 9 ++--
.../reference/testkit/MessageFactory.scala | 3 +-
.../test/scala/docs/scaladsl/ReferenceSpec.scala | 13 +++--
.../pekko/stream/connectors/s3/S3Headers.scala | 2 -
.../s3/headers/ServerSideEncryption.scala | 18 +++----
.../stream/connectors/s3/impl/HttpRequests.scala | 1 -
.../pekko/stream/connectors/s3/impl/S3Stream.scala | 62 +++++++++++-----------
.../apache/pekko/stream/connectors/s3/model.scala | 4 +-
.../connectors/s3/impl/MarshallingSpec.scala | 1 -
.../connectors/s3/scaladsl/S3IntegrationSpec.scala | 2 +-
.../scala/docs/scaladsl/RecordIOFramingSpec.scala | 1 -
.../docs/scaladsl/SlickWithTryResultSpec.scala | 4 +-
.../connectors/solr/impl/SolrFlowStage.scala | 25 +++++----
.../stream/connectors/solr/javadsl/SolrFlow.scala | 13 +++--
.../stream/connectors/solr/scaladsl/SolrFlow.scala | 18 +++----
.../stream/connectors/solr/scaladsl/SolrSink.scala | 7 ++-
solr/src/test/scala/docs/scaladsl/SolrSpec.scala | 5 +-
.../stream/connectors/sqs/SqsSourceSettings.scala | 21 ++++----
.../connectors/sqs/scaladsl/SqsAckFlow.scala | 3 +-
.../test/scala/docs/scaladsl/SqsSourceSpec.scala | 5 +-
.../test/scala/docs/scaladsl/EventSourceSpec.scala | 3 +-
.../docs/scaladsl/CharsetCodingFlowsDoc.scala | 3 +-
.../text/scaladsl/CharsetCodingFlowsSpec.scala | 3 +-
.../stream/connectors/xml/impl/Subslice.scala | 4 +-
.../pekko/stream/connectors/xml/impl/Subtree.scala | 4 +-
.../connectors/xml/scaladsl/XmlParsing.scala | 6 +--
.../scala/docs/scaladsl/XmlProcessingSpec.scala | 2 +-
121 files changed, 410 insertions(+), 536 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 8509b3bd9..490cadeed 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
@@ -22,7 +22,6 @@ import pekko.annotation.DoNotInherit
import com.rabbitmq.client.{ Address, Connection, ConnectionFactory, ExceptionHandler }
import scala.annotation.tailrec
-import scala.collection.immutable
import scala.jdk.CollectionConverters._
/**
@@ -72,7 +71,7 @@ object AmqpUriConnectionProvider {
}
final class AmqpDetailsConnectionProvider private (
- val hostAndPortList: immutable.Seq[(String, Int)],
+ val hostAndPortList: Seq[(String, Int)],
val credentials: Option[AmqpCredentials] = None,
val virtualHost: Option[String] = None,
val sslConfiguration: Option[AmqpSSLConfiguration] = None,
@@ -87,9 +86,9 @@ final class AmqpDetailsConnectionProvider private (
val connectionName: Option[String] = None) extends AmqpConnectionProvider {
def withHostAndPort(host: String, port: Int): AmqpDetailsConnectionProvider =
- copy(hostAndPortList = immutable.Seq(host -> port))
+ copy(hostAndPortList = Seq(host -> port))
- def withHostsAndPorts(hostAndPorts: immutable.Seq[(String, Int)]): AmqpDetailsConnectionProvider =
+ def withHostsAndPorts(hostAndPorts: Seq[(String, Int)]): AmqpDetailsConnectionProvider =
copy(hostAndPortList = hostAndPorts)
def withHostsAndPorts(
@@ -162,7 +161,7 @@ final class AmqpDetailsConnectionProvider private (
factory.newConnection(hostAndPortList.map(hp => new Address(hp._1, hp._2)).asJava, connectionName.orNull)
}
- private def copy(hostAndPortList: immutable.Seq[(String, Int)] = hostAndPortList,
+ private def copy(hostAndPortList: Seq[(String, Int)] = hostAndPortList,
credentials: Option[AmqpCredentials] = credentials,
virtualHost: Option[String] = virtualHost,
sslConfiguration: Option[AmqpSSLConfiguration] = sslConfiguration,
@@ -211,9 +210,9 @@ final class AmqpDetailsConnectionProvider private (
object AmqpDetailsConnectionProvider {
def apply(host: String, port: Int): AmqpDetailsConnectionProvider =
- new AmqpDetailsConnectionProvider(immutable.Seq(host -> port))
+ new AmqpDetailsConnectionProvider(Seq(host -> port))
- def apply(hostAndPorts: immutable.Seq[(String, Int)]): AmqpDetailsConnectionProvider =
+ def apply(hostAndPorts: Seq[(String, Int)]): AmqpDetailsConnectionProvider =
new AmqpDetailsConnectionProvider(hostAndPorts)
/**
@@ -312,7 +311,7 @@ object AmqpSSLConfiguration {
* If empty, it defaults to the host and port in the underlying factory.
*/
final class AmqpConnectionFactoryConnectionProvider private (val factory: ConnectionFactory,
- private val hostAndPorts: immutable.Seq[(String, Int)] =
+ private val hostAndPorts: Seq[(String, Int)] =
Nil)
extends AmqpConnectionProvider {
@@ -320,16 +319,16 @@ final class AmqpConnectionFactoryConnectionProvider private (val factory: Connec
* @return A list of hosts and ports for this AMQP connection factory.
* Uses host and port from the underlying factory if hostAndPorts was left out on construction.
*/
- def hostAndPortList: immutable.Seq[(String, Int)] =
+ def hostAndPortList: Seq[(String, Int)] =
if (hostAndPorts.isEmpty)
- immutable.Seq((factory.getHost, factory.getPort))
+ Seq((factory.getHost, factory.getPort))
else
hostAndPorts.toList
def withHostAndPort(host: String, port: Int): AmqpConnectionFactoryConnectionProvider =
- copy(hostAndPorts = immutable.Seq(host -> port))
+ copy(hostAndPorts = Seq(host -> port))
- def withHostsAndPorts(hostAndPorts: immutable.Seq[(String, Int)]): AmqpConnectionFactoryConnectionProvider =
+ def withHostsAndPorts(hostAndPorts: Seq[(String, Int)]): AmqpConnectionFactoryConnectionProvider =
copy(hostAndPorts = hostAndPorts)
/**
@@ -343,7 +342,7 @@ final class AmqpConnectionFactoryConnectionProvider private (val factory: Connec
factory.newConnection(hostAndPortList.map(hp => new Address(hp._1, hp._2)).asJava)
}
- private def copy(hostAndPorts: immutable.Seq[(String, Int)]) =
+ private def copy(hostAndPorts: Seq[(String, Int)]) =
new AmqpConnectionFactoryConnectionProvider(factory, hostAndPorts)
override def toString: String =
diff --git a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectorSettings.scala b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectorSettings.scala
index 92506b846..d38330d3d 100644
--- a/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectorSettings.scala
+++ b/amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectorSettings.scala
@@ -16,7 +16,6 @@ package org.apache.pekko.stream.connectors.amqp
import org.apache.pekko
import pekko.annotation.InternalApi
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
import scala.jdk.DurationConverters._
@@ -27,7 +26,7 @@ import scala.jdk.DurationConverters._
@InternalApi
sealed trait AmqpConnectorSettings {
def connectionProvider: AmqpConnectionProvider
- def declarations: immutable.Seq[Declaration]
+ def declarations: Seq[Declaration]
def reuseByteArray: Boolean
}
@@ -36,7 +35,7 @@ sealed trait AmqpSourceSettings extends AmqpConnectorSettings
final class NamedQueueSourceSettings private (
val connectionProvider: AmqpConnectionProvider,
val queue: String,
- val declarations: immutable.Seq[Declaration] = immutable.Seq.empty,
+ val declarations: Seq[Declaration] = Seq.empty,
val noLocal: Boolean = false,
val exclusive: Boolean = false,
val ackRequired: Boolean = true,
@@ -45,9 +44,9 @@ final class NamedQueueSourceSettings private (
val reuseByteArray: Boolean = false) extends AmqpSourceSettings {
def withDeclaration(declaration: Declaration): NamedQueueSourceSettings =
- copy(declarations = immutable.Seq(declaration))
+ copy(declarations = Seq(declaration))
- def withDeclarations(declarations: immutable.Seq[Declaration]): NamedQueueSourceSettings =
+ def withDeclarations(declarations: Seq[Declaration]): NamedQueueSourceSettings =
copy(declarations = declarations)
/**
@@ -84,7 +83,7 @@ final class NamedQueueSourceSettings private (
def withReuseByteArray(reuseByteArray: Boolean): NamedQueueSourceSettings =
copy(reuseByteArray = reuseByteArray)
- private def copy(declarations: immutable.Seq[Declaration] = declarations,
+ private def copy(declarations: Seq[Declaration] = declarations,
noLocal: Boolean = noLocal,
exclusive: Boolean = exclusive,
ackRequired: Boolean = ackRequired,
@@ -130,14 +129,14 @@ object NamedQueueSourceSettings {
final class TemporaryQueueSourceSettings private (
val connectionProvider: AmqpConnectionProvider,
val exchange: String,
- val declarations: immutable.Seq[Declaration] = Nil,
+ val declarations: Seq[Declaration] = Nil,
val routingKey: Option[String] = None,
val reuseByteArray: Boolean = false) extends AmqpSourceSettings {
def withDeclaration(declaration: Declaration): TemporaryQueueSourceSettings =
- copy(declarations = immutable.Seq(declaration))
+ copy(declarations = Seq(declaration))
- def withDeclarations(declarations: immutable.Seq[Declaration]): TemporaryQueueSourceSettings =
+ def withDeclarations(declarations: Seq[Declaration]): TemporaryQueueSourceSettings =
copy(declarations = declarations)
/**
@@ -151,7 +150,7 @@ final class TemporaryQueueSourceSettings private (
def withReuseByteArray(reuseByteArray: Boolean): TemporaryQueueSourceSettings =
copy(reuseByteArray = reuseByteArray)
- private def copy(declarations: immutable.Seq[Declaration] = declarations,
+ private def copy(declarations: Seq[Declaration] = declarations,
routingKey: Option[String] = routingKey,
reuseByteArray: Boolean = reuseByteArray) =
new TemporaryQueueSourceSettings(connectionProvider, exchange, declarations = declarations,
@@ -218,7 +217,7 @@ final class AmqpWriteSettings private (
val connectionProvider: AmqpConnectionProvider,
val exchange: Option[String] = None,
val routingKey: Option[String] = None,
- val declarations: immutable.Seq[Declaration] = Nil,
+ val declarations: Seq[Declaration] = Nil,
val bufferSize: Int = 10,
val confirmationTimeout: FiniteDuration = 100.millis,
val reuseByteArray: Boolean = false) extends AmqpConnectorSettings {
@@ -230,9 +229,9 @@ final class AmqpWriteSettings private (
copy(routingKey = Some(routingKey))
def withDeclaration(declaration: Declaration): AmqpWriteSettings =
- copy(declarations = immutable.Seq(declaration))
+ copy(declarations = Seq(declaration))
- def withDeclarations(declarations: immutable.Seq[Declaration]): AmqpWriteSettings =
+ def withDeclarations(declarations: Seq[Declaration]): AmqpWriteSettings =
copy(declarations = declarations)
def withReuseByteArray(reuseByteArray: Boolean): AmqpWriteSettings =
@@ -259,7 +258,7 @@ final class AmqpWriteSettings private (
private def copy(connectionProvider: AmqpConnectionProvider = connectionProvider,
exchange: Option[String] = exchange,
routingKey: Option[String] = routingKey,
- declarations: immutable.Seq[Declaration] = declarations,
+ declarations: Seq[Declaration] = declarations,
bufferSize: Int = bufferSize,
confirmationTimeout: FiniteDuration = confirmationTimeout,
reuseByteArray: Boolean = reuseByteArray) =
diff --git a/amqp/src/test/scala/docs/scaladsl/AmqpDocsSpec.scala b/amqp/src/test/scala/docs/scaladsl/AmqpDocsSpec.scala
index 3220ba8c4..90743573a 100644
--- a/amqp/src/test/scala/docs/scaladsl/AmqpDocsSpec.scala
+++ b/amqp/src/test/scala/docs/scaladsl/AmqpDocsSpec.scala
@@ -26,7 +26,6 @@ import pekko.util.ByteString
import scala.concurrent.duration._
import scala.concurrent.{ Await, Future, Promise }
-import scala.collection.immutable
/**
* Needs a local running AMQP server on the default port with no password.
@@ -46,7 +45,7 @@ class AmqpDocsSpec extends AmqpSpec {
// use a list of host/port pairs where one is normally invalid, but
// it should still work as expected,
val connectionProvider =
- AmqpDetailsConnectionProvider("invalid", 5673).withHostsAndPorts(immutable.Seq("localhost" -> 5672))
+ AmqpDetailsConnectionProvider("invalid", 5673).withHostsAndPorts(Seq("localhost" -> 5672))
// #queue-declaration
val queueName = "amqp-conn-it-spec-simple-queue-" + System.currentTimeMillis()
@@ -76,7 +75,7 @@ class AmqpDocsSpec extends AmqpSpec {
.withAckRequired(false),
bufferSize = 10)
- val result: Future[immutable.Seq[ReadResult]] =
+ val result: Future[Seq[ReadResult]] =
amqpSource
.take(input.size)
.runWith(Sink.seq)
@@ -198,7 +197,7 @@ class AmqpDocsSpec extends AmqpSpec {
.withDeclaration(queueDeclaration),
bufferSize = 10)
- val result: Future[immutable.Seq[ReadResult]] = amqpSource
+ val result: Future[Seq[ReadResult]] = amqpSource
.mapAsync(1)(businessLogic)
.mapAsync(1)(cm => cm.ack().map(_ => cm.message))
.take(input.size)
@@ -227,7 +226,7 @@ class AmqpDocsSpec extends AmqpSpec {
// #create-source-withoutautoack
- val nackedResults: Future[immutable.Seq[ReadResult]] = amqpSource
+ val nackedResults: Future[Seq[ReadResult]] = amqpSource
.mapAsync(1)(businessLogic)
.take(input.size)
.mapAsync(1)(cm => cm.nack(multiple = false, requeue = true).map(_ => cm.message))
diff --git a/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpConnectorsSpec.scala b/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpConnectorsSpec.scala
index c58542cdb..7ac4dd938 100644
--- a/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpConnectorsSpec.scala
+++ b/amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpConnectorsSpec.scala
@@ -30,7 +30,6 @@ import org.scalatestplus.scalacheck.ScalaCheckDrivenPropertyChecks
import scala.concurrent.Await
import scala.concurrent.duration._
-import scala.collection.immutable
/**
* Needs a local running AMQP server on the default port with no password.
@@ -66,7 +65,7 @@ class AmqpConnectorsSpec extends AmqpSpec with ScalaCheckDrivenPropertyChecks {
forAll { (reuseByteArray: Boolean) =>
val connectionProvider =
AmqpDetailsConnectionProvider("invalid", 5673)
- .withHostsAndPorts(immutable.Seq("localhost" -> 5672))
+ .withHostsAndPorts(Seq("localhost" -> 5672))
.withCredentials(AmqpCredentials("guest", "guest1"))
val queueName = "amqp-conn-it-spec-simple-queue-" + System.currentTimeMillis()
@@ -396,12 +395,12 @@ class AmqpConnectorsSpec extends AmqpSpec with ScalaCheckDrivenPropertyChecks {
val amqpSink = AmqpSink(
AmqpWriteSettings(connectionProvider)
.withExchange(exchangeName)
- .withDeclarations(immutable.Seq(exchangeDeclaration, queueDeclaration, bindingDeclaration))
+ .withDeclarations(Seq(exchangeDeclaration, queueDeclaration, bindingDeclaration))
.withReuseByteArray(reuseByteArray))
val amqpSource = AmqpSource.atMostOnceSource(
NamedQueueSourceSettings(connectionProvider, queueName)
- .withDeclarations(immutable.Seq(exchangeDeclaration, queueDeclaration, bindingDeclaration))
+ .withDeclarations(Seq(exchangeDeclaration, queueDeclaration, bindingDeclaration))
.withReuseByteArray(reuseByteArray),
bufferSize = 10)
diff --git a/aws-spi-pekko-http/src/main/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClient.scala b/aws-spi-pekko-http/src/main/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClient.scala
index a1dc973d0..528d724a7 100644
--- a/aws-spi-pekko-http/src/main/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClient.scala
+++ b/aws-spi-pekko-http/src/main/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClient.scala
@@ -41,7 +41,6 @@ import software.amazon.awssdk.http.async._
import software.amazon.awssdk.http.{ SdkHttpConfigurationOption, SdkHttpRequest }
import software.amazon.awssdk.utils.AttributeMap
-import scala.collection.immutable
import scala.concurrent.duration.Duration
import scala.concurrent.{ Await, ExecutionContext }
import scala.jdk.DurationConverters._
@@ -132,7 +131,7 @@ object PekkoHttpClient {
// This method converts the headers to Pekko-http headers, drops content-length (returning its value separately),
// and returns content-type separately
private[awsspi] def convertHeaders(
- headers: java.util.Map[String, java.util.List[String]]): (Option[HttpHeader], immutable.Seq[HttpHeader],
+ headers: java.util.Map[String, java.util.List[String]]): (Option[HttpHeader], Seq[HttpHeader],
Option[Long]) = {
val headersAsScala = {
val builder = collection.mutable.Map.newBuilder[String, java.util.List[String]]
diff --git a/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/CqlSessionProvider.scala b/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/CqlSessionProvider.scala
index e4dabdacc..93c0c0001 100644
--- a/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/CqlSessionProvider.scala
+++ b/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/CqlSessionProvider.scala
@@ -18,7 +18,6 @@ import pekko.actor.{ ActorSystem, ClassicActorSystemProvider, ExtendedActorSyste
import com.datastax.oss.driver.api.core.CqlSession
import com.typesafe.config.{ Config, ConfigFactory }
-import scala.collection.immutable
import scala.concurrent.{ ExecutionContext, Future }
import scala.jdk.FutureConverters._
import scala.util.Failure
@@ -77,7 +76,7 @@ object CqlSessionProvider {
val className = config.getString("session-provider")
val dynamicAccess = system.asInstanceOf[ExtendedActorSystem].dynamicAccess
val clazz = dynamicAccess.getClassFor[CqlSessionProvider](className).get
- def instantiate(args: immutable.Seq[(Class[?], AnyRef)]) =
+ def instantiate(args: Seq[(Class[?], AnyRef)]) =
dynamicAccess.createInstanceFor[CqlSessionProvider](clazz, args)
val params = List((classOf[ActorSystem], system), (classOf[Config], config))
diff --git a/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/PekkoDiscoverySessionProvider.scala b/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/PekkoDiscoverySessionProvider.scala
index 4c1ae7443..c2ba4666a 100644
--- a/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/PekkoDiscoverySessionProvider.scala
+++ b/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/PekkoDiscoverySessionProvider.scala
@@ -20,7 +20,6 @@ import pekko.discovery.Discovery
import com.datastax.oss.driver.api.core.CqlSession
import com.typesafe.config.{ Config, ConfigFactory }
-import scala.collection.immutable
import scala.concurrent.duration.FiniteDuration
import scala.concurrent.{ ExecutionContext, Future }
import scala.jdk.DurationConverters._
@@ -83,7 +82,7 @@ private[cassandra] object PekkoDiscoverySessionProvider {
* Expect a `service` section in Config and use Pekko Discovery to read the addresses for `name` within `lookup-timeout`.
*/
private def readNodes(config: Config)(implicit system: ActorSystem,
- ec: ExecutionContext): Future[immutable.Seq[String]] = {
+ ec: ExecutionContext): Future[Seq[String]] = {
val serviceConfig = config.getConfig("service-discovery")
val serviceName = serviceConfig.getString("name")
val lookupTimeout = serviceConfig.getDuration("lookup-timeout").toScala
@@ -96,7 +95,7 @@ private[cassandra] object PekkoDiscoverySessionProvider {
private def readNodes(
serviceName: String,
lookupTimeout: FiniteDuration)(
- implicit system: ActorSystem, ec: ExecutionContext): Future[immutable.Seq[String]] = {
+ implicit system: ActorSystem, ec: ExecutionContext): Future[Seq[String]] = {
Discovery(system).discovery.lookup(serviceName, lookupTimeout).map { resolved =>
resolved.addresses.map { target =>
target.host + ":" + target.port.getOrElse {
diff --git a/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraSession.scala b/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraSession.scala
index c2bba9034..a625596af 100644
--- a/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraSession.scala
+++ b/cassandra/src/main/scala/org/apache/pekko/stream/connectors/cassandra/scaladsl/CassandraSession.scala
@@ -26,7 +26,6 @@ import com.datastax.oss.driver.api.core.CqlSession
import com.datastax.oss.driver.api.core.cql._
import com.datastax.oss.driver.api.core.servererrors.InvalidQueryException
-import scala.collection.immutable
import scala.concurrent.{ ExecutionContext, Future }
import scala.jdk.FutureConverters._
import scala.util.control.NonFatal
@@ -269,7 +268,7 @@ final class CassandraSession(system: pekko.actor.ActorSystem,
*
* The returned `Future` is completed with the found rows.
*/
- def selectAll(stmt: Statement[?]): Future[immutable.Seq[Row]] = {
+ def selectAll(stmt: Statement[?]): Future[Seq[Row]] = {
select(stmt)
.runWith(Sink.seq)
}
@@ -283,7 +282,7 @@ final class CassandraSession(system: pekko.actor.ActorSystem,
*
* The returned `Future` is completed with the found rows.
*/
- def selectAll(stmt: String, bindValues: AnyRef*): Future[immutable.Seq[Row]] = {
+ def selectAll(stmt: String, bindValues: AnyRef*): Future[Seq[Row]] = {
bind(stmt, bindValues).flatMap(bs => selectAll(bs))
}
diff --git a/cassandra/src/test/scala/docs/scaladsl/CassandraFlowSpec.scala b/cassandra/src/test/scala/docs/scaladsl/CassandraFlowSpec.scala
index 3ba096514..c6b27f66e 100644
--- a/cassandra/src/test/scala/docs/scaladsl/CassandraFlowSpec.scala
+++ b/cassandra/src/test/scala/docs/scaladsl/CassandraFlowSpec.scala
@@ -26,7 +26,6 @@ import pekko.stream.connectors.cassandra.scaladsl.{
import pekko.stream.scaladsl.{ Sink, Source, SourceWithContext }
import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import scala.collection.immutable
import scala.concurrent.Future
class CassandraFlowSpec extends CassandraSpecBase(ActorSystem("CassandraFlowSpec")) {
@@ -85,12 +84,12 @@ class CassandraFlowSpec extends CassandraSpecBase(ActorSystem("CassandraFlowSpec
case class Person(id: Int, name: String, city: String)
val persons =
- immutable.Seq(Person(12, "John", "London"), Person(43, "Umberto", "Roma"), Person(56, "James", "Chicago"))
+ Seq(Person(12, "John", "London"), Person(43, "Umberto", "Roma"), Person(56, "James", "Chicago"))
val statementBinder: (Person, PreparedStatement) => BoundStatement =
(person, preparedStatement) => preparedStatement.bind(Int.box(person.id), person.name, person.city)
- val written: Future[immutable.Seq[Person]] = Source(persons)
+ val written: Future[Seq[Person]] = Source(persons)
.via(
CassandraFlow.create(CassandraWriteSettings.defaults,
s"INSERT INTO $table(id, name, city) VALUES (?, ?, ?)",
@@ -125,7 +124,7 @@ class CassandraFlowSpec extends CassandraSpecBase(ActorSystem("CassandraFlowSpec
def ack(): Future[Done] = Future.successful(Done)
}
val persons =
- immutable.Seq(Person(12, "John", "London") -> AckHandle(12),
+ Seq(Person(12, "John", "London") -> AckHandle(12),
Person(43, "Umberto", "Roma") -> AckHandle(43),
Person(56, "James", "Chicago") -> AckHandle(56))
@@ -173,7 +172,7 @@ class CassandraFlowSpec extends CassandraSpecBase(ActorSystem("CassandraFlowSpec
case class Person(id: Int, name: String, city: String)
val persons =
- immutable.Seq(Person(12, "John", "London"), Person(43, "Umberto", "Roma"), Person(56, "James", "Chicago"))
+ Seq(Person(12, "John", "London"), Person(43, "Umberto", "Roma"), Person(56, "James", "Chicago"))
val written = Source(persons)
.via(
CassandraFlow.createBatch(
diff --git a/cassandra/src/test/scala/docs/scaladsl/CassandraSourceSpec.scala b/cassandra/src/test/scala/docs/scaladsl/CassandraSourceSpec.scala
index 5b5c1c39d..621137f0b 100644
--- a/cassandra/src/test/scala/docs/scaladsl/CassandraSourceSpec.scala
+++ b/cassandra/src/test/scala/docs/scaladsl/CassandraSourceSpec.scala
@@ -20,7 +20,6 @@ import pekko.stream.connectors.cassandra.scaladsl.{ CassandraSession, CassandraS
import pekko.stream.scaladsl.Sink
import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
-import scala.collection.immutable
import scala.concurrent.Future
class CassandraSourceSpec extends CassandraSpecBase(ActorSystem("CassandraSourceSpec")) {
@@ -73,7 +72,7 @@ class CassandraSourceSpec extends CassandraSpecBase(ActorSystem("CassandraSource
// #cql
import org.apache.pekko.stream.connectors.cassandra.scaladsl.CassandraSource
- val ids: Future[immutable.Seq[Int]] =
+ val ids: Future[Seq[Int]] =
CassandraSource(s"SELECT id FROM $intTable").map(row => row.getInt("id")).runWith(Sink.seq)
// #cql
@@ -95,7 +94,7 @@ class CassandraSourceSpec extends CassandraSpecBase(ActorSystem("CassandraSource
val stmt = SimpleStatement.newInstance(s"SELECT * FROM $intTable").setPageSize(20)
- val rows: Future[immutable.Seq[Row]] = CassandraSource(stmt).runWith(Sink.seq)
+ val rows: Future[Seq[Row]] = CassandraSource(stmt).runWith(Sink.seq)
// #statement
rows.futureValue.map(_.getInt("id")) must contain theSameElementsAs data
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 646647c39..7705a12d7 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
@@ -30,7 +30,6 @@ import pekko.stream.testkit.scaladsl.TestSink
import com.datastax.oss.driver.api.core.cql.Row
-import scala.collection.immutable
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
@@ -58,7 +57,7 @@ final class CassandraSessionSpec extends CassandraSpecBase(ActorSystem("Cassandr
|)
|""".stripMargin)
_ <- executeCql(
- immutable.Seq(
+ Seq(
s"INSERT INTO $dataTable (partition, key, count) VALUES ('A', 'a', 1);",
s"INSERT INTO $dataTable (partition, key, count) VALUES ('A', 'b', 2);",
s"INSERT INTO $dataTable (partition, key, count) VALUES ('A', 'c', 3);",
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 c91eaff2d..69cddcfaf 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
@@ -22,7 +22,6 @@ import com.datastax.oss.driver.api.core.cql._
import org.scalatest._
import org.scalatest.concurrent.{ PatienceConfiguration, ScalaFutures }
-import scala.collection.immutable
import scala.concurrent.duration.DurationInt
import scala.concurrent.{ Await, ExecutionContext, Future }
import scala.jdk.CollectionConverters._
@@ -32,7 +31,7 @@ import scala.util.control.NonFatal
trait CassandraLifecycleBase {
def lifecycleSession: CassandraSession
- def execute(session: CassandraSession, statements: immutable.Seq[BatchableStatement[?]]): Future[Done] = {
+ def execute(session: CassandraSession, statements: Seq[BatchableStatement[?]]): Future[Done] = {
val batch = new BatchStatementBuilder(BatchType.LOGGED)
statements.foreach { stmt =>
batch.addStatement(stmt)
@@ -40,7 +39,7 @@ trait CassandraLifecycleBase {
session.executeWriteBatch(batch.build())
}
- def executeCql(session: CassandraSession, statements: immutable.Seq[String]): Future[Done] = {
+ def executeCql(session: CassandraSession, statements: Seq[String]): Future[Done] = {
execute(session, statements.map(stmt => SimpleStatement.newInstance(stmt)))
}
@@ -62,9 +61,9 @@ trait CassandraLifecycleBase {
def dropKeyspace(name: String): Future[Done] = withSchemaMetadataDisabled(dropKeyspace(lifecycleSession, name))
- def execute(statements: immutable.Seq[BatchableStatement[?]]): Future[Done] = execute(lifecycleSession, statements)
+ def execute(statements: Seq[BatchableStatement[?]]): Future[Done] = execute(lifecycleSession, statements)
- def executeCql(statements: immutable.Seq[String]): Future[Done] = executeCql(lifecycleSession, statements)
+ def executeCql(statements: Seq[String]): Future[Done] = executeCql(lifecycleSession, statements)
def executeCqlList(statements: java.util.List[String]): CompletionStage[Done] =
executeCql(lifecycleSession, statements.asScala.toList).asJava
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 e8182a07a..d50b0efa7 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
@@ -23,7 +23,6 @@ import com.couchbase.client.java.env.CouchbaseEnvironment
import com.couchbase.client.java.{ PersistTo, ReplicateTo }
import com.typesafe.config.Config
-import scala.collection.immutable
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
@@ -182,7 +181,7 @@ object CouchbaseSessionSettings {
final class CouchbaseSessionSettings private (
val username: String,
val password: String,
- val nodes: immutable.Seq[String],
+ val nodes: Seq[String],
val environment: Option[CouchbaseEnvironment],
val enrichAsync: CouchbaseSessionSettings => Future[CouchbaseSessionSettings]) {
@@ -195,7 +194,7 @@ final class CouchbaseSessionSettings private (
def withNodes(nodes: String): CouchbaseSessionSettings =
copy(nodes = nodes :: Nil)
- def withNodes(nodes: immutable.Seq[String]): CouchbaseSessionSettings =
+ def withNodes(nodes: Seq[String]): CouchbaseSessionSettings =
copy(nodes = nodes)
/** Java API */
@@ -231,7 +230,7 @@ final class CouchbaseSessionSettings private (
private def copy(
username: String = username,
password: String = password,
- nodes: immutable.Seq[String] = nodes,
+ nodes: Seq[String] = nodes,
environment: Option[CouchbaseEnvironment] = environment,
enrichAsync: CouchbaseSessionSettings => Future[CouchbaseSessionSettings] = enrichAsync)
: CouchbaseSessionSettings =
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 5f047c304..61e067abb 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
@@ -21,7 +21,6 @@ import pekko.discovery.Discovery
import pekko.stream.connectors.couchbase.CouchbaseSessionSettings
import com.typesafe.config.Config
-import scala.collection.immutable
import scala.concurrent.{ ExecutionContext, Future }
import scala.concurrent.duration.FiniteDuration
@@ -39,7 +38,7 @@ sealed class DiscoverySupport private {
*/
private def readNodes(
serviceName: String,
- lookupTimeout: FiniteDuration)(implicit system: ClassicActorSystemProvider): Future[immutable.Seq[String]] = {
+ lookupTimeout: FiniteDuration)(implicit system: ClassicActorSystemProvider): Future[Seq[String]] = {
implicit val ec: ExecutionContext = system.classicSystem.dispatcher
val discovery = Discovery(system).discovery
discovery.lookup(serviceName, lookupTimeout).map { resolved =>
@@ -50,7 +49,7 @@ sealed class DiscoverySupport private {
/**
* Expect a `service` section in Config and use Pekko Discovery to read the addresses for `name` within `lookup-timeout`.
*/
- private def readNodes(config: Config)(implicit system: ClassicActorSystemProvider): Future[immutable.Seq[String]] =
+ private def readNodes(config: Config)(implicit system: ClassicActorSystemProvider): Future[Seq[String]] =
if (config.hasPath("service")) {
val serviceName = config.getString("service.name")
val lookupTimeout = config.getDuration("service.lookup-timeout").toScala
diff --git a/couchbase/src/test/scala/docs/scaladsl/CouchbaseFlowSpec.scala b/couchbase/src/test/scala/docs/scaladsl/CouchbaseFlowSpec.scala
index 95254c2d7..b5eed0ef9 100644
--- a/couchbase/src/test/scala/docs/scaladsl/CouchbaseFlowSpec.scala
+++ b/couchbase/src/test/scala/docs/scaladsl/CouchbaseFlowSpec.scala
@@ -37,8 +37,6 @@ import com.couchbase.client.java.{ PersistTo, ReplicateTo }
import pekko.stream.testkit.scaladsl.StreamTestKit._
import com.couchbase.client.java.document.{ BinaryDocument, RawJsonDocument, StringDocument }
-import scala.collection.immutable
-import scala.collection.immutable.Seq
import scala.concurrent.duration._
import scala.concurrent.Future
@@ -177,7 +175,7 @@ class CouchbaseFlowSpec
bulkUpsertResult.futureValue
- val resultsAsFuture: Future[immutable.Seq[RawJsonDocument]] =
+ val resultsAsFuture: Future[Seq[RawJsonDocument]] =
Source(sampleSequence.map(_.id))
.via(CouchbaseFlow.fromId(sessionSettings, bucketName, classOf[RawJsonDocument]))
.runWith(Sink.seq)
@@ -195,9 +193,9 @@ class CouchbaseFlowSpec
bulkUpsertResult.futureValue
// #fromId
- val ids = immutable.Seq("First", "Second", "Third", "Fourth")
+ val ids = Seq("First", "Second", "Third", "Fourth")
- val futureResult: Future[immutable.Seq[JsonDocument]] =
+ val futureResult: Future[Seq[JsonDocument]] =
Source(ids)
.via(
CouchbaseFlow.fromId(
@@ -220,7 +218,7 @@ class CouchbaseFlowSpec
.runWith(Sink.ignore)
bulkUpsertResult.futureValue
- val resultsAsFuture: Future[immutable.Seq[StringDocument]] =
+ val resultsAsFuture: Future[Seq[StringDocument]] =
Source(sampleSequence.map(_.id))
.via(
CouchbaseFlow.fromId(
@@ -243,7 +241,7 @@ class CouchbaseFlowSpec
.runWith(Sink.ignore)
bulkUpsertResult.futureValue
- val resultsAsFuture: Future[immutable.Seq[BinaryDocument]] =
+ val resultsAsFuture: Future[Seq[BinaryDocument]] =
Source(sampleSequence.map(_.id))
.via(
CouchbaseFlow.fromId(
@@ -255,7 +253,7 @@ class CouchbaseFlowSpec
}
"fails stream when ReplicateTo higher then #of nodes" in assertAllStagesStopped {
- val bulkUpsertResult: Future[immutable.Seq[JsonDocument]] = Source(sampleSequence)
+ val bulkUpsertResult: Future[Seq[JsonDocument]] = Source(sampleSequence)
.map(toJsonDocument)
.via(
CouchbaseFlow.upsert(sessionSettings,
@@ -430,7 +428,7 @@ class CouchbaseFlowSpec
bulkReplaceResult.futureValue
- val resultsAsFuture: Future[immutable.Seq[JsonDocument]] =
+ val resultsAsFuture: Future[Seq[JsonDocument]] =
Source(sampleSequence.map(_.id))
.via(CouchbaseFlow.fromId(sessionSettings, bucketName))
.runWith(Sink.seq)
@@ -471,7 +469,7 @@ class CouchbaseFlowSpec
upsertSampleData(bucketName)
- val bulkReplaceResult: Future[immutable.Seq[JsonDocument]] = Source(sampleSequence)
+ val bulkReplaceResult: Future[Seq[JsonDocument]] = Source(sampleSequence)
.map(toJsonDocument)
.via(
CouchbaseFlow.replace(sessionSettings,
@@ -490,7 +488,7 @@ class CouchbaseFlowSpec
"write documents" in assertAllStagesStopped {
// #upsertDocWithResult
- val result: Future[immutable.Seq[CouchbaseWriteResult[RawJsonDocument]]] =
+ val result: Future[Seq[CouchbaseWriteResult[RawJsonDocument]]] =
Source(sampleSequence)
.map(toRawJsonDocument)
.via(
@@ -500,7 +498,7 @@ class CouchbaseFlowSpec
bucketName))
.runWith(Sink.seq)
- val failedDocs: immutable.Seq[CouchbaseWriteFailure[RawJsonDocument]] = result.futureValue.collect {
+ val failedDocs: Seq[CouchbaseWriteFailure[RawJsonDocument]] = result.futureValue.collect {
case res: CouchbaseWriteFailure[RawJsonDocument] => res
}
// #upsertDocWithResult
@@ -512,7 +510,7 @@ class CouchbaseFlowSpec
"expose failures in-stream" in assertAllStagesStopped {
- val result: Future[immutable.Seq[CouchbaseWriteResult[JsonDocument]]] = Source(sampleSequence)
+ val result: Future[Seq[CouchbaseWriteResult[JsonDocument]]] = Source(sampleSequence)
.map(toJsonDocument)
.via(
CouchbaseFlow.upsertDocWithResult(sessionSettings,
@@ -524,7 +522,7 @@ class CouchbaseFlowSpec
.runWith(Sink.seq)
result.futureValue should have size sampleSequence.size
- val failedDocs: immutable.Seq[CouchbaseWriteFailure[JsonDocument]] = result.futureValue.collect {
+ val failedDocs: Seq[CouchbaseWriteFailure[JsonDocument]] = result.futureValue.collect {
case res: CouchbaseWriteFailure[JsonDocument] => res
}
failedDocs.head.failure shouldBe a[com.couchbase.client.java.error.DurabilityException]
@@ -573,7 +571,7 @@ class CouchbaseFlowSpec
// #replaceDocWithResult
- val result: Future[immutable.Seq[CouchbaseWriteResult[RawJsonDocument]]] =
+ val result: Future[Seq[CouchbaseWriteResult[RawJsonDocument]]] =
Source(sampleSequence)
.map(toRawJsonDocument)
.via(
@@ -583,7 +581,7 @@ class CouchbaseFlowSpec
bucketName))
.runWith(Sink.seq)
- val failedDocs: immutable.Seq[CouchbaseWriteFailure[RawJsonDocument]] = result.futureValue.collect {
+ val failedDocs: Seq[CouchbaseWriteFailure[RawJsonDocument]] = result.futureValue.collect {
case res: CouchbaseWriteFailure[RawJsonDocument] => res
}
// #replaceDocWithResult
@@ -597,7 +595,7 @@ class CouchbaseFlowSpec
cleanAllInBucket(bucketName)
- val result: Future[immutable.Seq[CouchbaseWriteResult[JsonDocument]]] = Source(sampleSequence)
+ val result: Future[Seq[CouchbaseWriteResult[JsonDocument]]] = Source(sampleSequence)
.map(toJsonDocument)
.via(
CouchbaseFlow.replaceDocWithResult(sessionSettings,
@@ -609,7 +607,7 @@ class CouchbaseFlowSpec
.runWith(Sink.seq)
result.futureValue should have size sampleSequence.size
- val failedDocs: immutable.Seq[CouchbaseWriteFailure[JsonDocument]] = result.futureValue.collect {
+ val failedDocs: Seq[CouchbaseWriteFailure[JsonDocument]] = result.futureValue.collect {
case res: CouchbaseWriteFailure[JsonDocument] => res
}
failedDocs.head.failure shouldBe a[com.couchbase.client.java.error.DocumentDoesNotExistException]
diff --git a/couchbase/src/test/scala/docs/scaladsl/CouchbaseSourceSpec.scala b/couchbase/src/test/scala/docs/scaladsl/CouchbaseSourceSpec.scala
index f8494a858..6b8b5dc77 100644
--- a/couchbase/src/test/scala/docs/scaladsl/CouchbaseSourceSpec.scala
+++ b/couchbase/src/test/scala/docs/scaladsl/CouchbaseSourceSpec.scala
@@ -25,7 +25,6 @@ import com.couchbase.client.java.document.json.JsonObject
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.BeforeAndAfterAll
-import scala.collection.immutable.Seq
import scala.concurrent.duration._
import scala.concurrent.Future
import org.scalatest.matchers.should.Matchers
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 2aa72d95e..650156d2d 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
@@ -28,7 +28,6 @@ import com.fasterxml.jackson.databind.json.JsonMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.slf4j.LoggerFactory
-import scala.collection.immutable.Seq
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.concurrent.{ Await, Future }
diff --git a/couchbase3/src/test/scala/docs/scaladsl/CouchbaseTestFlowSpec.scala b/couchbase3/src/test/scala/docs/scaladsl/CouchbaseTestFlowSpec.scala
index a50d90a59..bc4b999f2 100644
--- a/couchbase3/src/test/scala/docs/scaladsl/CouchbaseTestFlowSpec.scala
+++ b/couchbase3/src/test/scala/docs/scaladsl/CouchbaseTestFlowSpec.scala
@@ -31,7 +31,6 @@ import org.scalatest.wordspec.AnyWordSpec
import java.time.Duration
import java.util.Collections
-import scala.collection.immutable.Seq
import scala.concurrent.Await
import scala.concurrent.duration.DurationInt
import scala.util.Random
diff --git a/csv/src/main/scala/org/apache/pekko/stream/connectors/csv/impl/CsvToMapStage.scala b/csv/src/main/scala/org/apache/pekko/stream/connectors/csv/impl/CsvToMapStage.scala
index 685fc487d..3b0cc94e4 100644
--- a/csv/src/main/scala/org/apache/pekko/stream/connectors/csv/impl/CsvToMapStage.scala
+++ b/csv/src/main/scala/org/apache/pekko/stream/connectors/csv/impl/CsvToMapStage.scala
@@ -20,8 +20,6 @@ import pekko.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler }
import pekko.stream.{ Attributes, FlowShape, Inlet, Outlet }
import pekko.util.ByteString
-import scala.collection.immutable
-
/**
* Internal API: Converts incoming [[List[ByteString]]] to [[Map[String, ByteString]]].
*
@@ -32,24 +30,24 @@ import scala.collection.immutable
* @param customFieldValuePlaceholder placeholder used when there are more data than headers.
* @param headerPlaceholder placeholder used when there are more headers than data.
*/
-@InternalApi private[csv] abstract class CsvToMapStageBase[V](columnNames: Option[immutable.Seq[String]],
+@InternalApi private[csv] abstract class CsvToMapStageBase[V](columnNames: Option[Seq[String]],
charset: Charset,
combineAll: Boolean,
customFieldValuePlaceholder: Option[V],
headerPlaceholder: Option[String])
- extends GraphStage[FlowShape[immutable.Seq[ByteString], Map[String, V]]] {
+ extends GraphStage[FlowShape[Seq[ByteString], Map[String, V]]] {
override protected def initialAttributes: Attributes = Attributes.name("CsvToMap")
private type Headers = Option[Seq[String]]
- private val in = Inlet[immutable.Seq[ByteString]]("CsvToMap.in")
+ private val in = Inlet[Seq[ByteString]]("CsvToMap.in")
private val out = Outlet[Map[String, V]]("CsvToMap.out")
override val shape = FlowShape.of(in, out)
val fieldValuePlaceholder: V
- protected def transformElements(elements: immutable.Seq[ByteString]): immutable.Seq[V]
+ protected def transformElements(elements: Seq[ByteString]): Seq[V]
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
new GraphStageLogic(shape) with InHandler with OutHandler {
@@ -66,7 +64,7 @@ import scala.collection.immutable
}
}
- private def process(elem: immutable.Seq[ByteString], combine: => Headers => Map[String, V]): Unit = {
+ private def process(elem: Seq[ByteString], combine: => Headers => Map[String, V]): Unit = {
if (headers.isDefined) {
push(out, combine(headers))
} else {
@@ -78,7 +76,7 @@ import scala.collection.immutable
override def onPull(): Unit = pull(in)
}
- private def combineUsingPlaceholder(elem: immutable.Seq[ByteString]): Headers => Map[String, V] = headers => {
+ private def combineUsingPlaceholder(elem: Seq[ByteString]): Headers => Map[String, V] = headers => {
val combined = headers.get
.zipAll(transformElements(elem),
headerPlaceholder.getOrElse("MissingHeader"),
@@ -110,7 +108,7 @@ import scala.collection.immutable
/**
* Internal API
*/
-@InternalApi private[csv] class CsvToMapStage(columnNames: Option[immutable.Seq[String]],
+@InternalApi private[csv] class CsvToMapStage(columnNames: Option[Seq[String]],
charset: Charset,
combineAll: Boolean,
customFieldValuePlaceholder: Option[ByteString],
@@ -123,14 +121,14 @@ import scala.collection.immutable
override val fieldValuePlaceholder: ByteString = ByteString.empty
- override protected def transformElements(elements: immutable.Seq[ByteString]): immutable.Seq[ByteString] = elements
+ override protected def transformElements(elements: Seq[ByteString]): Seq[ByteString] = elements
}
/**
* Internal API
*/
-@InternalApi private[csv] class CsvToMapAsStringsStage(columnNames: Option[immutable.Seq[String]],
+@InternalApi private[csv] class CsvToMapAsStringsStage(columnNames: Option[Seq[String]],
charset: Charset,
combineAll: Boolean,
customFieldValuePlaceholder: Option[String],
@@ -140,6 +138,6 @@ import scala.collection.immutable
override val fieldValuePlaceholder: String = ""
- override protected def transformElements(elements: immutable.Seq[ByteString]): immutable.Seq[String] =
+ override protected def transformElements(elements: Seq[ByteString]): Seq[String] =
elements.map(_.decodeString(charset))
}
diff --git a/csv/src/test/scala/docs/scaladsl/CsvFormattingSpec.scala b/csv/src/test/scala/docs/scaladsl/CsvFormattingSpec.scala
index 62dcb85d8..8d6b09405 100644
--- a/csv/src/test/scala/docs/scaladsl/CsvFormattingSpec.scala
+++ b/csv/src/test/scala/docs/scaladsl/CsvFormattingSpec.scala
@@ -20,8 +20,6 @@ import pekko.stream.scaladsl.{ Flow, Sink, Source }
import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import pekko.util.ByteString
-import scala.collection.immutable
-
class CsvFormattingSpec extends CsvSpec {
def documentation(): Unit = {
@@ -36,7 +34,7 @@ class CsvFormattingSpec extends CsvSpec {
val endOfLine = "\r\n"
// format: off
// #flow-type
- val flow: Flow[immutable.Seq[String], ByteString, _]
+ val flow: Flow[Seq[String], ByteString, _]
= CsvFormatting.format(delimiter,
quoteChar,
escapeChar,
diff --git a/csv/src/test/scala/docs/scaladsl/CsvParsingSpec.scala b/csv/src/test/scala/docs/scaladsl/CsvParsingSpec.scala
index 8894b0071..af84e2534 100644
--- a/csv/src/test/scala/docs/scaladsl/CsvParsingSpec.scala
+++ b/csv/src/test/scala/docs/scaladsl/CsvParsingSpec.scala
@@ -23,7 +23,6 @@ import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import pekko.stream.testkit.scaladsl.{ TestSink, TestSource }
import pekko.util.ByteString
-import scala.collection.immutable.Seq
import scala.concurrent.duration.DurationInt
class CsvParsingSpec extends CsvSpec {
diff --git a/docs/src/main/paradox/data-transformations/csv.md b/docs/src/main/paradox/data-transformations/csv.md
index f7ee1ce7a..665ac65a6 100644
--- a/docs/src/main/paradox/data-transformations/csv.md
+++ b/docs/src/main/paradox/data-transformations/csv.md
@@ -111,7 +111,7 @@ Java
## CSV formatting
-To emit CSV files ``immutable.Seq[String]`` can be formatted into ``ByteString`` e.g to be written to file.
+To emit CSV files ``Seq[String]`` can be formatted into ``ByteString`` e.g to be written to file.
The formatter takes care of quoting and escaping.
Certain CSV readers (e.g. Microsoft Excel) require CSV files to indicate their character encoding with a *Byte
diff --git a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSimpleFlowStage.scala b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSimpleFlowStage.scala
index 4a7e502b5..2aa2c2bb9 100644
--- a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSimpleFlowStage.scala
+++ b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSimpleFlowStage.scala
@@ -24,7 +24,6 @@ import pekko.stream.stage._
import pekko.stream._
import pekko.stream.connectors.elasticsearch
-import scala.collection.immutable
import scala.concurrent.{ ExecutionContext, Future }
/**
@@ -38,13 +37,13 @@ private[elasticsearch] final class ElasticsearchSimpleFlowStage[T, C](
settings: WriteSettingsBase[?, ?],
writer: MessageWriter[T])(implicit http: HttpExt, mat: Materializer, ec: ExecutionContext)
extends GraphStage[
- FlowShape[(immutable.Seq[WriteMessage[T, C]], immutable.Seq[WriteResult[T, C]]),
- immutable.Seq[WriteResult[T,
+ FlowShape[(Seq[WriteMessage[T, C]], Seq[WriteResult[T, C]]),
+ Seq[WriteResult[T,
C]]]] {
private val in =
- Inlet[(immutable.Seq[WriteMessage[T, C]], immutable.Seq[WriteResult[T, C]])]("messagesAndResultPassthrough")
- private val out = Outlet[immutable.Seq[WriteResult[T, C]]]("result")
+ Inlet[(Seq[WriteMessage[T, C]], Seq[WriteResult[T, C]])]("messagesAndResultPassthrough")
+ private val out = Outlet[Seq[WriteResult[T, C]]]("result")
override val shape = FlowShape(in, out)
private val restApi: RestBulkApi[T, C] = settings.apiVersion match {
@@ -72,9 +71,9 @@ private[elasticsearch] final class ElasticsearchSimpleFlowStage[T, C](
private var inflight = false
private val failureHandler =
- getAsyncCallback[(immutable.Seq[WriteResult[T, C]], Throwable)](handleFailure)
+ getAsyncCallback[(Seq[WriteResult[T, C]], Throwable)](handleFailure)
private val responseHandler =
- getAsyncCallback[(immutable.Seq[WriteMessage[T, C]], immutable.Seq[WriteResult[T, C]], String)](handleResponse)
+ getAsyncCallback[(Seq[WriteMessage[T, C]], Seq[WriteResult[T, C]], String)](handleResponse)
setHandlers(in, out, this)
@@ -123,7 +122,7 @@ private[elasticsearch] final class ElasticsearchSimpleFlowStage[T, C](
}
private def handleFailure(
- args: (immutable.Seq[WriteResult[T, C]], Throwable)): Unit = {
+ args: (Seq[WriteResult[T, C]], Throwable)): Unit = {
inflight = false
val (resultsPassthrough, exception) = args
@@ -134,7 +133,7 @@ private[elasticsearch] final class ElasticsearchSimpleFlowStage[T, C](
}
private def handleResponse(
- args: (immutable.Seq[WriteMessage[T, C]], immutable.Seq[WriteResult[T, C]], String)): Unit = {
+ args: (Seq[WriteMessage[T, C]], Seq[WriteResult[T, C]], String)): Unit = {
inflight = false
val (messages, resultsPassthrough, response) = args
diff --git a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApi.scala b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApi.scala
index e9db5c111..18dd8b50b 100644
--- a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApi.scala
+++ b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/RestBulkApi.scala
@@ -27,10 +27,10 @@ import scala.collection.immutable
@InternalApi
private[impl] abstract class RestBulkApi[T, C] {
- def toJson(messages: immutable.Seq[WriteMessage[T, C]]): String
+ def toJson(messages: Seq[WriteMessage[T, C]]): String
- def toWriteResults(messages: immutable.Seq[WriteMessage[T, C]],
- jsonString: String): immutable.Seq[WriteResult[T, C]] = {
+ def toWriteResults(messages: Seq[WriteMessage[T, C]],
+ jsonString: String): Seq[WriteResult[T, C]] = {
val responseJson = jsonString.parseJson
// If some commands in bulk request failed, pass failed messages to follows.
@@ -56,7 +56,7 @@ private[impl] abstract class RestBulkApi[T, C] {
/** NOPs don't come back so slip them into the results like this: */
private def buildMessageResults(items: JsArray,
- messages: immutable.Seq[WriteMessage[T, C]]): immutable.Seq[WriteResult[T, C]] = {
+ messages: Seq[WriteMessage[T, C]]): Seq[WriteResult[T, C]] = {
val ret = new immutable.VectorBuilder[WriteResult[T, C]]
ret.sizeHint(messages)
val itemsIter = items.elements.iterator
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 f6514bc7a..ce8db9f6f 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
@@ -19,8 +19,6 @@ import pekko.stream.connectors.elasticsearch.Operation._
import pekko.stream.connectors.elasticsearch.{ MessageWriter, WriteMessage }
import spray.json._
-import scala.collection.immutable
-
/**
* Internal API.
*
@@ -37,7 +35,7 @@ private[impl] final class RestBulkApiV5[T, C](indexName: String,
private lazy val typeNameTuple = "_type" -> JsString(typeName)
- def toJson(messages: immutable.Seq[WriteMessage[T, C]]): String =
+ def toJson(messages: Seq[WriteMessage[T, C]]): String =
messages
.map { message =>
val sharedFields = constructSharedFields(message)
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 214b8d7dd..b990fffc6 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
@@ -19,8 +19,6 @@ import pekko.stream.connectors.elasticsearch.Operation._
import pekko.stream.connectors.elasticsearch.{ MessageWriter, WriteMessage }
import spray.json._
-import scala.collection.immutable
-
/**
* Internal API.
*
@@ -34,7 +32,7 @@ private[impl] final class RestBulkApiV7[T, C](indexName: String,
messageWriter: MessageWriter[T])
extends RestBulkApi[T, C] {
- def toJson(messages: immutable.Seq[WriteMessage[T, C]]): String =
+ def toJson(messages: Seq[WriteMessage[T, C]]): String =
messages
.map { message =>
val sharedFields = constructSharedFields(message)
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 1e10386fc..6a7b5faf7 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
@@ -22,7 +22,6 @@ import pekko.stream.connectors.elasticsearch.{ impl, _ }
import pekko.stream.scaladsl.{ Flow, FlowWithContext, RetryFlow }
import spray.json._
-import scala.collection.immutable
import scala.concurrent.ExecutionContextExecutor
/**
@@ -50,7 +49,7 @@ object ElasticsearchFlow {
settings: WriteSettingsBase[?, ?],
writer: MessageWriter[T]): Flow[WriteMessage[T, NotUsed], WriteResult[T, NotUsed], NotUsed] = {
Flow[WriteMessage[T, NotUsed]]
- .batch(settings.bufferSize, immutable.Seq(_)) { case (seq, wm) => seq :+ wm }
+ .batch(settings.bufferSize, Seq(_)) { case (seq, wm) => seq :+ wm }
.via(stageFlow(elasticsearchParams, settings, writer))
.mapConcat(identity)
}
@@ -77,36 +76,36 @@ object ElasticsearchFlow {
settings: WriteSettingsBase[?, ?],
writer: MessageWriter[T]): Flow[WriteMessage[T, C], WriteResult[T, C], NotUsed] = {
Flow[WriteMessage[T, C]]
- .batch(settings.bufferSize, immutable.Seq(_)) { case (seq, wm) => seq :+ wm }
+ .batch(settings.bufferSize, Seq(_)) { case (seq, wm) => seq :+ wm }
.via(stageFlow(elasticsearchParams, settings, writer))
.mapConcat(identity)
}
/**
* Create a flow to update Elasticsearch with
- * `immutable.Seq[WriteMessage]`s containing type `T`
+ * `Seq[WriteMessage]`s containing type `T`
* with `passThrough` of type `C`.
- * The result status is part of the `immutable.Seq[WriteResult]`
+ * The result status is part of the `Seq[WriteResult]`
* and must be checked for successful execution.
*
* This factory method requires an implicit Spray JSON writer for `T`.
*/
def createBulk[T, C](elasticsearchParams: ElasticsearchParams, settings: WriteSettingsBase[?, ?])(
implicit sprayJsonWriter: JsonWriter[T])
- : Flow[immutable.Seq[WriteMessage[T, C]], immutable.Seq[WriteResult[T, C]], NotUsed] =
+ : Flow[Seq[WriteMessage[T, C]], Seq[WriteResult[T, C]], NotUsed] =
createBulk[T, C](elasticsearchParams, settings, new SprayJsonWriter[T]()(sprayJsonWriter))
/**
* Create a flow to update Elasticsearch with
- * `immutable.Seq[WriteMessage]`s containing type `T`
+ * `Seq[WriteMessage]`s containing type `T`
* with `passThrough` of type `C`.
- * The result status is part of the `immutable.Seq[WriteResult]`
+ * The result status is part of the `Seq[WriteResult]`
* and must be checked for successful execution.
*/
def createBulk[T, C](
elasticsearchParams: ElasticsearchParams,
settings: WriteSettingsBase[?, ?],
- writer: MessageWriter[T]): Flow[immutable.Seq[WriteMessage[T, C]], immutable.Seq[WriteResult[T, C]], NotUsed] = {
+ writer: MessageWriter[T]): Flow[Seq[WriteMessage[T, C]], Seq[WriteResult[T, C]], NotUsed] = {
stageFlow(elasticsearchParams, settings, writer)
}
@@ -136,7 +135,7 @@ object ElasticsearchFlow {
settings: WriteSettingsBase[?, ?],
writer: MessageWriter[T]): FlowWithContext[WriteMessage[T, NotUsed], C, WriteResult[T, C], C, NotUsed] = {
Flow[WriteMessage[T, C]]
- .batch(settings.bufferSize, immutable.Seq(_)) { case (seq, wm) => seq :+ wm }
+ .batch(settings.bufferSize, Seq(_)) { case (seq, wm) => seq :+ wm }
.via(stageFlow(elasticsearchParams, settings, writer))
.mapConcat(identity)
.asFlowWithContext[WriteMessage[T, NotUsed], C, C]((res, c) => res.withPassThrough(c))(p => p.message.passThrough)
@@ -146,16 +145,16 @@ object ElasticsearchFlow {
private def stageFlow[T, C](
elasticsearchParams: ElasticsearchParams,
settings: WriteSettingsBase[?, ?],
- writer: MessageWriter[T]): Flow[immutable.Seq[WriteMessage[T, C]], immutable.Seq[WriteResult[T, C]], NotUsed] = {
+ writer: MessageWriter[T]): Flow[Seq[WriteMessage[T, C]], Seq[WriteResult[T, C]], NotUsed] = {
if (settings.retryLogic == RetryNever) {
val basicFlow = basicStageFlow[T, C](elasticsearchParams, settings, writer)
- Flow[immutable.Seq[WriteMessage[T, C]]]
- .map(messages => messages -> immutable.Seq.empty[WriteResult[T, C]])
+ Flow[Seq[WriteMessage[T, C]]]
+ .map(messages => messages -> Seq.empty[WriteResult[T, C]])
.via(basicFlow)
} else {
def retryLogic(
- results: immutable.Seq[WriteResult[T, (Int, C)]])
- : Option[(immutable.Seq[WriteMessage[T, (Int, C)]], immutable.Seq[WriteResult[T, (Int, C)]])] = {
+ results: Seq[WriteResult[T, (Int, C)]])
+ : Option[(Seq[WriteMessage[T, (Int, C)]], Seq[WriteResult[T, (Int, C)]])] = {
val (successful, failed) = results.partition(_.success)
failed match {
@@ -180,12 +179,12 @@ object ElasticsearchFlow {
}
@InternalApi
- private def amendWithIndexFlow[T, C]: Flow[immutable.Seq[WriteMessage[T, C]],
+ private def amendWithIndexFlow[T, C]: Flow[Seq[WriteMessage[T, C]],
(
- immutable.Seq[WriteMessage[T,
+ Seq[WriteMessage[T,
(Int,
- C)]], immutable.Seq[WriteResult[T, (Int, C)]]), NotUsed] = {
- Flow[immutable.Seq[WriteMessage[T, C]]].map { messages =>
+ C)]], Seq[WriteResult[T, (Int, C)]]), NotUsed] = {
+ Flow[Seq[WriteMessage[T, C]]].map { messages =>
val indexedMessages = messages.zipWithIndex.map {
case (m, idx) =>
m.withPassThrough(idx -> m.passThrough)
@@ -196,8 +195,8 @@ object ElasticsearchFlow {
@InternalApi
private def applyOrderingFlow[T, C]
- : Flow[immutable.Seq[WriteResult[T, (Int, C)]], immutable.Seq[WriteResult[T, C]], NotUsed] = {
- Flow[immutable.Seq[WriteResult[T, (Int, C)]]].map { results =>
+ : Flow[Seq[WriteResult[T, (Int, C)]], Seq[WriteResult[T, C]], NotUsed] = {
+ Flow[Seq[WriteResult[T, (Int, C)]]].map { results =>
val orderedResults = results.sortBy(_.message.passThrough._1)
val finalResults = orderedResults.map { r =>
new WriteResult(r.message.withPassThrough(r.message.passThrough._2), r.error)
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala
index 56ce918c2..e16e22136 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala
@@ -29,7 +29,6 @@ import org.scalatest.concurrent.ScalaFutures
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec
-import scala.collection.immutable
import scala.concurrent.duration._
trait ElasticsearchConnectorBehaviour {
@@ -133,7 +132,7 @@ trait ElasticsearchConnectorBehaviour {
"ElasticsearchFlow" should {
"pass through data in `withContext`" in {
- val books = immutable.Seq(
+ val books = Seq(
"Akka in Action",
"Pekko Connectors Patterns")
@@ -157,7 +156,7 @@ trait ElasticsearchConnectorBehaviour {
}
"not post invalid encoded JSON" in {
- val books = immutable.Seq(
+ val books = Seq(
"Akka in Action",
"Akka \u00DF Concurrency")
@@ -188,10 +187,9 @@ trait ElasticsearchConnectorBehaviour {
createStrictMapping(indexName)
val createBooks = Source(
- immutable
- .Seq(
- Book("Akka in Action").toJson,
- JsObject("subject" -> "Akka Concurrency".toJson))
+ Seq(
+ Book("Akka in Action").toJson,
+ JsObject("subject" -> "Akka Concurrency".toJson))
.zipWithIndex).map {
case (book: JsObject, index: Int) =>
WriteMessage.createIndexMessage(index.toString, book)
@@ -273,11 +271,10 @@ trait ElasticsearchConnectorBehaviour {
createStrictMapping(indexName)
val createBooks = Source(
- immutable
- .Seq(
- Book("Akka in Action").toJson,
- JsObject("subject" -> "Akka Concurrency".toJson),
- Book("Learning Scala").toJson)
+ Seq(
+ Book("Akka in Action").toJson,
+ JsObject("subject" -> "Akka Concurrency".toJson),
+ Book("Learning Scala").toJson)
.zipWithIndex).map {
case (book: JsObject, index: Int) =>
WriteMessage.createIndexMessage(index.toString, book) -> index
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchSpecUtils.scala b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchSpecUtils.scala
index 050b2ca17..68379da86 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchSpecUtils.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchSpecUtils.scala
@@ -31,7 +31,6 @@ import pekko.stream.scaladsl.Sink
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.wordspec.AnyWordSpec
-import scala.collection.immutable
import scala.concurrent.Future
trait ElasticsearchSpecUtils { this: AnyWordSpec with ScalaFutures =>
@@ -70,7 +69,7 @@ trait ElasticsearchSpecUtils { this: AnyWordSpec with ScalaFutures =>
def readTitlesFrom(apiVersion: ApiVersionBase,
sourceSettings: SourceSettingsBase[?, ?],
- indexName: String): Future[immutable.Seq[String]] =
+ indexName: String): Future[Seq[String]] =
ElasticsearchSource
.typed[Book](
constructElasticsearchParams(indexName, "_doc", apiVersion),
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala
index 8e92a0b9f..65a79fb44 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV5Spec.scala
@@ -32,7 +32,6 @@ import pekko.testkit.TestKit
import pekko.{ Done, NotUsed }
import spray.json.jsonReader
-import scala.collection.immutable
import scala.concurrent.Future
import spray.json._
@@ -156,8 +155,8 @@ class ElasticsearchV5Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
"store properly formatted JSON from Strings" in {
val indexName = "sink3-0"
// #string
- val write: Future[immutable.Seq[WriteResult[String, NotUsed]]] = Source(
- immutable.Seq(
+ val write: Future[Seq[WriteResult[String, NotUsed]]] = Source(
+ Seq(
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(
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala
index 461b8535e..cbb1a5ff8 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchV7Spec.scala
@@ -23,7 +23,6 @@ import pekko.testkit.TestKit
import pekko.{ Done, NotUsed }
import spray.json.jsonReader
-import scala.collection.immutable
import scala.concurrent.Future
import spray.json._
@@ -145,8 +144,8 @@ class ElasticsearchV7Spec extends ElasticsearchSpecBase with ElasticsearchSpecUt
"store properly formatted JSON from Strings" in {
val indexName = "sink3-0"
- val write: Future[immutable.Seq[WriteResult[String, NotUsed]]] = Source(
- immutable.Seq(
+ val write: Future[Seq[WriteResult[String, NotUsed]]] = Source(
+ Seq(
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(
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchConnectorBehaviour.scala b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchConnectorBehaviour.scala
index 4e2755391..3abe5f1ea 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchConnectorBehaviour.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchConnectorBehaviour.scala
@@ -29,7 +29,6 @@ import org.scalatest.concurrent.ScalaFutures
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec
-import scala.collection.immutable
import scala.concurrent.duration._
trait OpensearchConnectorBehaviour {
@@ -133,7 +132,7 @@ trait OpensearchConnectorBehaviour {
"ElasticsearchFlow" should {
"pass through data in `withContext`" in {
- val books = immutable.Seq(
+ val books = Seq(
"Akka in Action",
"Pekko Connectors Patterns")
@@ -157,7 +156,7 @@ trait OpensearchConnectorBehaviour {
}
"not post invalid encoded JSON" in {
- val books = immutable.Seq(
+ val books = Seq(
"Akka in Action",
"Akka \u00DF Concurrency")
@@ -188,10 +187,9 @@ trait OpensearchConnectorBehaviour {
createStrictMapping(indexName)
val createBooks = Source(
- immutable
- .Seq(
- Book("Akka in Action").toJson,
- JsObject("subject" -> "Akka Concurrency".toJson))
+ Seq(
+ Book("Akka in Action").toJson,
+ JsObject("subject" -> "Akka Concurrency".toJson))
.zipWithIndex).map {
case (book: JsObject, index: Int) =>
WriteMessage.createIndexMessage(index.toString, book)
@@ -273,11 +271,10 @@ trait OpensearchConnectorBehaviour {
createStrictMapping(indexName)
val createBooks = Source(
- immutable
- .Seq(
- Book("Akka in Action").toJson,
- JsObject("subject" -> "Akka Concurrency".toJson),
- Book("Learning Scala").toJson)
+ Seq(
+ Book("Akka in Action").toJson,
+ JsObject("subject" -> "Akka Concurrency".toJson),
+ Book("Learning Scala").toJson)
.zipWithIndex).map {
case (book: JsObject, index: Int) =>
WriteMessage.createIndexMessage(index.toString, book) -> index
diff --git a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
index ba7c5d443..5bba48a91 100644
--- a/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
+++ b/elasticsearch/src/test/scala/docs/scaladsl/OpensearchV1Spec.scala
@@ -32,7 +32,6 @@ import pekko.testkit.TestKit
import pekko.{ Done, NotUsed }
import spray.json.jsonReader
-import scala.collection.immutable
import scala.concurrent.Future
import spray.json._
@@ -161,8 +160,8 @@ class OpensearchV1Spec extends ElasticsearchSpecBase with ElasticsearchSpecUtils
val indexName = "sink3-0"
// #string
- val write: Future[immutable.Seq[WriteResult[String, NotUsed]]] = Source(
- immutable.Seq(
+ val write: Future[Seq[WriteResult[String, NotUsed]]] = Source(
+ Seq(
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(
diff --git a/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSimpleFlowStageTest.scala b/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSimpleFlowStageTest.scala
index 49dbdacc7..4818e39cd 100644
--- a/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSimpleFlowStageTest.scala
+++ b/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchSimpleFlowStageTest.scala
@@ -26,7 +26,6 @@ import pekko.testkit.TestKit
import org.scalatest.BeforeAndAfterAll
import org.scalatest.wordspec.AnyWordSpecLike
-import scala.collection.immutable
import scala.concurrent.ExecutionContext.Implicits.global
class ElasticsearchSimpleFlowStageTest
@@ -41,18 +40,18 @@ class ElasticsearchSimpleFlowStageTest
val writer: StringMessageWriter = StringMessageWriter.getInstance
val settings: ElasticsearchWriteSettings = ElasticsearchWriteSettings(
ElasticsearchConnectionSettings("http://localhost:9202"))
- val dummyMessages: (immutable.Seq[WriteMessage[String, NotUsed]], immutable.Seq[WriteResult[String, NotUsed]]) = (
- immutable.Seq(
+ val dummyMessages: (Seq[WriteMessage[String, NotUsed]], Seq[WriteResult[String, NotUsed]]) = (
+ Seq(
WriteMessage.createIndexMessage("1", """{"foo": "bar"}"""),
WriteMessage.createIndexMessage("2", """{"foo2": "bar2"}"""),
WriteMessage.createIndexMessage("3", """{"foo3": "bar3"}""")),
- immutable.Seq[WriteResult[String, NotUsed]]())
+ Seq[WriteResult[String, NotUsed]]())
"ElasticsearchSimpleFlowStage" when {
"stream ends" should {
"emit element only when downstream requests" in {
val (upstream, downstream) =
- TestSource[(immutable.Seq[WriteMessage[String, NotUsed]], immutable.Seq[WriteResult[String, NotUsed]])]()
+ TestSource[(Seq[WriteMessage[String, NotUsed]], Seq[WriteResult[String, NotUsed]])]()
.via(
new impl.ElasticsearchSimpleFlowStage[String, NotUsed](
ElasticsearchParams.V7("es-simple-flow-index"),
@@ -76,7 +75,7 @@ class ElasticsearchSimpleFlowStageTest
"client cannot connect to ES" should {
"stop the stream" in {
val (upstream, downstream) =
- TestSource[(immutable.Seq[WriteMessage[String, NotUsed]], immutable.Seq[WriteResult[String, NotUsed]])]()
+ TestSource[(Seq[WriteMessage[String, NotUsed]], Seq[WriteResult[String, NotUsed]])]()
.via(
new impl.ElasticsearchSimpleFlowStage[String, NotUsed](
ElasticsearchParams.V7("es-simple-flow-index"),
diff --git a/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/Directory.scala b/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/Directory.scala
index d6ea0502f..f9d31f9d1 100644
--- a/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/Directory.scala
+++ b/file/src/main/scala/org/apache/pekko/stream/connectors/file/scaladsl/Directory.scala
@@ -20,8 +20,6 @@ import pekko.NotUsed
import pekko.stream.ActorAttributes
import pekko.stream.scaladsl.{ Flow, FlowWithContext, Source, StreamConverters }
-import scala.collection.immutable
-
object Directory {
/**
@@ -41,7 +39,7 @@ object Directory {
*/
def walk(directory: Path,
maxDepth: Option[Int] = None,
- fileVisitOptions: immutable.Seq[FileVisitOption] = Nil): Source[Path, NotUsed] = {
+ fileVisitOptions: Seq[FileVisitOption] = Nil): Source[Path, NotUsed] = {
require(Files.isDirectory(directory), s"Path must be a directory, $directory isn't")
val factory = maxDepth match {
case None =>
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 49c972517..5266470dc 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
@@ -23,7 +23,6 @@ import pekko.stream.scaladsl.{ FileIO, Sink, Source }
import pekko.stream.stage._
import pekko.util.ByteString
-import scala.collection.immutable
import scala.concurrent.{ ExecutionContext, Future, Promise }
import scala.util.{ Failure, Success }
@@ -98,7 +97,7 @@ final private class LogRotatorSink[T, C, R](triggerGeneratorCreator: () => T =>
private final class Logic(promise: Promise[Done]) extends GraphStageLogic(shape) {
val triggerGenerator: T => Option[C] = triggerGeneratorCreator()
var sourceOut: SubSourceOutlet[T] = null
- var sinkCompletions: immutable.Seq[Future[R]] = immutable.Seq.empty
+ var sinkCompletions: Seq[Future[R]] = Seq.empty
var isFinishing = false
def failThisStage(ex: Throwable): Unit =
diff --git a/file/src/test/scala/docs/scaladsl/DirectorySpec.scala b/file/src/test/scala/docs/scaladsl/DirectorySpec.scala
index dd3a7f986..368654c25 100644
--- a/file/src/test/scala/docs/scaladsl/DirectorySpec.scala
+++ b/file/src/test/scala/docs/scaladsl/DirectorySpec.scala
@@ -26,7 +26,6 @@ import com.google.common.jimfs.{ Configuration, Jimfs }
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.BeforeAndAfterAll
-import scala.collection.immutable
import scala.concurrent.Future
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
@@ -97,8 +96,8 @@ class DirectorySpec
val flow: Flow[Path, Path, NotUsed] = Directory.mkdirs()
- val created: Future[immutable.Seq[Path]] =
- Source(immutable.Seq(dir.resolve("dirA"), dir.resolve("dirB")))
+ val created: Future[Seq[Path]] =
+ Source(Seq(dir.resolve("dirA"), dir.resolve("dirB")))
.via(flow)
.runWith(Sink.seq)
// #mkdirs
@@ -118,8 +117,8 @@ class DirectorySpec
val flowWithContext: FlowWithContext[Path, SomeContext, Path, SomeContext, NotUsed] =
Directory.mkdirsWithContext[SomeContext]()
// #mkdirs
- val created: Future[immutable.Seq[(Any, Any)]] =
- Source(immutable.Seq(dir.resolve("dirA"), dir.resolve("dirB")))
+ val created: Future[Seq[(Any, Any)]] =
+ Source(Seq(dir.resolve("dirA"), dir.resolve("dirB")))
.asSourceWithContext(_ => SomeContext())
.via(flowWithContext)
.runWith(Sink.seq)
diff --git a/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala b/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala
index 14489b7c5..d4c1ae985 100644
--- a/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala
+++ b/file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala
@@ -32,7 +32,6 @@ import com.google.common.jimfs.{ Configuration, Jimfs }
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.BeforeAndAfterAll
-import scala.collection.immutable
import scala.collection.mutable.ListBuffer
import scala.concurrent.duration._
import scala.concurrent.{ Await, ExecutionContext, Future, Promise }
@@ -129,7 +128,7 @@ class LogRotatorSinkSpec
LogRotatorSink(fileSizeTriggerCreator)
// #size
- val fileSizeCompletion = Source(immutable.Seq("test1", "test2", "test3", "test4", "test5", "test6"))
+ val fileSizeCompletion = Source(Seq("test1", "test2", "test3", "test4", "test5", "test6"))
.map(ByteString(_))
.runWith(sizeRotatorSink)
@@ -162,7 +161,7 @@ class LogRotatorSinkSpec
LogRotatorSink(timeBasedTriggerCreator)
// #time
- val timeBaseCompletion = Source(immutable.Seq("test1", "test2", "test3", "test4", "test5", "test6"))
+ val timeBaseCompletion = Source(Seq("test1", "test2", "test3", "test4", "test5", "test6"))
.map(ByteString(_))
.runWith(timeBasedSink)
@@ -174,7 +173,7 @@ class LogRotatorSinkSpec
*/
val triggerFunctionCreator: () => ByteString => Option[Path] = timeBasedTriggerCreator
// #sample
- val completion = Source(immutable.Seq("test1", "test2", "test3", "test4", "test5", "test6"))
+ val completion = Source(Seq("test1", "test2", "test3", "test4", "test5", "test6"))
.map(ByteString(_))
.runWith(LogRotatorSink(triggerFunctionCreator))
// #sample
@@ -211,7 +210,7 @@ class LogRotatorSinkSpec
// #stream
val timeBaseCompletion = Source(
- immutable.Seq(
+ Seq(
("stream1", "test1"), ("stream1", "test2"), ("stream1", "test3"), ("stream2", "test4"), ("stream2", "test5"),
("stream2", "test6"))).runWith(timeBasedSink)
diff --git a/file/src/test/scala/docs/scaladsl/TarArchiveSpec.scala b/file/src/test/scala/docs/scaladsl/TarArchiveSpec.scala
index 60be20713..0dba227cb 100644
--- a/file/src/test/scala/docs/scaladsl/TarArchiveSpec.scala
+++ b/file/src/test/scala/docs/scaladsl/TarArchiveSpec.scala
@@ -32,7 +32,6 @@ import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
import scala.annotation.nowarn
-import scala.collection.immutable
import scala.concurrent.{ ExecutionContext, Future }
import scala.jdk.CollectionConverters._
@@ -239,7 +238,7 @@ class TarArchiveSpec
val metadata1 = TarArchiveMetadata(name1, file1.length)
val metadata2 = TarArchiveMetadata(name2, file2.length)
val tar = Source(
- immutable.Seq(
+ Seq(
metadata1 -> Source.single(file1),
metadata2 -> Source.single(file2))).via(Archive.tar())
// emit in short byte strings
@@ -261,7 +260,7 @@ class TarArchiveSpec
val metadata1 = TarArchiveMetadata("empty.txt", 0)
val metadata2 = TarArchiveMetadata("file2.txt", tenDigits.length)
val tarFile = Source(
- immutable.Seq(
+ Seq(
metadata1 -> Source.single(ByteString.empty),
metadata2 -> Source.single(tenDigits))).via(Archive.tar())
// swallow the whole input as one ByteString
@@ -350,7 +349,7 @@ class TarArchiveSpec
// Create a TAR archive with 3 files, emitted as a single ByteString
// This simulates an S3 stream that closes after sending all data at once
val multiFileArchive = Source(
- immutable.Seq(
+ Seq(
metadata1 -> Source.single(file1Content),
metadata2 -> Source.single(file2Content),
metadata3 -> Source.single(file3Content)
@@ -486,7 +485,7 @@ class TarArchiveSpec
private def getPathFromResources(fileName: String): Path =
Paths.get(getClass.getClassLoader.getResource(fileName).toURI)
- private def generateInputFiles(numberOfFiles: Int, lengthOfFile: Int): immutable.Seq[(String, ByteString)] = {
+ private def generateInputFiles(numberOfFiles: Int, lengthOfFile: Int): Seq[(String, ByteString)] = {
val r = new scala.util.Random(31)
(1 to numberOfFiles)
.map(number => s"file-$number" -> ByteString.fromArray(r.nextString(lengthOfFile).getBytes))
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/CommonFtpOperations.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/CommonFtpOperations.scala
index b3af9eddb..893cb29b0 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/CommonFtpOperations.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/CommonFtpOperations.scala
@@ -23,7 +23,6 @@ import pekko.annotation.InternalApi
import pekko.stream.connectors.ftp.FtpFile
import org.apache.commons.net.ftp.{ FTPClient, FTPFile }
-import scala.collection.immutable
import scala.util.Try
/**
@@ -33,7 +32,7 @@ import scala.util.Try
private[ftp] trait CommonFtpOperations {
type Handler = FTPClient
- def listFiles(basePath: String, handler: Handler): immutable.Seq[FtpFile] = {
+ def listFiles(basePath: String, handler: Handler): Seq[FtpFile] = {
val path = if (basePath.nonEmpty && basePath.head != '/') s"/$basePath" else if (basePath == "/") "" else basePath
handler
.listFiles(path)
@@ -70,7 +69,7 @@ private[ftp] trait CommonFtpOperations {
case (perm, true) => perm
}.toSet
- def listFiles(handler: Handler): immutable.Seq[FtpFile] = listFiles("", handler)
+ def listFiles(handler: Handler): Seq[FtpFile] = listFiles("", handler)
def retrieveFileInputStream(name: String, handler: Handler): Try[InputStream] =
retrieveFileInputStream(name, handler, 0L)
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpLike.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpLike.scala
index e4e3211d0..66c6ab26f 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpLike.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpLike.scala
@@ -17,7 +17,6 @@ package impl
import net.schmizz.sshj.SSHClient
import org.apache.commons.net.ftp.{ FTPClient, FTPSClient }
-import scala.collection.immutable
import scala.util.Try
import java.io.{ InputStream, OutputStream }
@@ -35,9 +34,9 @@ protected[ftp] trait FtpLike[FtpClient, S <: RemoteFileSettings] {
def disconnect(handler: Handler)(implicit ftpClient: FtpClient): Unit
- def listFiles(basePath: String, handler: Handler): immutable.Seq[FtpFile]
+ def listFiles(basePath: String, handler: Handler): Seq[FtpFile]
- def listFiles(handler: Handler): immutable.Seq[FtpFile]
+ def listFiles(handler: Handler): Seq[FtpFile]
def retrieveFileInputStream(name: String, handler: Handler): Try[InputStream]
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala
index 9ea4d32ab..69a6a21c0 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala
@@ -29,7 +29,6 @@ import net.schmizz.sshj.userauth.password.{ PasswordFinder, PasswordUtils, Resou
import net.schmizz.sshj.xfer.FilePermission
import org.apache.commons.net.DefaultSocketFactory
-import scala.collection.immutable
import scala.jdk.CollectionConverters._
import scala.util.{ Failure, Try }
@@ -89,7 +88,7 @@ private[ftp] trait SftpOperations { self: FtpLike[SSHClient, SftpSettings] =>
if (ssh.isConnected) ssh.disconnect()
}
- def listFiles(basePath: String, handler: Handler): immutable.Seq[FtpFile] = {
+ def listFiles(basePath: String, handler: Handler): Seq[FtpFile] = {
val path = if (basePath.nonEmpty && basePath.head != '/') s"/$basePath" else basePath
val entries = handler.ls(path).asScala
entries.map { file =>
@@ -125,7 +124,7 @@ private[ftp] trait SftpOperations { self: FtpLike[SSHClient, SftpSettings] =>
}.toSet
}
- def listFiles(handler: Handler): immutable.Seq[FtpFile] = listFiles(".", handler)
+ def listFiles(handler: Handler): Seq[FtpFile] = listFiles(".", handler)
def retrieveFileInputStream(name: String, handler: Handler): Try[InputStream] =
retrieveFileInputStream(name, handler, 0L)
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 cb34d5561..cd7485661 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
@@ -30,7 +30,6 @@ import pekko.util.ByteString
import org.scalatest.concurrent.Eventually
import org.scalatest.time.{ Millis, Seconds, Span }
-import scala.collection.immutable
import scala.concurrent.{ Await, ExecutionContext, ExecutionContextExecutor }
import scala.concurrent.duration._
import scala.util.Random
@@ -479,7 +478,7 @@ trait CommonFtpStageSpec extends BaseSpec with Eventually {
branchSelector = _ => true,
emitTraversedDirectories = true).runWith(Sink.seq)
- val listingRes: immutable.Seq[FtpFile] = Await.result(listing, Duration(1, TimeUnit.MINUTES))
+ val listingRes: Seq[FtpFile] = Await.result(listing, Duration(1, TimeUnit.MINUTES))
listingRes.map(_.name) should contain allElementsOf Seq(name)
}
@@ -507,7 +506,7 @@ trait CommonFtpStageSpec extends BaseSpec with Eventually {
branchSelector = _ => true,
emitTraversedDirectories = true).runWith(Sink.seq)
- val listingRes: immutable.Seq[FtpFile] = Await.result(listing, Duration(1, TimeUnit.MINUTES))
+ val listingRes: Seq[FtpFile] = Await.result(listing, Duration(1, TimeUnit.MINUTES))
listingRes.map(_.name) should contain allElementsOf Seq(name)
@@ -516,7 +515,7 @@ trait CommonFtpStageSpec extends BaseSpec with Eventually {
branchSelector = _ => true,
emitTraversedDirectories = true).runWith(Sink.seq)
- val listingInnerDirRes: immutable.Seq[FtpFile] =
+ val listingInnerDirRes: Seq[FtpFile] =
Await.result(listingOnlyInnerDir, Duration(1, TimeUnit.MINUTES))
listingInnerDirRes.map(_.name) should contain allElementsOf Seq(innerDirName)
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 e0edffff4..fc450d131 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
@@ -20,7 +20,6 @@ import pekko.stream.connectors.googlecloud.bigquery.scaladsl.spray.BigQueryRestJ
import spray.json.{ JsonFormat, RootJsonFormat }
import java.util
-import scala.collection.immutable.Seq
import scala.jdk.CollectionConverters._
import scala.jdk.OptionConverters._
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 96fb89bc3..b1fcb7d44 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
@@ -21,7 +21,6 @@ import spray.json.{ JsonFormat, RootJsonFormat }
import java.util
import scala.annotation.nowarn
-import scala.collection.immutable.Seq
import scala.jdk.CollectionConverters._
import scala.jdk.OptionConverters._
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 16aaf8e26..f4282a172 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
@@ -25,7 +25,6 @@ import java.time.Duration
import java.{ lang, util }
import scala.annotation.nowarn
import scala.annotation.unchecked.uncheckedVariance
-import scala.collection.immutable.Seq
import scala.concurrent.duration.FiniteDuration
import scala.jdk.CollectionConverters._
import scala.jdk.OptionConverters._
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 1d1906fd8..9b6733d77 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 spray.json.{ JsonFormat, RootJsonFormat, RootJsonReader, RootJsonWriter }
import java.{ lang, util }
import scala.annotation.nowarn
import scala.annotation.unchecked.uncheckedVariance
-import scala.collection.immutable.Seq
import scala.jdk.CollectionConverters._
import scala.jdk.OptionConverters._
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 c44d95c82..ddc457f0a 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
@@ -23,7 +23,6 @@ import java.util
import scala.annotation.nowarn
import scala.annotation.varargs
-import scala.collection.immutable.Seq
import scala.jdk.CollectionConverters._
import scala.jdk.OptionConverters._
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 f24b6c1f0..f1876422b 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
@@ -36,7 +36,6 @@ import pekko.stream.scaladsl.{ Flow, Keep, Sink, Source }
import java.util.{ SplittableRandom, UUID }
-import scala.collection.immutable.Seq
import scala.concurrent.{ ExecutionContext, Future }
private[scaladsl] trait BigQueryTableData { this: BigQueryRest =>
diff --git a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/schema/ProductSchemas.scala b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/schema/ProductSchemas.scala
index 01e6c42e6..d32209ed2 100644
--- a/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/schema/ProductSchemas.scala
+++ b/google-cloud-bigquery/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/bigquery/scaladsl/schema/ProductSchemas.scala
@@ -18,7 +18,6 @@ import pekko.stream.connectors.googlecloud.bigquery.model.TableFieldSchemaType.R
import pekko.stream.connectors.googlecloud.bigquery.model.{ TableFieldSchema, TableFieldSchemaMode, TableSchema }
import spray.json.{ AdditionalFormats, ProductFormats, StandardFormats }
-import scala.collection.immutable.Seq
import scala.reflect.ClassTag
/**
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 2c71e35e2..8bd4f70be 100644
--- a/google-cloud-bigquery/src/test/scala/docs/scaladsl/BigQueryDoc.scala
+++ b/google-cloud-bigquery/src/test/scala/docs/scaladsl/BigQueryDoc.scala
@@ -37,7 +37,6 @@ import pekko.stream.scaladsl.{ Flow, Sink, Source }
import pekko.{ Done, NotUsed }
import scala.annotation.nowarn
-import scala.collection.immutable.Seq
import scala.concurrent.Future
//#imports
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 f4c5dce78..6f07785bb 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
@@ -113,12 +113,12 @@ private[pubsub] trait PubSubApi {
private implicit val pubSubRequestFormat: RootJsonFormat[PublishRequest] = new RootJsonFormat[PublishRequest] {
def read(json: JsValue): PublishRequest =
- PublishRequest(json.asJsObject.fields("messages").convertTo[immutable.Seq[PublishMessage]])
+ PublishRequest(json.asJsObject.fields("messages").convertTo[Seq[PublishMessage]])
def write(pr: PublishRequest): JsValue = JsObject("messages" -> pr.messages.toJson)
}
private implicit val gcePubSubResponseFormat: RootJsonFormat[PublishResponse] = new RootJsonFormat[PublishResponse] {
def read(json: JsValue): PublishResponse =
- PublishResponse(json.asJsObject.fields("messageIds").convertTo[immutable.Seq[String]])
+ PublishResponse(json.asJsObject.fields("messageIds").convertTo[Seq[String]])
def write(pr: PublishResponse): JsValue = JsObject("messageIds" -> pr.messageIds.toJson)
}
@@ -131,7 +131,7 @@ private[pubsub] trait PubSubApi {
}
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]]))
+ PullResponse(json.asJsObject.fields.get("receivedMessages").map(_.convertTo[Seq[ReceivedMessage]]))
def write(pr: PullResponse): JsValue =
pr.receivedMessages.map(rm => JsObject("receivedMessages" -> rm.toJson)).getOrElse(JsObject.empty)
}
@@ -139,7 +139,7 @@ private[pubsub] trait PubSubApi {
private implicit val acknowledgeRequestFormat: RootJsonFormat[AcknowledgeRequest] =
new RootJsonFormat[AcknowledgeRequest] {
def read(json: JsValue): AcknowledgeRequest =
- AcknowledgeRequest(json.asJsObject.fields("ackIds").convertTo[immutable.Seq[String]]: _*)
+ AcknowledgeRequest(json.asJsObject.fields("ackIds").convertTo[Seq[String]]: _*)
def write(ar: AcknowledgeRequest): JsValue = JsObject("ackIds" -> ar.ackIds.toJson)
}
private implicit val pullRequestFormat: RootJsonFormat[PullRequest] =
diff --git a/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/model.scala b/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/model.scala
index 957d2dfaf..556b22cf1 100644
--- a/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/model.scala
+++ b/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/model.scala
@@ -169,7 +169,7 @@ object PubSubMessage {
}
-final class PublishRequest private (val messages: immutable.Seq[PublishMessage]) {
+final class PublishRequest private (val messages: Seq[PublishMessage]) {
override def equals(other: Any): Boolean = other match {
case that: PublishRequest => messages == that.messages
@@ -183,7 +183,7 @@ final class PublishRequest private (val messages: immutable.Seq[PublishMessage])
object PublishRequest {
- def apply(messages: immutable.Seq[PublishMessage]): PublishRequest = new PublishRequest(messages)
+ def apply(messages: Seq[PublishMessage]): PublishRequest = new PublishRequest(messages)
/**
* Java API
@@ -218,7 +218,7 @@ object ReceivedMessage {
new ReceivedMessage(ackId, message)
}
-final class AcknowledgeRequest private (val ackIds: immutable.Seq[String]) {
+final class AcknowledgeRequest private (val ackIds: Seq[String]) {
override def equals(other: Any): Boolean = other match {
case that: AcknowledgeRequest => ackIds == that.ackIds
@@ -242,7 +242,7 @@ object AcknowledgeRequest {
new AcknowledgeRequest(ackIds.asScala.toList)
}
-private final class PublishResponse private (val messageIds: immutable.Seq[String]) {
+private final class PublishResponse private (val messageIds: Seq[String]) {
override def equals(other: Any): Boolean = other match {
case that: PublishResponse => messageIds == that.messageIds
@@ -256,7 +256,7 @@ private final class PublishResponse private (val messageIds: immutable.Seq[Strin
object PublishResponse {
- @InternalApi private[pubsub] def apply(messageIds: immutable.Seq[String]): PublishResponse =
+ @InternalApi private[pubsub] def apply(messageIds: Seq[String]): PublishResponse =
new PublishResponse(messageIds)
}
@@ -264,7 +264,7 @@ object PublishResponse {
private[pubsub] final case class PullRequest(returnImmediately: Boolean, maxMessages: Int)
@InternalApi
-private final class PullResponse private[pubsub] (val receivedMessages: Option[immutable.Seq[ReceivedMessage]]) {
+private final class PullResponse private[pubsub] (val receivedMessages: Option[Seq[ReceivedMessage]]) {
override def equals(other: Any): Boolean = other match {
case that: PullResponse => receivedMessages == that.receivedMessages
@@ -278,7 +278,7 @@ private final class PullResponse private[pubsub] (val receivedMessages: Option[i
object PullResponse {
- @InternalApi private[pubsub] def apply(receivedMessages: Option[immutable.Seq[ReceivedMessage]]) =
+ @InternalApi private[pubsub] def apply(receivedMessages: Option[Seq[ReceivedMessage]]) =
new PullResponse(receivedMessages)
}
diff --git a/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/scaladsl/GooglePubSub.scala b/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/scaladsl/GooglePubSub.scala
index b0d7a7ad2..97610b8a6 100644
--- a/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/scaladsl/GooglePubSub.scala
+++ b/google-cloud-pub-sub/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/scaladsl/GooglePubSub.scala
@@ -20,7 +20,6 @@ import pekko.stream.connectors.googlecloud.pubsub.impl._
import pekko.stream.scaladsl.{ Flow, FlowWithContext, Keep, Sink, Source }
import pekko.{ Done, NotUsed }
-import scala.collection.immutable
import scala.concurrent.Future
import scala.concurrent.duration._
@@ -43,7 +42,7 @@ protected[pubsub] trait GooglePubSub {
def publish(topic: String,
config: PubSubConfig,
overrideHost: String,
- parallelism: Int): Flow[PublishRequest, immutable.Seq[String], NotUsed] =
+ parallelism: Int): Flow[PublishRequest, Seq[String], NotUsed] =
internalPublish(topic, config, Some(overrideHost), parallelism)
/**
@@ -54,7 +53,7 @@ protected[pubsub] trait GooglePubSub {
*/
def publish(topic: String,
config: PubSubConfig,
- overrideHost: String): Flow[PublishRequest, immutable.Seq[String], NotUsed] =
+ overrideHost: String): Flow[PublishRequest, Seq[String], NotUsed] =
internalPublish(topic, config, Some(overrideHost), parallelism = 1)
/**
@@ -62,13 +61,13 @@ protected[pubsub] trait GooglePubSub {
*/
def publish(topic: String,
config: PubSubConfig,
- parallelism: Int = 1): Flow[PublishRequest, immutable.Seq[String], NotUsed] =
+ parallelism: Int = 1): Flow[PublishRequest, Seq[String], NotUsed] =
internalPublish(topic, config, None, parallelism)
private def internalPublish(topic: String,
config: PubSubConfig,
overrideHost: Option[String],
- parallelism: Int): Flow[PublishRequest, immutable.Seq[String], NotUsed] =
+ parallelism: Int): Flow[PublishRequest, Seq[String], NotUsed] =
Flow[PublishRequest]
.map((_, ()))
.via(
@@ -86,7 +85,7 @@ protected[pubsub] trait GooglePubSub {
topic: String,
config: PubSubConfig,
overrideHost: String,
- parallelism: Int): FlowWithContext[PublishRequest, C, immutable.Seq[String], C, NotUsed] =
+ parallelism: Int): FlowWithContext[PublishRequest, C, Seq[String], C, NotUsed] =
internalPublishWithContext(topic, config, Some(overrideHost), parallelism)
/**
@@ -99,7 +98,7 @@ protected[pubsub] trait GooglePubSub {
def publishWithContext[C](
topic: String,
config: PubSubConfig,
- overrideHost: String): FlowWithContext[PublishRequest, C, immutable.Seq[String], C, NotUsed] =
+ overrideHost: String): FlowWithContext[PublishRequest, C, Seq[String], C, NotUsed] =
internalPublishWithContext(topic, config, Some(overrideHost), parallelism = 1)
/**
@@ -109,14 +108,14 @@ protected[pubsub] trait GooglePubSub {
def publishWithContext[C](
topic: String,
config: PubSubConfig,
- parallelism: Int = 1): FlowWithContext[PublishRequest, C, immutable.Seq[String], C, NotUsed] =
+ parallelism: Int = 1): FlowWithContext[PublishRequest, C, Seq[String], C, NotUsed] =
internalPublishWithContext(topic, config, None, parallelism)
private def internalPublishWithContext[C](
topic: String,
config: PubSubConfig,
overrideHost: Option[String],
- parallelism: Int): FlowWithContext[PublishRequest, C, immutable.Seq[String], C, NotUsed] =
+ parallelism: Int): FlowWithContext[PublishRequest, C, Seq[String], C, NotUsed] =
// some wrapping back and forth as FlowWithContext doesn't offer `setup`
// https://github.com/akka/akka/issues/27883
FlowWithContext
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 d3928b8a2..ed327e329 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
@@ -24,7 +24,6 @@ import pekko.stream.connectors.googlecloud.pubsub.scaladsl.GooglePubSub
import pekko.stream.scaladsl.{ Flow, FlowWithContext, RestartFlow, Sink, Source }
import pekko.{ Done, NotUsed }
-import scala.collection.immutable.Seq
import scala.concurrent.duration._
import scala.concurrent.{ Future, Promise }
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 b563f41c1..cb010e3bf 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
@@ -31,7 +31,6 @@ import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec
import org.scalatest.{ BeforeAndAfterAll, OptionValues }
-import scala.collection.immutable.Seq
import scala.concurrent.Future
import scala.concurrent.duration._
diff --git a/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/google/AuthorizationHeaderSpec.scala b/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/google/AuthorizationHeaderSpec.scala
index 55b75b319..88c1ad006 100644
--- a/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/google/AuthorizationHeaderSpec.scala
+++ b/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/google/AuthorizationHeaderSpec.scala
@@ -38,7 +38,6 @@ import org.scalatestplus.mockito.MockitoSugar
import java.util.Base64
import scala.concurrent.{ ExecutionContext, Future }
-import scala.collection.immutable.Seq
import scala.concurrent.duration._
class AuthorizationHeaderSpec extends AnyFlatSpec with BeforeAndAfterAll with ScalaFutures with Matchers
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 03c179ca5..0de56e4cc 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
@@ -31,7 +31,6 @@ import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
import org.scalatestplus.mockito.MockitoSugar
-import scala.collection.immutable.Seq
import scala.concurrent.duration._
class GooglePubSubSpec
diff --git a/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/ModelSpec.scala b/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/ModelSpec.scala
index adf249271..754849505 100644
--- a/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/ModelSpec.scala
+++ b/google-cloud-pub-sub/src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/ModelSpec.scala
@@ -15,7 +15,6 @@ package org.apache.pekko.stream.connectors.googlecloud.pubsub
import org.scalatest.funsuite.AnyFunSuite
import org.scalatest.matchers.should.Matchers
-import scala.collection.immutable.Seq
import java.time.Instant
import org.apache.pekko.stream.connectors.testkit.scaladsl.LogCapturing
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 c01634bc6..5ae463cb8 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
@@ -31,7 +31,6 @@ import java.security.cert.X509Certificate
import java.time.Instant
import java.util.Base64
import javax.net.ssl.X509TrustManager
-import scala.collection.immutable.Seq
import scala.concurrent.Await
import scala.concurrent.duration._
diff --git a/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/FailedUpload.scala b/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/FailedUpload.scala
index 016c9c46c..2ac264a86 100644
--- a/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/FailedUpload.scala
+++ b/google-cloud-storage/src/main/scala/org/apache/pekko/stream/connectors/googlecloud/storage/FailedUpload.scala
@@ -13,7 +13,6 @@
package org.apache.pekko.stream.connectors.googlecloud.storage
-import scala.collection.immutable.Seq
import scala.jdk.CollectionConverters._
final class FailedUpload private (
diff --git a/google-fcm/src/test/scala/docs/scaladsl/FcmExamples.scala b/google-fcm/src/test/scala/docs/scaladsl/FcmExamples.scala
index 8b8e46a51..d4bc1957f 100644
--- a/google-fcm/src/test/scala/docs/scaladsl/FcmExamples.scala
+++ b/google-fcm/src/test/scala/docs/scaladsl/FcmExamples.scala
@@ -23,7 +23,6 @@ import pekko.stream.connectors.google.firebase.fcm.v1.scaladsl.GoogleFcm
//#imports
import pekko.stream.scaladsl.{ Sink, Source }
-import scala.collection.immutable
import scala.concurrent.Future
class FcmExamples {
@@ -39,7 +38,7 @@ class FcmExamples {
// #simple-send
// #asFlow-send
- val result1: Future[immutable.Seq[FcmResponse]] =
+ val result1: Future[Seq[FcmResponse]] =
Source
.single(notification)
.via(GoogleFcm.send(fcmConfig))
@@ -55,7 +54,7 @@ class FcmExamples {
// #asFlow-send
// #withData-send
- val result2: Future[immutable.Seq[(FcmResponse, String)]] =
+ val result2: Future[Seq[(FcmResponse, String)]] =
Source
.single((notification, "superData"))
.via(GoogleFcm.sendWithPassThrough(fcmConfig))
diff --git a/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/HTableSettings.scala b/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/HTableSettings.scala
index 335763695..75997985d 100644
--- a/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/HTableSettings.scala
+++ b/hbase/src/main/scala/org/apache/pekko/stream/connectors/hbase/HTableSettings.scala
@@ -17,14 +17,13 @@ import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client.Mutation
-import scala.collection.immutable
import scala.jdk.CollectionConverters._
import scala.jdk.FunctionConverters._
final class HTableSettings[T] private (val conf: Configuration,
val tableName: TableName,
- val columnFamilies: immutable.Seq[String],
- val converter: T => immutable.Seq[Mutation]) {
+ val columnFamilies: Seq[String],
+ val converter: T => Seq[Mutation]) {
def withConf(conf: Configuration): HTableSettings[T] =
copy(conf = conf)
@@ -32,7 +31,7 @@ final class HTableSettings[T] private (val conf: Configuration,
def withTableName(tableName: TableName): HTableSettings[T] =
copy(tableName = tableName)
- def withColumnFamilies(columnFamilies: immutable.Seq[String]): HTableSettings[T] =
+ def withColumnFamilies(columnFamilies: Seq[String]): HTableSettings[T] =
copy(columnFamilies = columnFamilies)
/**
@@ -41,7 +40,7 @@ final class HTableSettings[T] private (val conf: Configuration,
def withColumnFamilies(columnFamilies: java.util.List[String]): HTableSettings[T] =
copy(columnFamilies = columnFamilies.asScala.toIndexedSeq)
- def withConverter(converter: T => immutable.Seq[Mutation]): HTableSettings[T] =
+ def withConverter(converter: T => Seq[Mutation]): HTableSettings[T] =
copy(converter = converter)
/**
@@ -60,8 +59,8 @@ final class HTableSettings[T] private (val conf: Configuration,
private def copy(conf: Configuration = conf,
tableName: TableName = tableName,
- columnFamilies: immutable.Seq[String] = columnFamilies,
- converter: T => immutable.Seq[Mutation] = converter) =
+ columnFamilies: Seq[String] = columnFamilies,
+ converter: T => Seq[Mutation] = converter) =
new HTableSettings[T](conf, tableName, columnFamilies, converter)
}
@@ -73,8 +72,8 @@ object HTableSettings {
*/
def apply[T](conf: Configuration,
tableName: TableName,
- columnFamilies: immutable.Seq[String],
- converter: T => immutable.Seq[Mutation]) =
+ columnFamilies: Seq[String],
+ converter: T => Seq[Mutation]) =
new HTableSettings(conf, tableName, columnFamilies, converter)
/**
diff --git a/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala b/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
index 8ae2b7c07..8d54461d4 100644
--- a/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
+++ b/hbase/src/test/scala/docs/scaladsl/HBaseStageSpec.scala
@@ -27,7 +27,6 @@ import org.apache.hadoop.hbase.{ HBaseConfiguration, TableName }
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.BeforeAndAfterAll
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.language.implicitConversions
import scala.util.Random
@@ -49,7 +48,7 @@ class HBaseStageSpec
implicit def toBytes(string: String): Array[Byte] = Bytes.toBytes(string)
case class Person(id: Int, name: String)
- val hBaseConverter: Person => immutable.Seq[Mutation] = { person =>
+ val hBaseConverter: Person => Seq[Mutation] = { person =>
val put = new Put(s"id_${person.id}")
put.addColumn("info", "name", person.name)
List(put)
@@ -57,7 +56,7 @@ class HBaseStageSpec
// #create-converter-put
// #create-converter-append
- val appendHBaseConverter: Person => immutable.Seq[Mutation] = { person =>
+ val appendHBaseConverter: Person => Seq[Mutation] = { person =>
// Append to a cell
val append = new Append(s"id_${person.id}")
append.addColumn("info", "aliases", person.name)
@@ -66,7 +65,7 @@ class HBaseStageSpec
// #create-converter-append
// #create-converter-delete
- val deleteHBaseConverter: Person => immutable.Seq[Mutation] = { person =>
+ val deleteHBaseConverter: Person => Seq[Mutation] = { person =>
// Delete the specified row
val delete = new Delete(s"id_${person.id}")
List(delete)
@@ -74,7 +73,7 @@ class HBaseStageSpec
// #create-converter-delete
// #create-converter-increment
- val incrementHBaseConverter: Person => immutable.Seq[Mutation] = { person =>
+ val incrementHBaseConverter: Person => Seq[Mutation] = { person =>
// Increment a cell value
val increment = new Increment(s"id_${person.id}")
increment.addColumn("info", "numberOfChanges", 1)
@@ -83,7 +82,7 @@ class HBaseStageSpec
// #create-converter-increment
// #create-converter-complex
- val mutationsHBaseConverter: Person => immutable.Seq[Mutation] = { person =>
+ val mutationsHBaseConverter: Person => Seq[Mutation] = { person =>
if (person.id != 0) {
if (person.name.isEmpty) {
// Delete the specified row
@@ -107,7 +106,7 @@ class HBaseStageSpec
// #create-settings
val tableSettings =
- HTableSettings(HBaseConfiguration.create(), TableName.valueOf("person"), immutable.Seq("info"), hBaseConverter)
+ HTableSettings(HBaseConfiguration.create(), TableName.valueOf("person"), Seq("info"), hBaseConverter)
// #create-settings
"HBase stage" must {
diff --git a/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSender.scala b/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSender.scala
index 92cbfc698..0f75e1f08 100644
--- a/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSender.scala
+++ b/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/PushKitSender.scala
@@ -27,7 +27,6 @@ import pekko.stream.connectors.huawei.pushkit.HmsSettings
import pekko.stream.connectors.huawei.pushkit.models.{ ErrorResponse, PushKitResponse, Response }
import spray.json.enrichAny
-import scala.collection.immutable
import scala.concurrent.{ ExecutionContext, Future }
/**
@@ -49,7 +48,7 @@ private[pushkit] class PushKitSender {
HttpRequest(
HttpMethods.POST,
url,
- immutable.Seq(Authorization(OAuth2BearerToken(token))),
+ Seq(Authorization(OAuth2BearerToken(token))),
HttpEntity(ContentTypes.`application/json`, hmsSend.toJson.compactPrint)),
connectionContext = fp.httpsContext(system),
settings = fp.poolSettings(system))
@@ -58,7 +57,7 @@ private[pushkit] class PushKitSender {
HttpRequest(
HttpMethods.POST,
url,
- immutable.Seq(Authorization(OAuth2BearerToken(token))),
+ Seq(Authorization(OAuth2BearerToken(token))),
HttpEntity(ContentTypes.`application/json`, hmsSend.toJson.compactPrint)))
}
parse(response)
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 7a6d0f5e9..7b23042e9 100644
--- a/huawei-push-kit/src/test/scala/docs/scaladsl/PushKitExamples.scala
+++ b/huawei-push-kit/src/test/scala/docs/scaladsl/PushKitExamples.scala
@@ -33,7 +33,6 @@ import pekko.stream.connectors.huawei.pushkit.models.Tokens
import pekko.stream.scaladsl.Source
import pekko.stream.scaladsl.Sink
-import scala.collection.immutable
import scala.concurrent.Future
class PushKitExamples {
@@ -63,7 +62,7 @@ class PushKitExamples {
// #simple-send
// #asFlow-send
- val result1: Future[immutable.Seq[Response]] =
+ val result1: Future[Seq[Response]] =
Source
.single(notification)
.via(HmsPushKit.send(config))
diff --git a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/InfluxDbFlowStage.scala b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/InfluxDbFlowStage.scala
index b32eaf112..4a2f37f50 100644
--- a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/InfluxDbFlowStage.scala
+++ b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/impl/InfluxDbFlowStage.scala
@@ -20,7 +20,6 @@ import pekko.stream.connectors.influxdb.{ InfluxDbWriteMessage, InfluxDbWriteRes
import pekko.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler }
import org.influxdb.InfluxDB
-import scala.collection.immutable
import org.influxdb.dto.{ BatchPoints, Point }
import scala.annotation.tailrec
@@ -32,9 +31,9 @@ import scala.annotation.tailrec
private[influxdb] class InfluxDbFlowStage[C](
influxDB: InfluxDB)
extends GraphStage[
- FlowShape[immutable.Seq[InfluxDbWriteMessage[Point, C]], immutable.Seq[InfluxDbWriteResult[Point, C]]]] {
- private val in = Inlet[immutable.Seq[InfluxDbWriteMessage[Point, C]]]("in")
- private val out = Outlet[immutable.Seq[InfluxDbWriteResult[Point, C]]]("out")
+ FlowShape[Seq[InfluxDbWriteMessage[Point, C]], Seq[InfluxDbWriteResult[Point, C]]]] {
+ private val in = Inlet[Seq[InfluxDbWriteMessage[Point, C]]]("in")
+ private val out = Outlet[Seq[InfluxDbWriteResult[Point, C]]]("out")
override val shape = FlowShape(in, out)
@@ -53,10 +52,10 @@ private[influxdb] class InfluxDbFlowStage[C](
private[influxdb] class InfluxDbMapperFlowStage[T, C](
clazz: Class[T],
influxDB: InfluxDB)
- extends GraphStage[FlowShape[immutable.Seq[InfluxDbWriteMessage[T, C]], immutable.Seq[InfluxDbWriteResult[T, C]]]] {
+ extends GraphStage[FlowShape[Seq[InfluxDbWriteMessage[T, C]], Seq[InfluxDbWriteResult[T, C]]]] {
- private val in = Inlet[immutable.Seq[InfluxDbWriteMessage[T, C]]]("in")
- private val out = Outlet[immutable.Seq[InfluxDbWriteResult[T, C]]]("out")
+ private val in = Inlet[Seq[InfluxDbWriteMessage[T, C]]]("in")
+ private val out = Outlet[Seq[InfluxDbWriteResult[T, C]]]("out")
override val shape = FlowShape(in, out)
@@ -71,16 +70,16 @@ private[influxdb] class InfluxDbMapperFlowStage[T, C](
@InternalApi
private[influxdb] sealed abstract class InfluxDbLogic[T, C](
influxDB: InfluxDB,
- in: Inlet[immutable.Seq[InfluxDbWriteMessage[T, C]]],
- out: Outlet[immutable.Seq[InfluxDbWriteResult[T, C]]],
- shape: FlowShape[immutable.Seq[InfluxDbWriteMessage[T, C]], immutable.Seq[InfluxDbWriteResult[T, C]]])
+ in: Inlet[Seq[InfluxDbWriteMessage[T, C]]],
+ out: Outlet[Seq[InfluxDbWriteResult[T, C]]],
+ shape: FlowShape[Seq[InfluxDbWriteMessage[T, C]], Seq[InfluxDbWriteResult[T, C]]])
extends GraphStageLogic(shape)
with InHandler
with OutHandler {
setHandlers(in, out, this)
- protected def write(messages: immutable.Seq[InfluxDbWriteMessage[T, C]]): Unit
+ protected def write(messages: Seq[InfluxDbWriteMessage[T, C]]): Unit
override def onPull(): Unit = if (!isClosed(in) && !hasBeenPulled(in)) pull(in)
@@ -122,12 +121,12 @@ private[influxdb] sealed abstract class InfluxDbLogic[T, C](
@InternalApi
private[influxdb] final class InfluxDbRecordLogic[C](
influxDB: InfluxDB,
- in: Inlet[immutable.Seq[InfluxDbWriteMessage[Point, C]]],
- out: Outlet[immutable.Seq[InfluxDbWriteResult[Point, C]]],
- shape: FlowShape[immutable.Seq[InfluxDbWriteMessage[Point, C]], immutable.Seq[InfluxDbWriteResult[Point, C]]])
+ in: Inlet[Seq[InfluxDbWriteMessage[Point, C]]],
+ out: Outlet[Seq[InfluxDbWriteResult[Point, C]]],
+ shape: FlowShape[Seq[InfluxDbWriteMessage[Point, C]], Seq[InfluxDbWriteResult[Point, C]]])
extends InfluxDbLogic(influxDB, in, out, shape) {
- override protected def write(messages: immutable.Seq[InfluxDbWriteMessage[Point, C]]): Unit =
+ override protected def write(messages: Seq[InfluxDbWriteMessage[Point, C]]): Unit =
messages
.groupBy(im => (im.databaseName, im.retentionPolicy))
.map(wm => toBatchPoints(wm._1._1, wm._1._2, wm._2))
@@ -140,14 +139,14 @@ private[influxdb] final class InfluxDbRecordLogic[C](
@InternalApi
private[influxdb] final class InfluxDbMapperRecordLogic[T, C](
influxDB: InfluxDB,
- in: Inlet[immutable.Seq[InfluxDbWriteMessage[T, C]]],
- out: Outlet[immutable.Seq[InfluxDbWriteResult[T, C]]],
- shape: FlowShape[immutable.Seq[InfluxDbWriteMessage[T, C]], immutable.Seq[InfluxDbWriteResult[T, C]]])
+ in: Inlet[Seq[InfluxDbWriteMessage[T, C]]],
+ out: Outlet[Seq[InfluxDbWriteResult[T, C]]],
+ shape: FlowShape[Seq[InfluxDbWriteMessage[T, C]], Seq[InfluxDbWriteResult[T, C]]])
extends InfluxDbLogic(influxDB, in, out, shape) {
private val mapperHelper: PekkoConnectorsResultMapperHelper = new PekkoConnectorsResultMapperHelper
- override protected def write(messages: immutable.Seq[InfluxDbWriteMessage[T, C]]): Unit =
+ override protected def write(messages: Seq[InfluxDbWriteMessage[T, C]]): Unit =
messages
.groupBy(groupByDbRp)
.map(convertToBatchPoints)
@@ -164,7 +163,7 @@ private[influxdb] final class InfluxDbMapperRecordLogic[T, C](
case None => Some(mapperHelper.retentionPolicy(im.point.getClass))
})
- def convertToBatchPoints(wm: ((Some[String], Some[String]), immutable.Seq[InfluxDbWriteMessage[T, C]])) =
+ def convertToBatchPoints(wm: ((Some[String], Some[String]), Seq[InfluxDbWriteMessage[T, C]])) =
toBatchPoints(wm._1._1,
wm._1._2,
wm._2.map(im => im.withPoint(mapperHelper.convertModelToPoint(im.point).asInstanceOf[T])))
diff --git a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/scaladsl/InfluxDbFlow.scala b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/scaladsl/InfluxDbFlow.scala
index 06caf0f5e..1f97ab658 100644
--- a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/scaladsl/InfluxDbFlow.scala
+++ b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/scaladsl/InfluxDbFlow.scala
@@ -21,8 +21,6 @@ import pekko.stream.scaladsl.Flow
import org.influxdb.InfluxDB
import org.influxdb.dto.Point
-import scala.collection.immutable
-
/**
* Scala API to create InfluxDB flows.
*
@@ -32,23 +30,23 @@ import scala.collection.immutable
object InfluxDbFlow {
def create()(
- implicit influxDB: InfluxDB): Flow[immutable.Seq[InfluxDbWriteMessage[Point, NotUsed]],
- immutable.Seq[InfluxDbWriteResult[Point, NotUsed]], NotUsed] =
+ implicit influxDB: InfluxDB): Flow[Seq[InfluxDbWriteMessage[Point, NotUsed]],
+ Seq[InfluxDbWriteResult[Point, NotUsed]], NotUsed] =
Flow.fromGraph(new impl.InfluxDbFlowStage[NotUsed](influxDB))
def typed[T](clazz: Class[T])(
implicit influxDB: InfluxDB)
- : Flow[immutable.Seq[InfluxDbWriteMessage[T, NotUsed]], immutable.Seq[InfluxDbWriteResult[T, NotUsed]], NotUsed] =
+ : Flow[Seq[InfluxDbWriteMessage[T, NotUsed]], Seq[InfluxDbWriteResult[T, NotUsed]], NotUsed] =
Flow.fromGraph(new impl.InfluxDbMapperFlowStage[T, NotUsed](clazz, influxDB))
def createWithPassThrough[C](
implicit influxDB: InfluxDB)
- : Flow[immutable.Seq[InfluxDbWriteMessage[Point, C]], immutable.Seq[InfluxDbWriteResult[Point, C]], NotUsed] =
+ : Flow[Seq[InfluxDbWriteMessage[Point, C]], Seq[InfluxDbWriteResult[Point, C]], NotUsed] =
Flow.fromGraph(new impl.InfluxDbFlowStage[C](influxDB))
def typedWithPassThrough[T, C](clazz: Class[T])(
implicit influxDB: InfluxDB)
- : Flow[immutable.Seq[InfluxDbWriteMessage[T, C]], immutable.Seq[InfluxDbWriteResult[T, C]], NotUsed] =
+ : Flow[Seq[InfluxDbWriteMessage[T, C]], Seq[InfluxDbWriteResult[T, C]], NotUsed] =
Flow.fromGraph(new impl.InfluxDbMapperFlowStage[T, C](clazz, influxDB))
}
diff --git a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/scaladsl/InfluxDbSink.scala b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/scaladsl/InfluxDbSink.scala
index aa4b02af9..f13185221 100644
--- a/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/scaladsl/InfluxDbSink.scala
+++ b/influxdb/src/main/scala/org/apache/pekko/stream/connectors/influxdb/scaladsl/InfluxDbSink.scala
@@ -22,7 +22,6 @@ import org.influxdb.InfluxDB
import org.influxdb.dto.Point
import scala.concurrent.Future
-import scala.collection.immutable
/**
* API may change.
@@ -30,12 +29,12 @@ import scala.collection.immutable
@ApiMayChange
object InfluxDbSink {
- def create()(implicit influxDB: InfluxDB): Sink[immutable.Seq[InfluxDbWriteMessage[Point, NotUsed]], Future[Done]] =
+ def create()(implicit influxDB: InfluxDB): Sink[Seq[InfluxDbWriteMessage[Point, NotUsed]], Future[Done]] =
InfluxDbFlow.create().toMat(Sink.ignore)(Keep.right)
def typed[T](
clazz: Class[T])(
- implicit influxDB: InfluxDB): Sink[immutable.Seq[InfluxDbWriteMessage[T, NotUsed]], Future[Done]] =
+ implicit influxDB: InfluxDB): Sink[Seq[InfluxDbWriteMessage[T, NotUsed]], Future[Done]] =
InfluxDbFlow
.typed(clazz)
.toMat(Sink.ignore)(Keep.right)
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 5f7cd974d..c065fb1a5 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
@@ -83,7 +83,7 @@ private[ironmq] final class IronMqClient(settings: IronMqSettings)(implicit acto
* name prefix.
*/
def listQueues(prefix: Option[String] = None, from: Option[String] = None, noOfQueues: Int = 50)(
- implicit ec: ExecutionContext): Future[scala.collection.immutable.Seq[String]] = {
+ implicit ec: ExecutionContext): Future[Seq[String]] = {
def parseQueues(json: Json) = {
@@ -99,7 +99,7 @@ private[ironmq] final class IronMqClient(settings: IronMqSettings)(implicit acto
xs.map(extractName)
}
}
- .as[scala.collection.immutable.Seq[String]]
+ .as[Seq[String]]
}
val query = List(prefix.map("prefix" -> _), from.map("previous" -> _))
diff --git a/ironmq/src/test/scala/docs/scaladsl/IronMqDocsSpec.scala b/ironmq/src/test/scala/docs/scaladsl/IronMqDocsSpec.scala
index 0c0aaae73..8605534ee 100644
--- a/ironmq/src/test/scala/docs/scaladsl/IronMqDocsSpec.scala
+++ b/ironmq/src/test/scala/docs/scaladsl/IronMqDocsSpec.scala
@@ -25,7 +25,6 @@ import pekko.{ Done, NotUsed }
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.BeforeAndAfterAll
-import scala.collection.immutable
import scala.concurrent.Future
import scala.concurrent.duration._
import org.scalatest.matchers.should.Matchers
@@ -61,7 +60,7 @@ class IronMqDocsSpec
val source: Source[Message, NotUsed] =
IronMqConsumer.atMostOnceSource(queueName, ironMqSettings)
- val receivedMessages: Future[immutable.Seq[Message]] = source
+ val receivedMessages: Future[Seq[Message]] = source
.take(100)
.runWith(Sink.seq)
// #atMostOnce
@@ -89,7 +88,7 @@ class IronMqDocsSpec
val businessLogic: Flow[CommittableMessage, CommittableMessage, NotUsed] =
Flow[CommittableMessage] // do something useful with the received messages
- val receivedMessages: Future[immutable.Seq[Message]] = source
+ val receivedMessages: Future[Seq[Message]] = source
.take(100)
.via(businessLogic)
.mapAsync(1)(m => m.commit().map(_ => m.message))
@@ -108,15 +107,15 @@ class IronMqDocsSpec
// #flow
import org.apache.pekko.stream.connectors.ironmq.{ Message, PushMessage }
- val messages: immutable.Seq[String] = (1 to messageCount).map(i => s"test-$i")
- val producedIds: Future[immutable.Seq[Message.Id]] = Source(messages)
+ val messages: Seq[String] = (1 to messageCount).map(i => s"test-$i")
+ val producedIds: Future[Seq[Message.Id]] = Source(messages)
.map(PushMessage(_))
.via(IronMqProducer.flow(queueName, ironMqSettings))
.runWith(Sink.seq)
// #flow
producedIds.futureValue.size shouldBe messageCount
- val receivedMessages: Future[immutable.Seq[Message]] = IronMqConsumer
+ val receivedMessages: Future[Seq[Message]] = IronMqConsumer
.atMostOnceSource(queueName, ironMqSettings)
.take(messages.size)
.runWith(Sink.seq)
@@ -129,7 +128,7 @@ class IronMqDocsSpec
val targetQueue = givenQueue().futureValue
val messageCount = 10
- val messages: immutable.Seq[String] = (1 to messageCount).map(i => s"test-$i")
+ val messages: Seq[String] = (1 to messageCount).map(i => s"test-$i")
val produced = Source(messages)
.map(PushMessage(_))
.runWith(IronMqProducer.sink(sourceQueue, ironMqSettings))
@@ -143,7 +142,7 @@ class IronMqDocsSpec
val pushAndCommit: Flow[(PushMessage, Committable), Message.Id, NotUsed] =
IronMqProducer.atLeastOnceFlow(targetQueue, ironMqSettings)
- val producedIds: Future[immutable.Seq[Message.Id]] = IronMqConsumer
+ val producedIds: Future[Seq[Message.Id]] = IronMqConsumer
.atLeastOnceSource(sourceQueue, ironMqSettings)
.take(messages.size)
.map { committableMessage =>
@@ -154,7 +153,7 @@ class IronMqDocsSpec
// #atLeastOnceFlow
producedIds.futureValue.size shouldBe messageCount
- val receivedMessages: Future[immutable.Seq[Message]] = IronMqConsumer
+ val receivedMessages: Future[Seq[Message]] = IronMqConsumer
.atMostOnceSource(targetQueue, ironMqSettings)
.take(messages.size)
.runWith(Sink.seq)
@@ -169,14 +168,14 @@ class IronMqDocsSpec
// #sink
import org.apache.pekko.stream.connectors.ironmq.{ Message, PushMessage }
- val messages: immutable.Seq[String] = (1 to messageCount).map(i => s"test-$i")
+ val messages: Seq[String] = (1 to messageCount).map(i => s"test-$i")
val producedIds: Future[Done] = Source(messages)
.map(PushMessage(_))
.runWith(IronMqProducer.sink(queueName, ironMqSettings))
// #sink
producedIds.futureValue shouldBe Done
- val receivedMessages: Future[immutable.Seq[Message]] = IronMqConsumer
+ val receivedMessages: Future[Seq[Message]] = IronMqConsumer
.atMostOnceSource(queueName, ironMqSettings)
.take(messages.size)
.runWith(Sink.seq)
diff --git a/jakartams/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala b/jakartams/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala
index 257e432d1..f426fdb69 100644
--- a/jakartams/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala
+++ b/jakartams/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala
@@ -27,7 +27,7 @@ import org.scalatest.time.Span.convertSpanToDuration
import java.util.concurrent.{ LinkedBlockingQueue, TimeUnit }
import scala.annotation.tailrec
-import scala.collection.{ immutable, mutable }
+import scala.collection.mutable
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.util.{ Failure, Success }
@@ -78,7 +78,7 @@ class JmsBufferedAckConnectorsSpec extends JmsSharedServerSpec {
.withSessionCount(5)
.withQueue(queueName))
- val result: Future[immutable.Seq[jakarta.jms.Message]] =
+ val result: Future[Seq[jakarta.jms.Message]] =
jmsSource
.take(msgsIn.size)
.map { ackEnvelope =>
diff --git a/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala b/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala
index 2bb356491..683833aad 100644
--- a/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala
+++ b/jakartams/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala
@@ -32,7 +32,7 @@ import java.util.UUID
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.{ CountDownLatch, LinkedBlockingQueue, ThreadLocalRandom, TimeUnit }
import scala.annotation.tailrec
-import scala.collection.{ immutable, mutable }
+import scala.collection.mutable
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.util.{ Failure, Success }
@@ -69,7 +69,7 @@ class JmsConnectorsSpec extends JmsSpec {
val jmsSource: Source[String, JmsConsumerControl] = JmsConsumer.textSource(
JmsConsumerSettings(system, connectionFactory).withQueue("test"))
- val result: Future[immutable.Seq[String]] = jmsSource.take(in.size).runWith(Sink.seq)
+ val result: Future[Seq[String]] = jmsSource.take(in.size).runWith(Sink.seq)
// #text-source
streamCompletion.futureValue shouldEqual Done
@@ -159,7 +159,7 @@ class JmsConnectorsSpec extends JmsSpec {
val jmsSource: Source[Map[String, Any], JmsConsumerControl] = JmsConsumer.mapSource(
JmsConsumerSettings(system, connectionFactory).withQueue("test"))
- val result: Future[immutable.Seq[Map[String, Any]]] =
+ val result: Future[Seq[Map[String, Any]]] =
jmsSource
.take(1)
.runWith(Sink.seq)
@@ -202,7 +202,7 @@ class JmsConnectorsSpec extends JmsSpec {
val jmsSource: Source[jakarta.jms.Message, JmsConsumerControl] = JmsConsumer(
JmsConsumerSettings(system, connectionFactory).withQueue("numbers"))
- val (control, result): (JmsConsumerControl, Future[immutable.Seq[String]]) =
+ val (control, result): (JmsConsumerControl, Future[Seq[String]]) =
jmsSource
.take(msgsIn.size)
.map {
@@ -226,7 +226,7 @@ class JmsConnectorsSpec extends JmsSpec {
JmsProducerSettings(producerConfig, connectionFactory).withQueue("numbers"))
val finished: Future[Done] =
- Source(immutable.Seq("Message A", "Message B"))
+ Source(Seq("Message A", "Message B"))
.map(JmsTextMessage(_))
.runWith(jmsSink)
// #create-jms-sink
@@ -238,7 +238,7 @@ class JmsConnectorsSpec extends JmsSpec {
finished.futureValue shouldBe Done
- result.futureValue.zip(immutable.Seq("Message A", "Message B")).foreach {
+ result.futureValue.zip(Seq("Message A", "Message B")).foreach {
case (out, in) =>
out.asInstanceOf[TextMessage].getText shouldEqual in
}
@@ -292,7 +292,7 @@ class JmsConnectorsSpec extends JmsSpec {
JmsProducerSettings(producerConfig, connectionFactory)
.withDestination(CustomDestination("custom", createQueue("custom"))))
- val msgsIn: immutable.Seq[JmsTextMessage] = (1 to 10).toList.map { n =>
+ val msgsIn: Seq[JmsTextMessage] = (1 to 10).toList.map { n =>
JmsTextMessage(n.toString)
}
@@ -305,7 +305,7 @@ class JmsConnectorsSpec extends JmsSpec {
.withDestination(CustomDestination("custom", createQueue("custom"))))
// #custom-destination
- val result: Future[immutable.Seq[jakarta.jms.Message]] = jmsSource.take(msgsIn.size).runWith(Sink.seq)
+ val result: Future[Seq[jakarta.jms.Message]] = jmsSource.take(msgsIn.size).runWith(Sink.seq)
// The sent message and the receiving one should have the same properties
result.futureValue.zip(msgsIn).foreach {
@@ -752,7 +752,7 @@ class JmsConnectorsSpec extends JmsSpec {
JmsBrowseSettings(system, connectionFactory)
.withQueue("test"))
- val result: Future[immutable.Seq[jakarta.jms.Message]] =
+ val result: Future[Seq[jakarta.jms.Message]] =
browseSource.runWith(Sink.seq)
// #browse-source
@@ -778,7 +778,7 @@ class JmsConnectorsSpec extends JmsSpec {
JmsProducerSettings(system, connectionFactory)
.withQueue("test"))
- val input: immutable.Seq[JmsTextMessage] = (1 to 100).map(i => JmsTextMessage(i.toString))
+ val input: Seq[JmsTextMessage] = (1 to 100).map(i => JmsTextMessage(i.toString))
val result: Future[Seq[JmsMessage]] = Source(input)
.via(flow)
@@ -1085,7 +1085,7 @@ class JmsConnectorsSpec extends JmsSpec {
JmsProducerSettings(system, connectionFactory).withQueue("test"))
val data = List("a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k")
- val in: immutable.Seq[JmsTextMessagePassThrough[String]] =
+ val in: Seq[JmsTextMessagePassThrough[String]] =
data.map(t => JmsTextMessage(t).withPassThrough(t))
val result = Source(in)
diff --git a/jakartams/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala b/jakartams/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala
index b9de7ce99..b83047b36 100644
--- a/jakartams/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala
+++ b/jakartams/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala
@@ -23,7 +23,6 @@ import pekko.stream.connectors.jakartams.scaladsl.{ JmsConsumer, JmsConsumerCont
import pekko.stream.scaladsl.{ Sink, Source }
import jakarta.jms.{ Session, TextMessage }
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.concurrent.Future
@@ -108,7 +107,7 @@ class JmsIbmmqConnectorsSpec extends JmsSpec {
// #ibmmq-custom-destination
- val result: Future[immutable.Seq[String]] = jmsSource.take(in.size).runWith(Sink.seq)
+ val result: Future[Seq[String]] = jmsSource.take(in.size).runWith(Sink.seq)
streamCompletion.futureValue shouldEqual Done
result.futureValue shouldEqual in
@@ -152,7 +151,7 @@ class JmsIbmmqConnectorsSpec extends JmsSpec {
val in = (0 to 25).map(i => ('a' + i).asInstanceOf[Char].toString)
- val result: Future[immutable.Seq[String]] = jmsTopicSource.take(in.size).runWith(Sink.seq)
+ val result: Future[Seq[String]] = jmsTopicSource.take(in.size).runWith(Sink.seq)
Thread.sleep(500)
diff --git a/jakartams/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala b/jakartams/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala
index 824a84c0f..963a29156 100644
--- a/jakartams/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala
+++ b/jakartams/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala
@@ -25,7 +25,7 @@ import org.slf4j.LoggerFactory
import java.util.concurrent.{ ConcurrentHashMap, LinkedBlockingQueue, TimeUnit }
import scala.annotation.tailrec
-import scala.collection.{ immutable, mutable }
+import scala.collection.mutable
import scala.concurrent.duration._
import scala.concurrent.{ Await, Future, Promise }
import scala.util.{ Failure, Success }
@@ -79,7 +79,7 @@ class JmsTxConnectorsSpec extends JmsSharedServerSpec {
.withAckTimeout(1.second)
.withQueue(queueName))
- val result: Future[immutable.Seq[jakarta.jms.Message]] =
+ val result: Future[Seq[jakarta.jms.Message]] =
jmsSource
.take(msgsIn.size)
.map { txEnvelope =>
diff --git a/jms/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala b/jms/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala
index bfc78430e..a7195235f 100644
--- a/jms/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala
+++ b/jms/src/test/scala/docs/scaladsl/JmsBufferedAckConnectorsSpec.scala
@@ -28,7 +28,7 @@ import org.scalatest.Inspectors._
import org.scalatest.time.Span.convertSpanToDuration
import scala.annotation.tailrec
-import scala.collection.{ immutable, mutable }
+import scala.collection.mutable
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.util.{ Failure, Success }
@@ -79,7 +79,7 @@ class JmsBufferedAckConnectorsSpec extends JmsSharedServerSpec {
.withSessionCount(5)
.withQueue(queueName))
- val result: Future[immutable.Seq[javax.jms.Message]] =
+ val result: Future[Seq[javax.jms.Message]] =
jmsSource
.take(msgsIn.size)
.map { ackEnvelope =>
diff --git a/jms/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala b/jms/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala
index 428308c51..5e920c1e3 100644
--- a/jms/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala
+++ b/jms/src/test/scala/docs/scaladsl/JmsConnectorsSpec.scala
@@ -34,7 +34,6 @@ import org.mockito.invocation.InvocationOnMock
import org.mockito.stubbing.Answer
import scala.annotation.tailrec
-import scala.collection.immutable
import scala.collection.mutable
import scala.concurrent.Future
import scala.concurrent.duration._
@@ -73,7 +72,7 @@ class JmsConnectorsSpec extends JmsSpec {
val jmsSource: Source[String, JmsConsumerControl] = JmsConsumer.textSource(
JmsConsumerSettings(system, connectionFactory).withQueue("test"))
- val result: Future[immutable.Seq[String]] = jmsSource.take(in.size).runWith(Sink.seq)
+ val result: Future[Seq[String]] = jmsSource.take(in.size).runWith(Sink.seq)
// #text-source
streamCompletion.futureValue shouldEqual Done
@@ -163,7 +162,7 @@ class JmsConnectorsSpec extends JmsSpec {
val jmsSource: Source[Map[String, Any], JmsConsumerControl] = JmsConsumer.mapSource(
JmsConsumerSettings(system, connectionFactory).withQueue("test"))
- val result: Future[immutable.Seq[Map[String, Any]]] =
+ val result: Future[Seq[Map[String, Any]]] =
jmsSource
.take(1)
.runWith(Sink.seq)
@@ -206,7 +205,7 @@ class JmsConnectorsSpec extends JmsSpec {
val jmsSource: Source[javax.jms.Message, JmsConsumerControl] = JmsConsumer(
JmsConsumerSettings(system, connectionFactory).withQueue("numbers"))
- val (control, result): (JmsConsumerControl, Future[immutable.Seq[String]]) =
+ val (control, result): (JmsConsumerControl, Future[Seq[String]]) =
jmsSource
.take(msgsIn.size)
.map {
@@ -230,7 +229,7 @@ class JmsConnectorsSpec extends JmsSpec {
JmsProducerSettings(producerConfig, connectionFactory).withQueue("numbers"))
val finished: Future[Done] =
- Source(immutable.Seq("Message A", "Message B"))
+ Source(Seq("Message A", "Message B"))
.map(JmsTextMessage(_))
.runWith(jmsSink)
// #create-jms-sink
@@ -242,7 +241,7 @@ class JmsConnectorsSpec extends JmsSpec {
finished.futureValue shouldBe Done
- result.futureValue.zip(immutable.Seq("Message A", "Message B")).foreach {
+ result.futureValue.zip(Seq("Message A", "Message B")).foreach {
case (out, in) =>
out.asInstanceOf[TextMessage].getText shouldEqual in
}
@@ -296,7 +295,7 @@ class JmsConnectorsSpec extends JmsSpec {
JmsProducerSettings(producerConfig, connectionFactory)
.withDestination(CustomDestination("custom", createQueue("custom"))))
- val msgsIn: immutable.Seq[JmsTextMessage] = (1 to 10).toList.map { n =>
+ val msgsIn: Seq[JmsTextMessage] = (1 to 10).toList.map { n =>
JmsTextMessage(n.toString)
}
@@ -309,7 +308,7 @@ class JmsConnectorsSpec extends JmsSpec {
.withDestination(CustomDestination("custom", createQueue("custom"))))
// #custom-destination
- val result: Future[immutable.Seq[javax.jms.Message]] = jmsSource.take(msgsIn.size).runWith(Sink.seq)
+ val result: Future[Seq[javax.jms.Message]] = jmsSource.take(msgsIn.size).runWith(Sink.seq)
// The sent message and the receiving one should have the same properties
result.futureValue.zip(msgsIn).foreach {
@@ -739,7 +738,7 @@ class JmsConnectorsSpec extends JmsSpec {
JmsBrowseSettings(system, connectionFactory)
.withQueue("test"))
- val result: Future[immutable.Seq[javax.jms.Message]] =
+ val result: Future[Seq[javax.jms.Message]] =
browseSource.runWith(Sink.seq)
// #browse-source
@@ -765,7 +764,7 @@ class JmsConnectorsSpec extends JmsSpec {
JmsProducerSettings(system, connectionFactory)
.withQueue("test"))
- val input: immutable.Seq[JmsTextMessage] = (1 to 100).map(i => JmsTextMessage(i.toString))
+ val input: Seq[JmsTextMessage] = (1 to 100).map(i => JmsTextMessage(i.toString))
val result: Future[Seq[JmsMessage]] = Source(input)
.via(flow)
@@ -1072,7 +1071,7 @@ class JmsConnectorsSpec extends JmsSpec {
JmsProducerSettings(system, connectionFactory).withQueue("test"))
val data = List("a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k")
- val in: immutable.Seq[JmsTextMessagePassThrough[String]] =
+ val in: Seq[JmsTextMessagePassThrough[String]] =
data.map(t => JmsTextMessage(t).withPassThrough(t))
val result = Source(in)
diff --git a/jms/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala b/jms/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala
index eb4048249..597b878fd 100644
--- a/jms/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala
+++ b/jms/src/test/scala/docs/scaladsl/JmsIbmmqConnectorsSpec.scala
@@ -22,7 +22,6 @@ import com.ibm.mq.jms.{ MQQueueConnectionFactory, MQQueueSession, MQTopicConnect
import com.ibm.msg.client.jms.JmsConstants
import com.ibm.msg.client.wmq.common.CommonConstants
import javax.jms.{ Session, TextMessage }
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.concurrent.Future
@@ -107,7 +106,7 @@ class JmsIbmmqConnectorsSpec extends JmsSpec {
// #ibmmq-custom-destination
- val result: Future[immutable.Seq[String]] = jmsSource.take(in.size).runWith(Sink.seq)
+ val result: Future[Seq[String]] = jmsSource.take(in.size).runWith(Sink.seq)
streamCompletion.futureValue shouldEqual Done
result.futureValue shouldEqual in
@@ -151,7 +150,7 @@ class JmsIbmmqConnectorsSpec extends JmsSpec {
val in = (0 to 25).map(i => ('a' + i).asInstanceOf[Char].toString)
- val result: Future[immutable.Seq[String]] = jmsTopicSource.take(in.size).runWith(Sink.seq)
+ val result: Future[Seq[String]] = jmsTopicSource.take(in.size).runWith(Sink.seq)
Thread.sleep(500)
diff --git a/jms/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala b/jms/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala
index fa74b1c06..1c910f044 100644
--- a/jms/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala
+++ b/jms/src/test/scala/docs/scaladsl/JmsTxConnectorsSpec.scala
@@ -26,7 +26,7 @@ import org.scalatest.Inspectors._
import org.slf4j.LoggerFactory
import scala.annotation.tailrec
-import scala.collection.{ immutable, mutable }
+import scala.collection.mutable
import scala.concurrent.{ Await, Future, Promise }
import scala.concurrent.duration._
import scala.util.{ Failure, Success }
@@ -80,7 +80,7 @@ class JmsTxConnectorsSpec extends JmsSharedServerSpec {
.withAckTimeout(1.second)
.withQueue(queueName))
- val result: Future[immutable.Seq[javax.jms.Message]] =
+ val result: Future[Seq[javax.jms.Message]] =
jmsSource
.take(msgsIn.size)
.map { txEnvelope =>
diff --git a/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/scaladsl/KinesisSchedulerSource.scala b/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/scaladsl/KinesisSchedulerSource.scala
index 019769dba..5f53ef0cb 100644
--- a/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/scaladsl/KinesisSchedulerSource.scala
+++ b/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/scaladsl/KinesisSchedulerSource.scala
@@ -27,7 +27,6 @@ import software.amazon.kinesis.coordinator.Scheduler
import software.amazon.kinesis.processor.ShardRecordProcessorFactory
import software.amazon.kinesis.retrieval.KinesisClientRecord
-import scala.collection.immutable
import scala.concurrent.Future
object KinesisSchedulerSource {
@@ -64,7 +63,7 @@ object KinesisSchedulerSource {
.mergeSubstreams
private val checkpointRecordBatch =
- Flow[immutable.Seq[CommittableRecord]]
+ Flow[Seq[CommittableRecord]]
.map(records => {
records.max(CommittableRecord.orderBySequenceNumber).tryToCheckpoint()
records
diff --git a/mongodb/src/test/scala/docs/scaladsl/MongoSourceSpec.scala b/mongodb/src/test/scala/docs/scaladsl/MongoSourceSpec.scala
index 4735f4346..e41edede8 100644
--- a/mongodb/src/test/scala/docs/scaladsl/MongoSourceSpec.scala
+++ b/mongodb/src/test/scala/docs/scaladsl/MongoSourceSpec.scala
@@ -28,7 +28,6 @@ import org.scalatest.wordspec.AnyWordSpec
import org.scalatest.concurrent.ScalaFutures
import scala.annotation.nowarn
-import scala.collection.immutable.Seq
import scala.concurrent._
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
diff --git a/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/BehaviorRunner.scala b/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/BehaviorRunner.scala
index 7d62945d1..357d4ca70 100644
--- a/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/BehaviorRunner.scala
+++ b/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/BehaviorRunner.scala
@@ -17,8 +17,6 @@ import org.apache.pekko
import pekko.actor.typed.scaladsl.{ ActorContext, Behaviors }
import pekko.actor.typed.{ Behavior, Signal }
-import scala.collection.immutable.Seq
-
object BehaviorRunner {
sealed trait Interpretable[T]
diff --git a/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/ClientState.scala b/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/ClientState.scala
index 2f977151f..d56e76b79 100644
--- a/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/ClientState.scala
+++ b/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/ClientState.scala
@@ -23,7 +23,6 @@ import pekko.stream.{ Materializer, OverflowStrategy, QueueOfferResult }
import pekko.stream.scaladsl.{ BroadcastHub, Keep, Source, SourceQueueWithComplete }
import pekko.util.ByteString
-import scala.collection.immutable.Seq
import scala.concurrent.Promise
import scala.concurrent.duration.FiniteDuration
import scala.util.control.NoStackTrace
diff --git a/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/QueueOfferState.scala b/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/QueueOfferState.scala
index 864da11c9..e7c8d290a 100644
--- a/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/QueueOfferState.scala
+++ b/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/QueueOfferState.scala
@@ -17,7 +17,6 @@ import org.apache.pekko
import pekko.actor.typed.Behavior
import pekko.actor.typed.scaladsl.Behaviors
import pekko.stream.QueueOfferResult
-import scala.collection.immutable.Seq
import scala.concurrent.Future
import scala.util.{ Failure, Success, Try }
diff --git a/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/ServerState.scala b/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/ServerState.scala
index fb6a0366f..a34b90eee 100644
--- a/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/ServerState.scala
+++ b/mqtt-streaming/src/main/scala/org/apache/pekko/stream/connectors/mqtt/streaming/impl/ServerState.scala
@@ -24,7 +24,6 @@ import pekko.annotation.InternalApi
import pekko.stream.{ Materializer, OverflowStrategy, QueueOfferResult }
import pekko.stream.scaladsl.{ BroadcastHub, Keep, Source, SourceQueueWithComplete }
import pekko.util.ByteString
-import scala.collection.immutable.Seq
import scala.concurrent.Promise
import scala.concurrent.duration.FiniteDuration
import scala.util.control.NoStackTrace
diff --git a/mqtt/src/main/scala/org/apache/pekko/stream/connectors/mqtt/settings.scala b/mqtt/src/main/scala/org/apache/pekko/stream/connectors/mqtt/settings.scala
index 2b62a6717..4280b0caa 100644
--- a/mqtt/src/main/scala/org/apache/pekko/stream/connectors/mqtt/settings.scala
+++ b/mqtt/src/main/scala/org/apache/pekko/stream/connectors/mqtt/settings.scala
@@ -16,7 +16,6 @@ package org.apache.pekko.stream.connectors.mqtt
import org.apache.pekko
import org.eclipse.paho.client.mqttv3.{ MqttClientPersistence, MqttConnectOptions }
-import scala.collection.immutable
import scala.collection.immutable.Map
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
@@ -166,7 +165,7 @@ final class MqttConnectionSettings private (val broker: String,
val disconnectTimeout: FiniteDuration,
val maxInFlight: Int,
val mqttVersion: Int,
- val serverUris: immutable.Seq[String],
+ val serverUris: Seq[String],
val sslHostnameVerifier: Option[javax.net.ssl.HostnameVerifier],
val sslProperties: Map[String, String],
val offlinePersistenceSettings: Option[MqttOfflinePersistenceSettings]) {
@@ -223,10 +222,10 @@ final class MqttConnectionSettings private (val broker: String,
def withMaxInFlight(value: Int): MqttConnectionSettings = copy(maxInFlight = value)
def withMqttVersion(value: Int): MqttConnectionSettings = copy(mqttVersion = value)
- def withServerUri(value: String): MqttConnectionSettings = copy(serverUris = immutable.Seq(value))
+ def withServerUri(value: String): MqttConnectionSettings = copy(serverUris = Seq(value))
/** Scala API */
- def withServerUris(values: immutable.Seq[String]): MqttConnectionSettings = copy(serverUris = values)
+ def withServerUris(values: Seq[String]): MqttConnectionSettings = copy(serverUris = values)
/** Java API */
def withServerUris(values: java.util.List[String]): MqttConnectionSettings = copy(serverUris = values.asScala.toList)
@@ -262,7 +261,7 @@ final class MqttConnectionSettings private (val broker: String,
disconnectTimeout: FiniteDuration = disconnectTimeout,
maxInFlight: Int = maxInFlight,
mqttVersion: Int = mqttVersion,
- serverUris: immutable.Seq[String] = serverUris,
+ serverUris: Seq[String] = serverUris,
sslHostnameVerifier: Option[javax.net.ssl.HostnameVerifier] = sslHostnameVerifier,
sslProperties: Map[String, java.lang.String] = sslProperties,
offlinePersistenceSettings: Option[MqttOfflinePersistenceSettings] = offlinePersistenceSettings)
@@ -337,7 +336,7 @@ object MqttConnectionSettings {
disconnectTimeout = 10.seconds,
maxInFlight = MqttConnectOptions.MAX_INFLIGHT_DEFAULT,
mqttVersion = MqttConnectOptions.MQTT_VERSION_3_1_1,
- serverUris = immutable.Seq.empty,
+ serverUris = Seq.empty,
sslHostnameVerifier = None,
sslProperties = Map.empty,
offlinePersistenceSettings = None)
diff --git a/mqtt/src/test/scala/docs/scaladsl/MqttSourceSpec.scala b/mqtt/src/test/scala/docs/scaladsl/MqttSourceSpec.scala
index d1fd588b8..db2ea4a85 100644
--- a/mqtt/src/test/scala/docs/scaladsl/MqttSourceSpec.scala
+++ b/mqtt/src/test/scala/docs/scaladsl/MqttSourceSpec.scala
@@ -26,7 +26,6 @@ import org.eclipse.paho.client.mqttv3.MqttException
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
import org.slf4j.LoggerFactory
-import scala.collection.immutable.Seq
import scala.concurrent.duration._
import scala.concurrent.{ Await, ExecutionContext, Future, Promise }
diff --git a/mqttv5/src/main/scala/org/apache/pekko/stream/connectors/mqttv5/model.scala b/mqttv5/src/main/scala/org/apache/pekko/stream/connectors/mqttv5/model.scala
index c49cad7f9..815e8ac50 100644
--- a/mqttv5/src/main/scala/org/apache/pekko/stream/connectors/mqttv5/model.scala
+++ b/mqttv5/src/main/scala/org/apache/pekko/stream/connectors/mqttv5/model.scala
@@ -15,7 +15,6 @@ package org.apache.pekko.stream.connectors.mqttv5
import org.apache.pekko
-import scala.collection.immutable
import scala.jdk.CollectionConverters._
import scala.jdk.OptionConverters._
@@ -63,7 +62,7 @@ final class MqttMessage private (
def withRetained(value: Boolean): MqttMessage = if (retained == value) this else copy(retained = value)
/** Scala API */
- def withUserProperties(value: immutable.Seq[MqttUserProperty]): MqttMessage =
+ def withUserProperties(value: Seq[MqttUserProperty]): MqttMessage =
copy(userProperties = value)
/** Java API */
diff --git a/mqttv5/src/main/scala/org/apache/pekko/stream/connectors/mqttv5/settings.scala b/mqttv5/src/main/scala/org/apache/pekko/stream/connectors/mqttv5/settings.scala
index 5505f24b4..0bc95142a 100644
--- a/mqttv5/src/main/scala/org/apache/pekko/stream/connectors/mqttv5/settings.scala
+++ b/mqttv5/src/main/scala/org/apache/pekko/stream/connectors/mqttv5/settings.scala
@@ -16,7 +16,6 @@ package org.apache.pekko.stream.connectors.mqttv5
import java.nio.charset.StandardCharsets
import java.util.Properties
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
import scala.jdk.DurationConverters._
@@ -375,7 +374,7 @@ final class MqttConnectionSettings private (
copy(serverUris = Array(value))
/** Scala API */
- def withServerUris(value: immutable.Seq[String]): MqttConnectionSettings =
+ def withServerUris(value: Seq[String]): MqttConnectionSettings =
copy(serverUris = value.toArray)
/** Java API */
diff --git a/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/impl/OrientDbFlowStage.scala b/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/impl/OrientDbFlowStage.scala
index 5f239aeee..5b5e3a301 100644
--- a/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/impl/OrientDbFlowStage.scala
+++ b/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/impl/OrientDbFlowStage.scala
@@ -25,7 +25,6 @@ import com.orientechnologies.orient.core.record.ORecord
import com.orientechnologies.orient.core.record.impl.ODocument
import com.orientechnologies.orient.core.tx.OTransaction
-import scala.collection.immutable
import scala.util.control.NonFatal
/**
@@ -36,12 +35,12 @@ private[orientdb] class OrientDbFlowStage[T, C](
className: String,
settings: OrientDbWriteSettings,
clazz: Option[Class[T]])
- extends GraphStage[FlowShape[immutable.Seq[OrientDbWriteMessage[T, C]],
- immutable.Seq[OrientDbWriteMessage[T,
+ extends GraphStage[FlowShape[Seq[OrientDbWriteMessage[T, C]],
+ Seq[OrientDbWriteMessage[T,
C]]]] {
- private val in = Inlet[immutable.Seq[OrientDbWriteMessage[T, C]]]("in")
- private val out = Outlet[immutable.Seq[OrientDbWriteMessage[T, C]]]("out")
+ private val in = Inlet[Seq[OrientDbWriteMessage[T, C]]]("in")
+ private val out = Outlet[Seq[OrientDbWriteMessage[T, C]]]("out")
override val shape = FlowShape(in, out)
override def initialAttributes: Attributes =
// see https://orientdb.com/docs/last/Java-Multi-Threading.html
@@ -68,7 +67,7 @@ private[orientdb] class OrientDbFlowStage[T, C](
client.close()
}
- protected def write(messages: immutable.Seq[OrientDbWriteMessage[T, C]]): Unit
+ protected def write(messages: Seq[OrientDbWriteMessage[T, C]]): Unit
setHandlers(in, out, this)
@@ -102,7 +101,7 @@ private[orientdb] class OrientDbFlowStage[T, C](
}
}
- protected def write(messages: immutable.Seq[OrientDbWriteMessage[T, C]]): Unit =
+ protected def write(messages: Seq[OrientDbWriteMessage[T, C]]): Unit =
messages.foreach {
case OrientDbWriteMessage(oDocument: ODocument, _) =>
val document = new ODocument()
@@ -129,7 +128,7 @@ private[orientdb] class OrientDbFlowStage[T, C](
oObjectClient.getEntityManager.registerEntityClass(clazz)
}
- protected def write(messages: immutable.Seq[OrientDbWriteMessage[T, C]]): Unit =
+ protected def write(messages: Seq[OrientDbWriteMessage[T, C]]): Unit =
messages.foreach {
case OrientDbWriteMessage(typeRecord, _) =>
oObjectClient.save[Object](typeRecord)
diff --git a/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/scaladsl/OrientDbFlow.scala b/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/scaladsl/OrientDbFlow.scala
index 881f79af4..31984b367 100644
--- a/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/scaladsl/OrientDbFlow.scala
+++ b/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/scaladsl/OrientDbFlow.scala
@@ -19,7 +19,6 @@ import pekko.stream.connectors.orientdb._
import pekko.stream.connectors.orientdb.impl.OrientDbFlowStage
import pekko.stream.scaladsl.Flow
import com.orientechnologies.orient.core.record.impl.ODocument
-import scala.collection.immutable
/**
* Scala API.
@@ -31,8 +30,8 @@ object OrientDbFlow {
*/
def create(
className: String,
- settings: OrientDbWriteSettings): Flow[immutable.Seq[OrientDbWriteMessage[ODocument, NotUsed]],
- immutable.Seq[OrientDbWriteMessage[ODocument, NotUsed]], NotUsed] =
+ settings: OrientDbWriteSettings): Flow[Seq[OrientDbWriteMessage[ODocument, NotUsed]],
+ Seq[OrientDbWriteMessage[ODocument, NotUsed]], NotUsed] =
Flow
.fromGraph(
new OrientDbFlowStage[ODocument, NotUsed](
@@ -46,8 +45,8 @@ object OrientDbFlow {
*/
def createWithPassThrough[C](
className: String,
- settings: OrientDbWriteSettings): Flow[immutable.Seq[OrientDbWriteMessage[ODocument, C]],
- immutable.Seq[OrientDbWriteMessage[ODocument, C]], NotUsed] =
+ settings: OrientDbWriteSettings): Flow[Seq[OrientDbWriteMessage[ODocument, C]],
+ Seq[OrientDbWriteMessage[ODocument, C]], NotUsed] =
Flow
.fromGraph(
new OrientDbFlowStage[ODocument, C](
@@ -61,8 +60,8 @@ object OrientDbFlow {
def typed[T](
className: String,
settings: OrientDbWriteSettings,
- clazz: Class[T]): Flow[immutable.Seq[OrientDbWriteMessage[T, NotUsed]],
- immutable.Seq[OrientDbWriteMessage[T,
+ clazz: Class[T]): Flow[Seq[OrientDbWriteMessage[T, NotUsed]],
+ Seq[OrientDbWriteMessage[T,
NotUsed]], NotUsed] =
Flow
.fromGraph(
@@ -79,7 +78,7 @@ object OrientDbFlow {
className: String,
settings: OrientDbWriteSettings,
clazz: Class[T])
- : Flow[immutable.Seq[OrientDbWriteMessage[T, C]], immutable.Seq[OrientDbWriteMessage[T, C]], NotUsed] =
+ : Flow[Seq[OrientDbWriteMessage[T, C]], Seq[OrientDbWriteMessage[T, C]], NotUsed] =
Flow
.fromGraph(
new OrientDbFlowStage[T, C](
diff --git a/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/scaladsl/OrientDbSink.scala b/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/scaladsl/OrientDbSink.scala
index d921149a1..a2d0044b0 100644
--- a/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/scaladsl/OrientDbSink.scala
+++ b/orientdb/src/main/scala/org/apache/pekko/stream/connectors/orientdb/scaladsl/OrientDbSink.scala
@@ -20,7 +20,6 @@ import pekko.stream.scaladsl.{ Keep, Sink }
import com.orientechnologies.orient.core.record.impl.ODocument
import scala.concurrent.Future
-import scala.collection.immutable
/**
* Scala API.
@@ -32,7 +31,7 @@ object OrientDbSink {
*/
def apply(
className: String,
- settings: OrientDbWriteSettings): Sink[immutable.Seq[OrientDbWriteMessage[ODocument, NotUsed]], Future[Done]] =
+ settings: OrientDbWriteSettings): Sink[Seq[OrientDbWriteMessage[ODocument, NotUsed]], Future[Done]] =
OrientDbFlow.create(className, settings).toMat(Sink.ignore)(Keep.right)
/**
@@ -41,7 +40,7 @@ object OrientDbSink {
def typed[T](
className: String,
settings: OrientDbWriteSettings,
- clazz: Class[T]): Sink[immutable.Seq[OrientDbWriteMessage[T, NotUsed]], Future[Done]] =
+ clazz: Class[T]): Sink[Seq[OrientDbWriteMessage[T, NotUsed]], Future[Done]] =
OrientDbFlow
.typed(className, settings, clazz)
.toMat(Sink.ignore)(Keep.right)
diff --git a/orientdb/src/test/scala/docs/scaladsl/OrientDbSpec.scala b/orientdb/src/test/scala/docs/scaladsl/OrientDbSpec.scala
index d3122bc88..e5ad7adaa 100644
--- a/orientdb/src/test/scala/docs/scaladsl/OrientDbSpec.scala
+++ b/orientdb/src/test/scala/docs/scaladsl/OrientDbSpec.scala
@@ -42,7 +42,6 @@ import docs.javadsl.OrientDbTest
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.BeforeAndAfterAll
-import scala.collection.immutable
import scala.concurrent.Future
import scala.concurrent.duration._
import org.scalatest.matchers.should.Matchers
@@ -170,7 +169,7 @@ class OrientDbSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with
f1.futureValue shouldBe Done
// #run-odocument
- val result: Future[immutable.Seq[String]] = OrientDbSource(
+ val result: Future[Seq[String]] = OrientDbSource(
sink4,
OrientDbSourceSettings(oDatabase)).map { (message: OrientDbReadResult[ODocument]) =>
message.oDocument.field[String]("book_title")
diff --git a/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/impl/ReferenceSourceStage.scala b/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/impl/ReferenceSourceStage.scala
index 787750cd3..73cd557e9 100644
--- a/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/impl/ReferenceSourceStage.scala
+++ b/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/impl/ReferenceSourceStage.scala
@@ -22,7 +22,6 @@ import pekko.stream.connectors.reference.{ ReferenceReadResult, SourceSettings }
import pekko.stream.stage.{ GraphStageLogic, GraphStageWithMaterializedValue, OutHandler }
import pekko.util.ByteString
-import scala.collection.immutable
import scala.concurrent.{ Future, Promise }
import scala.util.Success
@@ -50,7 +49,7 @@ import scala.util.Success
new OutHandler {
override def onPull(): Unit = push(
out,
- new ReferenceReadResult(immutable.Seq(ByteString("one")), Success(100)))
+ new ReferenceReadResult(Seq(ByteString("one")), Success(100)))
})
/**
diff --git a/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/model.scala b/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/model.scala
index 61eff50ed..5bee0ab19 100644
--- a/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/model.scala
+++ b/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/model.scala
@@ -19,7 +19,6 @@ import org.apache.pekko
import pekko.annotation.InternalApi
import pekko.util.ByteString
-import scala.collection.immutable
import scala.jdk.CollectionConverters._
import scala.jdk.OptionConverters._
import scala.util.{ Success, Try }
@@ -31,7 +30,7 @@ import scala.util.{ Success, Try }
* [[pekko.stream.connectors.reference.testkit.MessageFactory]].
*/
final class ReferenceReadResult @InternalApi private[reference] (
- val data: immutable.Seq[ByteString] = immutable.Seq.empty,
+ val data: Seq[ByteString] = Seq.empty,
val bytesRead: Try[Int] = Success(0)) {
/**
@@ -72,9 +71,9 @@ final class ReferenceReadResult @InternalApi private[reference] (
* Use "Write" in message data types to signify that the messages is to be written to outside.
*/
final class ReferenceWriteMessage private (
- val data: immutable.Seq[ByteString] = immutable.Seq.empty,
+ val data: Seq[ByteString] = Seq.empty,
val metrics: Map[String, Long] = Map.empty) {
- def withData(data: immutable.Seq[ByteString]): ReferenceWriteMessage =
+ def withData(data: Seq[ByteString]): ReferenceWriteMessage =
copy(data = data)
def withMetrics(metrics: Map[String, Long]): ReferenceWriteMessage =
@@ -117,7 +116,7 @@ final class ReferenceWriteMessage private (
case (key, value) => key -> java.lang.Long.valueOf(value)
}.asJava
- private def copy(data: immutable.Seq[ByteString] = data, metrics: Map[String, Long] = metrics) =
+ private def copy(data: Seq[ByteString] = data, metrics: Map[String, Long] = metrics) =
new ReferenceWriteMessage(data, metrics)
override def toString: String =
diff --git a/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/testkit/MessageFactory.scala b/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/testkit/MessageFactory.scala
index c8c685ccd..7a5965762 100644
--- a/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/testkit/MessageFactory.scala
+++ b/reference/src/main/scala/org/apache/pekko/stream/connectors/reference/testkit/MessageFactory.scala
@@ -18,7 +18,6 @@ import pekko.annotation.ApiMayChange
import pekko.stream.connectors.reference.{ ReferenceReadResult, ReferenceWriteMessage, ReferenceWriteResult }
import pekko.util.ByteString
-import scala.collection.immutable
import scala.jdk.CollectionConverters._
import scala.util.{ Failure, Success, Try }
@@ -26,7 +25,7 @@ import scala.util.{ Failure, Success, Try }
object MessageFactory {
@ApiMayChange
- def createReadResult(data: immutable.Seq[ByteString], bytesRead: Try[Int]): ReferenceReadResult =
+ def createReadResult(data: Seq[ByteString], bytesRead: Try[Int]): ReferenceReadResult =
new ReferenceReadResult(data, bytesRead)
/**
diff --git a/reference/src/test/scala/docs/scaladsl/ReferenceSpec.scala b/reference/src/test/scala/docs/scaladsl/ReferenceSpec.scala
index f8a2c19c2..eea515f57 100644
--- a/reference/src/test/scala/docs/scaladsl/ReferenceSpec.scala
+++ b/reference/src/test/scala/docs/scaladsl/ReferenceSpec.scala
@@ -27,7 +27,6 @@ import org.scalatest.concurrent.ScalaFutures
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec
-import scala.collection.immutable
import scala.concurrent.Future
/**
@@ -88,17 +87,17 @@ class ReferenceSpec extends AnyWordSpec with BeforeAndAfterAll with ScalaFutures
"run flow" in {
val flow = Reference.flow()
val source = Source(
- immutable.Seq(
+ Seq(
ReferenceWriteMessage()
- .withData(immutable.Seq(ByteString("one")))
+ .withData(Seq(ByteString("one")))
.withMetrics(Map("rps" -> 20L, "rpm" -> 30L)),
ReferenceWriteMessage().withData(
- immutable.Seq(
+ Seq(
ByteString("two"),
ByteString("three"),
ByteString("four"))),
ReferenceWriteMessage().withData(
- immutable.Seq(
+ Seq(
ByteString("five"),
ByteString("six"),
ByteString("seven")))))
@@ -119,7 +118,7 @@ class ReferenceSpec extends AnyWordSpec with BeforeAndAfterAll with ScalaFutures
"resolve resource from application config" in {
val result = Source
- .single(ReferenceWriteMessage().withData(immutable.Seq(ByteString("one"))))
+ .single(ReferenceWriteMessage().withData(Seq(ByteString("one"))))
.via(Reference.flowWithResource())
.runWith(Sink.seq)
@@ -130,7 +129,7 @@ class ReferenceSpec extends AnyWordSpec with BeforeAndAfterAll with ScalaFutures
val resource = Resource(ResourceSettings("attributes msg"))
val result = Source
- .single(ReferenceWriteMessage().withData(immutable.Seq(ByteString("one"))))
+ .single(ReferenceWriteMessage().withData(Seq(ByteString("one"))))
.via(Reference.flowWithResource().withAttributes(ReferenceAttributes.resource(resource)))
.runWith(Sink.seq)
diff --git a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/S3Headers.scala b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/S3Headers.scala
index f29bb36ff..143ed724a 100644
--- a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/S3Headers.scala
+++ b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/S3Headers.scala
@@ -24,8 +24,6 @@ import pekko.stream.connectors.s3.impl.S3Request
import scala.jdk.CollectionConverters._
-import scala.collection.immutable.Seq
-
final class MetaHeaders private (val metaHeaders: Map[String, String]) {
@InternalApi private[s3] def headers: Seq[HttpHeader] =
diff --git a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/headers/ServerSideEncryption.scala b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/headers/ServerSideEncryption.scala
index 8d2dd5e6c..ae7be0823 100644
--- a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/headers/ServerSideEncryption.scala
+++ b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/headers/ServerSideEncryption.scala
@@ -23,15 +23,13 @@ import pekko.stream.connectors.s3.impl._
import software.amazon.awssdk.utils.BinaryUtils
import software.amazon.awssdk.utils.Md5Utils
-import scala.collection.immutable
-
/**
* Documentation: http://docs.aws.amazon.com/AmazonS3/latest/dev/serv-side-encryption.html
*/
sealed abstract class ServerSideEncryption {
- @InternalApi private[s3] def headers: immutable.Seq[HttpHeader]
+ @InternalApi private[s3] def headers: Seq[HttpHeader]
- @InternalApi private[s3] def headersFor(request: S3Request): immutable.Seq[HttpHeader]
+ @InternalApi private[s3] def headersFor(request: S3Request): Seq[HttpHeader]
}
object ServerSideEncryption {
@@ -46,10 +44,10 @@ object ServerSideEncryption {
}
final class AES256 private[headers] () extends ServerSideEncryption {
- @InternalApi private[s3] override def headers: immutable.Seq[HttpHeader] =
+ @InternalApi private[s3] override def headers: Seq[HttpHeader] =
RawHeader("x-amz-server-side-encryption", "AES256") :: Nil
- @InternalApi private[s3] override def headersFor(request: S3Request): immutable.Seq[HttpHeader] = request match {
+ @InternalApi private[s3] override def headersFor(request: S3Request): Seq[HttpHeader] = request match {
case PutObject | InitiateMultipartUpload => headers
case _ => Nil
}
@@ -75,7 +73,7 @@ final class AES256 private[headers] () extends ServerSideEncryption {
*/
final class KMS private[headers] (val keyId: String, val context: Option[String]) extends ServerSideEncryption {
- @InternalApi private[s3] override def headers: immutable.Seq[HttpHeader] = {
+ @InternalApi private[s3] override def headers: Seq[HttpHeader] = {
val baseHeaders = RawHeader("x-amz-server-side-encryption", "aws:kms") ::
RawHeader("x-amz-server-side-encryption-aws-kms-key-id", keyId) ::
Nil
@@ -84,7 +82,7 @@ final class KMS private[headers] (val keyId: String, val context: Option[String]
headers
}
- @InternalApi private[s3] override def headersFor(request: S3Request): immutable.Seq[HttpHeader] = request match {
+ @InternalApi private[s3] override def headersFor(request: S3Request): Seq[HttpHeader] = request match {
case PutObject | InitiateMultipartUpload => headers
case _ => Nil
}
@@ -124,7 +122,7 @@ final class KMS private[headers] (val keyId: String, val context: Option[String]
final class CustomerKeys private[headers] (val key: String, val md5: Option[String] = None)
extends ServerSideEncryption {
- @InternalApi private[s3] override def headers: immutable.Seq[HttpHeader] =
+ @InternalApi private[s3] override def headers: Seq[HttpHeader] =
RawHeader("x-amz-server-side-encryption-customer-algorithm", "AES256") ::
RawHeader("x-amz-server-side-encryption-customer-key", key) ::
RawHeader("x-amz-server-side-encryption-customer-key-MD5",
@@ -134,7 +132,7 @@ final class CustomerKeys private[headers] (val key: String, val md5: Option[Stri
md5
}) :: Nil
- @InternalApi private[s3] override def headersFor(request: S3Request): immutable.Seq[HttpHeader] = request match {
+ @InternalApi private[s3] override def headersFor(request: S3Request): Seq[HttpHeader] = request match {
case GetObject | HeadObject | PutObject | InitiateMultipartUpload | UploadPart =>
headers
case CopyPart =>
diff --git a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/HttpRequests.scala b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/HttpRequests.scala
index d5bc46340..334df85c1 100644
--- a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/HttpRequests.scala
+++ b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/HttpRequests.scala
@@ -36,7 +36,6 @@ import pekko.stream.scaladsl.Source
import pekko.util.ByteString
import software.amazon.awssdk.regions.Region
-import scala.collection.immutable.Seq
import scala.concurrent.{ ExecutionContext, Future }
import scala.xml.NodeSeq
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 1020aa0c0..ce974da44 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
@@ -55,13 +55,13 @@ import scala.util.{ Failure, Success, Try }
versionId: Option[String] = None)
/** Internal Api */
-@InternalApi private[impl] final case class ListBucketsResult(buckets: immutable.Seq[ListBucketsResultContents])
+@InternalApi private[impl] final case class ListBucketsResult(buckets: Seq[ListBucketsResultContents])
/** Internal Api */
@InternalApi private[impl] final case class ListBucketResult(isTruncated: Boolean,
continuationToken: Option[String],
- contents: immutable.Seq[ListBucketResultContents],
- commonPrefixes: immutable.Seq[ListBucketResultCommonPrefixes])
+ contents: Seq[ListBucketResultContents],
+ commonPrefixes: Seq[ListBucketResultCommonPrefixes])
/** Internal Api */
@InternalApi private[impl] final case class ListMultipartUploadContinuationToken(nextKeyMarker: Option[String],
@@ -77,8 +77,8 @@ import scala.util.{ Failure, Success, Try }
delimiter: Option[String],
maxUploads: Int,
isTruncated: Boolean,
- uploads: immutable.Seq[ListMultipartUploadResultUploads],
- commonPrefixes: immutable.Seq[CommonPrefixes]) {
+ uploads: Seq[ListMultipartUploadResultUploads],
+ commonPrefixes: Seq[CommonPrefixes]) {
/**
* The continuation token for listing MultipartUpload is a union of both the nextKeyMarker
@@ -109,9 +109,9 @@ import scala.util.{ Failure, Success, Try }
delimiter: Option[String],
maxKeys: Int,
isTruncated: Boolean,
- versions: immutable.Seq[ListObjectVersionsResultVersions],
- commonPrefixes: immutable.Seq[CommonPrefixes],
- deleteMarkers: immutable.Seq[DeleteMarkers]) {
+ versions: Seq[ListObjectVersionsResultVersions],
+ commonPrefixes: Seq[CommonPrefixes],
+ deleteMarkers: Seq[DeleteMarkers]) {
/**
* The continuation token for listing ObjectVersions is a union of both the nextKeyMarker
@@ -133,7 +133,7 @@ import scala.util.{ Failure, Success, Try }
nextPartNumberMarker: Option[Int],
maxParts: Int,
isTruncated: Boolean,
- parts: immutable.Seq[ListPartsResultParts],
+ parts: Seq[ListPartsResultParts],
initiator: Option[AWSIdentity],
owner: Option[AWSIdentity],
storageClass: String) {
@@ -301,7 +301,7 @@ import scala.util.{ Failure, Success, Try }
delimiter: String,
prefix: Option[String] = None,
s3Headers: S3Headers)
- : Source[(immutable.Seq[ListBucketResultContents], immutable.Seq[ListBucketResultCommonPrefixes]), NotUsed] = {
+ : Source[(Seq[ListBucketResultContents], Seq[ListBucketResultCommonPrefixes]), NotUsed] = {
def listBucketCallContentsAndCommonPrefixes(token: Option[String])(implicit mat: Materializer, attr: Attributes) =
listBucketCall(bucket,
@@ -317,8 +317,8 @@ import scala.util.{ Failure, Success, Try }
implicit val attributes: Attributes = attr
Source
.unfoldAsync[ListBucketState,
- (immutable.Seq[ListBucketResultContents],
- immutable.Seq[
+ (Seq[ListBucketResultContents],
+ Seq[
ListBucketResultCommonPrefixes])](
Starting()) {
case Finished() => Future.successful(None)
@@ -377,7 +377,7 @@ import scala.util.{ Failure, Success, Try }
delimiter: String,
prefix: Option[String] = None,
s3Headers: S3Headers)
- : Source[(immutable.Seq[ListMultipartUploadResultUploads], immutable.Seq[CommonPrefixes]), NotUsed] = {
+ : Source[(Seq[ListMultipartUploadResultUploads], Seq[CommonPrefixes]), NotUsed] = {
def listMultipartUploadCallContentsAndCommonPrefixes(
token: Option[ListMultipartUploadContinuationToken])(implicit mat: Materializer, attr: Attributes) =
@@ -394,8 +394,8 @@ import scala.util.{ Failure, Success, Try }
implicit val attributes: Attributes = attr
Source
.unfoldAsync[ListMultipartUploadState,
- (immutable.Seq[ListMultipartUploadResultUploads],
- immutable.Seq[
+ (Seq[ListMultipartUploadResultUploads],
+ Seq[
CommonPrefixes])](
Starting()) {
case Finished() => Future.successful(None)
@@ -503,7 +503,7 @@ import scala.util.{ Failure, Success, Try }
delimiter: String,
prefix: Option[String],
s3Headers: S3Headers): Source[
- (immutable.Seq[ListObjectVersionsResultVersions], immutable.Seq[DeleteMarkers], immutable.Seq[CommonPrefixes]),
+ (Seq[ListObjectVersionsResultVersions], Seq[DeleteMarkers], Seq[CommonPrefixes]),
NotUsed] = {
def listObjectVersionsCallVersionsAndDeleteMarkersAndCommonPrefixes(
@@ -525,10 +525,10 @@ import scala.util.{ Failure, Success, Try }
implicit val attributes: Attributes = attr
Source
.unfoldAsync[ListObjectVersionsState,
- (immutable.Seq[ListObjectVersionsResultVersions],
- immutable.Seq[
+ (Seq[ListObjectVersionsResultVersions],
+ Seq[
DeleteMarkers],
- immutable.Seq[
+ Seq[
CommonPrefixes])](
Starting()) {
case Finished() => Future.successful(None)
@@ -543,7 +543,7 @@ import scala.util.{ Failure, Success, Try }
bucket: String,
prefix: Option[String],
s3Headers: S3Headers)
- : Source[(immutable.Seq[ListObjectVersionsResultVersions], immutable.Seq[DeleteMarkers]), NotUsed] = {
+ : Source[(Seq[ListObjectVersionsResultVersions], Seq[DeleteMarkers]), NotUsed] = {
def listObjectVersionsCallOnlyVersions(
token: Option[ListObjectVersionContinuationToken])(implicit mat: Materializer, attr: Attributes) =
@@ -561,8 +561,8 @@ import scala.util.{ Failure, Success, Try }
implicit val attributes: Attributes = attr
Source
.unfoldAsync[ListObjectVersionsState,
- (immutable.Seq[ListObjectVersionsResultVersions],
- immutable.Seq[
+ (Seq[ListObjectVersionsResultVersions],
+ Seq[
DeleteMarkers])](
Starting()) {
case Finished() => Future.successful(None)
@@ -693,7 +693,7 @@ import scala.util.{ Failure, Success, Try }
method: HttpMethod = HttpMethods.GET,
rangeOption: Option[ByteRange] = None,
versionId: Option[String] = None,
- s3Headers: immutable.Seq[HttpHeader] = Nil): Source[HttpResponse, NotUsed] =
+ s3Headers: Seq[HttpHeader] = Nil): Source[HttpResponse, NotUsed] =
Source
.fromMaterializer { (mat, attr) =>
issueRequest(s3Location, method, rangeOption, versionId, s3Headers)(mat, attr)
@@ -705,7 +705,7 @@ import scala.util.{ Failure, Success, Try }
method: HttpMethod = HttpMethods.GET,
rangeOption: Option[ByteRange] = None,
versionId: Option[String],
- s3Headers: immutable.Seq[HttpHeader])(
+ s3Headers: Seq[HttpHeader])(
implicit mat: Materializer, attr: Attributes): Source[HttpResponse, NotUsed] = {
implicit val sys: ActorSystem = mat.system
implicit val conf: S3Settings = resolveSettings(attr, sys)
@@ -980,7 +980,7 @@ import scala.util.{ Failure, Success, Try }
private def initiateMultipartUpload(s3Location: S3Location,
contentType: ContentType,
- s3Headers: immutable.Seq[HttpHeader]): Source[MultipartUpload, NotUsed] =
+ s3Headers: Seq[HttpHeader]): Source[MultipartUpload, NotUsed] =
Source
.fromMaterializer { (mat, attr) =>
implicit val materializer: Materializer = mat
@@ -1029,7 +1029,7 @@ import scala.util.{ Failure, Success, Try }
.toMat(completionSink(targetLocation, s3Headers))(Keep.right)
}
- private def computeMetaData(headers: immutable.Seq[HttpHeader], entity: ResponseEntity): ObjectMetadata =
+ private def computeMetaData(headers: Seq[HttpHeader], entity: ResponseEntity): ObjectMetadata =
ObjectMetadata(
headers ++
Seq(
@@ -1050,7 +1050,7 @@ import scala.util.{ Failure, Success, Try }
}
private def completeMultipartUpload(s3Location: S3Location,
- parts: immutable.Seq[SuccessfulUploadPart],
+ parts: Seq[SuccessfulUploadPart],
s3Headers: S3Headers)(
implicit mat: Materializer,
attr: Attributes): Future[CompleteMultipartUploadResult] = {
@@ -1078,7 +1078,7 @@ import scala.util.{ Failure, Success, Try }
private def initiateUpload(
s3Location: S3Location,
contentType: ContentType,
- s3Headers: immutable.Seq[HttpHeader]): Source[(MultipartUpload, Int), NotUsed] =
+ s3Headers: Seq[HttpHeader]): Source[(MultipartUpload, Int), NotUsed] =
Source
.single(s3Location)
.flatMapConcat(initiateMultipartUpload(_, contentType, s3Headers))
@@ -1398,9 +1398,9 @@ import scala.util.{ Failure, Success, Try }
import mat.executionContext
Sink
.seq[UploadPartResponse]
- .mapMaterializedValue { (responseFuture: Future[immutable.Seq[UploadPartResponse]]) =>
+ .mapMaterializedValue { (responseFuture: Future[Seq[UploadPartResponse]]) =>
responseFuture
- .flatMap { (responses: immutable.Seq[UploadPartResponse]) =>
+ .flatMap { (responses: Seq[UploadPartResponse]) =>
val successes = responses.collect { case r: SuccessfulUploadPart => r }
val failures = responses.collect { case r: FailedUploadPart => r }
if (responses.isEmpty) {
@@ -1476,7 +1476,7 @@ import scala.util.{ Failure, Success, Try }
}
private def entityForSuccess(
- resp: HttpResponse)(implicit mat: Materializer): Future[(ResponseEntity, immutable.Seq[HttpHeader])] = {
+ resp: HttpResponse)(implicit mat: Materializer): Future[(ResponseEntity, Seq[HttpHeader])] = {
resp match {
case HttpResponse(status, headers, entity, _) if status.isSuccess() && !status.isRedirection() =>
Future.successful((entity, headers))
diff --git a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/model.scala b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/model.scala
index 240cbb866..1300deaae 100644
--- a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/model.scala
+++ b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/model.scala
@@ -22,8 +22,6 @@ import pekko.stream.connectors.s3.AccessStyle.PathAccessStyle
import scala.jdk.OptionConverters._
import scala.annotation.nowarn
-import scala.collection.immutable.Seq
-import scala.collection.immutable
import scala.jdk.CollectionConverters._
final class MFA private (val serialNumber: String, val tokenCode: String) {
@@ -1292,7 +1290,7 @@ final class ObjectMetadata private (
* Java Api
*/
lazy val headers: java.util.List[pekko.http.javadsl.model.HttpHeader] =
- (metadata: immutable.Seq[pekko.http.javadsl.model.HttpHeader]).asJava
+ (metadata: Seq[pekko.http.javadsl.model.HttpHeader]).asJava
/**
* Gets the hex encoded 128-bit MD5 digest of the associated object
diff --git a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/MarshallingSpec.scala b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/MarshallingSpec.scala
index dc5e37b0e..1b905d39d 100644
--- a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/MarshallingSpec.scala
+++ b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/MarshallingSpec.scala
@@ -26,7 +26,6 @@ import org.scalatest.concurrent.ScalaFutures
import org.scalatest.flatspec.AnyFlatSpecLike
import org.scalatest.matchers.should.Matchers
-import scala.collection.immutable.Seq
import scala.concurrent.ExecutionContext
class MarshallingSpec(_system: ActorSystem)
diff --git a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/scaladsl/S3IntegrationSpec.scala b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/scaladsl/S3IntegrationSpec.scala
index 7ddefeba9..ed4221fe9 100644
--- a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/scaladsl/S3IntegrationSpec.scala
+++ b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/scaladsl/S3IntegrationSpec.scala
@@ -715,7 +715,7 @@ trait S3IntegrationSpec
case object AbortException extends Exception("Aborting multipart upload")
- def createSlowSource(data: immutable.Seq[ByteString],
+ def createSlowSource(data: Seq[ByteString],
killSwitch: Option[SharedKillSwitch]): Source[ByteString, NotUsed] = {
val base = Source(data)
.throttle(1, 10.seconds)
diff --git a/simple-codecs/src/test/scala/docs/scaladsl/RecordIOFramingSpec.scala b/simple-codecs/src/test/scala/docs/scaladsl/RecordIOFramingSpec.scala
index 974805778..9fd51b235 100644
--- a/simple-codecs/src/test/scala/docs/scaladsl/RecordIOFramingSpec.scala
+++ b/simple-codecs/src/test/scala/docs/scaladsl/RecordIOFramingSpec.scala
@@ -27,7 +27,6 @@ import pekko.util.ByteString
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.BeforeAndAfterAll
-import scala.collection.immutable.Seq
import scala.concurrent.Future
import org.scalatest.flatspec.AnyFlatSpecLike
import org.scalatest.matchers.should.Matchers
diff --git a/slick/src/test/scala/docs/scaladsl/SlickWithTryResultSpec.scala b/slick/src/test/scala/docs/scaladsl/SlickWithTryResultSpec.scala
index a2805b51b..05104b205 100644
--- a/slick/src/test/scala/docs/scaladsl/SlickWithTryResultSpec.scala
+++ b/slick/src/test/scala/docs/scaladsl/SlickWithTryResultSpec.scala
@@ -61,7 +61,7 @@ class SlickWithTryResultSpec extends AnyWordSpec
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
- val duplicateUser = scala.collection.immutable.Seq(users.head, users.head)
+ val duplicateUser = Seq(users.head, users.head)
val createTable =
sqlu"""CREATE TABLE PEKKO_CONNECTORS_SLICK_SCALADSL_TEST_USERS(ID INTEGER PRIMARY KEY, NAME VARCHAR(50))"""
@@ -321,7 +321,7 @@ class SlickWithTryResultSpec extends AnyWordSpec
}
"produce `Failure[_]` when inserting duplicate record (parallelism = 4)" in {
- val records = scala.collection.immutable.Seq.empty ++ users :+ users.head
+ val records = Seq.empty ++ users :+ users.head
val inserted = Source(records)
.runWith(SlickWithTryResult.sinkTry(parallelism = 4, insertUser))
diff --git a/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/impl/SolrFlowStage.scala b/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/impl/SolrFlowStage.scala
index 6a58f471e..83e98836a 100644
--- a/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/impl/SolrFlowStage.scala
+++ b/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/impl/SolrFlowStage.scala
@@ -26,7 +26,6 @@ import org.apache.solr.client.solrj.response.UpdateResponse
import org.apache.solr.common.SolrInputDocument
import scala.annotation.tailrec
-import scala.collection.immutable
import scala.jdk.CollectionConverters._
import scala.util.control.NonFatal
@@ -42,10 +41,10 @@ private[solr] final class SolrFlowStage[T, C](
client: SolrClient,
settings: SolrUpdateSettings,
messageBinder: T => SolrInputDocument)
- extends GraphStage[FlowShape[immutable.Seq[WriteMessage[T, C]], immutable.Seq[WriteResult[T, C]]]] {
+ extends GraphStage[FlowShape[Seq[WriteMessage[T, C]], Seq[WriteResult[T, C]]]] {
- private val in = Inlet[immutable.Seq[WriteMessage[T, C]]]("messages")
- private val out = Outlet[immutable.Seq[WriteResult[T, C]]]("result")
+ private val in = Inlet[Seq[WriteMessage[T, C]]]("messages")
+ private val out = Outlet[Seq[WriteResult[T, C]]]("result")
override val shape = FlowShape(in, out)
override protected def initialAttributes: Attributes =
@@ -61,9 +60,9 @@ private final class SolrFlowLogic[T, C](
decider: Supervision.Decider,
collection: String,
client: SolrClient,
- in: Inlet[immutable.Seq[WriteMessage[T, C]]],
- out: Outlet[immutable.Seq[WriteResult[T, C]]],
- shape: FlowShape[immutable.Seq[WriteMessage[T, C]], immutable.Seq[WriteResult[T, C]]],
+ in: Inlet[Seq[WriteMessage[T, C]]],
+ out: Outlet[Seq[WriteResult[T, C]]],
+ shape: FlowShape[Seq[WriteMessage[T, C]], Seq[WriteResult[T, C]]],
settings: SolrUpdateSettings,
messageBinder: T => SolrInputDocument) extends GraphStageLogic(shape)
with OutHandler
@@ -94,14 +93,14 @@ private final class SolrFlowLogic[T, C](
pull(in)
}
- private def updateBulkToSolr(messages: immutable.Seq[WriteMessage[T, C]]): UpdateResponse = {
+ private def updateBulkToSolr(messages: Seq[WriteMessage[T, C]]): UpdateResponse = {
val docs = messages.flatMap(_.source.map(messageBinder))
if (log.isDebugEnabled) log.debug("Upsert {}", docs)
client.add(collection, docs.asJava, settings.commitWithin)
}
- private def atomicUpdateBulkToSolr(messages: immutable.Seq[WriteMessage[T, C]]): UpdateResponse = {
+ private def atomicUpdateBulkToSolr(messages: Seq[WriteMessage[T, C]]): UpdateResponse = {
val docs = messages.map { message =>
val doc = new SolrInputDocument()
@@ -147,7 +146,7 @@ private final class SolrFlowLogic[T, C](
}
}
- private def deleteBulkToSolrByIds(messages: immutable.Seq[WriteMessage[T, C]]): UpdateResponse = {
+ private def deleteBulkToSolrByIds(messages: Seq[WriteMessage[T, C]]): UpdateResponse = {
val docsIds = messages
.filter { message =>
message.operation == DeleteByIds && message.idFieldValue.isDefined
@@ -159,7 +158,7 @@ private final class SolrFlowLogic[T, C](
client.deleteById(collection, docsIds.asJava, settings.commitWithin)
}
- private def deleteEachByQuery(messages: immutable.Seq[WriteMessage[T, C]]): UpdateResponse = {
+ private def deleteEachByQuery(messages: Seq[WriteMessage[T, C]]): UpdateResponse = {
val responses = messages.map { message =>
val query = message.query.get
log.debug("Delete by the query {}", query)
@@ -172,10 +171,10 @@ private final class SolrFlowLogic[T, C](
responses.find(_.getStatus != 0).getOrElse(responses.head)
}
- private def sendBulkToSolr(messages: immutable.Seq[WriteMessage[T, C]]): Unit = {
+ private def sendBulkToSolr(messages: Seq[WriteMessage[T, C]]): Unit = {
@tailrec
- def send(toSend: immutable.Seq[WriteMessage[T, C]]): Option[UpdateResponse] = {
+ def send(toSend: Seq[WriteMessage[T, C]]): Option[UpdateResponse] = {
val operation = toSend.head.operation
// Just take a subset of this operation
val (current, remaining) = toSend.span { m =>
diff --git a/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/javadsl/SolrFlow.scala b/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/javadsl/SolrFlow.scala
index 9a2308971..c3e40f400 100644
--- a/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/javadsl/SolrFlow.scala
+++ b/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/javadsl/SolrFlow.scala
@@ -23,7 +23,6 @@ import pekko.stream.scaladsl.Flow
import org.apache.solr.client.solrj.SolrClient
import org.apache.solr.common.SolrInputDocument
-import scala.collection.immutable
import scala.jdk.CollectionConverters._
/**
@@ -42,7 +41,7 @@ object SolrFlow {
WriteResult[SolrInputDocument, NotUsed]], NotUsed] =
Flow
.fromFunction[java.util.List[WriteMessage[SolrInputDocument, NotUsed]],
- immutable.Seq[WriteMessage[SolrInputDocument, NotUsed]]](
+ Seq[WriteMessage[SolrInputDocument, NotUsed]]](
_.asScala.toIndexedSeq)
.via(
scaladsl.SolrFlow
@@ -61,7 +60,7 @@ object SolrFlow {
clazz: Class[T])
: javadsl.Flow[java.util.List[WriteMessage[T, NotUsed]], java.util.List[WriteResult[T, NotUsed]], NotUsed] =
Flow
- .fromFunction[java.util.List[WriteMessage[T, NotUsed]], immutable.Seq[WriteMessage[T, NotUsed]]](
+ .fromFunction[java.util.List[WriteMessage[T, NotUsed]], Seq[WriteMessage[T, NotUsed]]](
_.asScala.toIndexedSeq)
.via(
scaladsl.SolrFlow
@@ -82,7 +81,7 @@ object SolrFlow {
clazz: Class[T])
: javadsl.Flow[java.util.List[WriteMessage[T, NotUsed]], java.util.List[WriteResult[T, NotUsed]], NotUsed] =
Flow
- .fromFunction[java.util.List[WriteMessage[T, NotUsed]], immutable.Seq[WriteMessage[T, NotUsed]]](
+ .fromFunction[java.util.List[WriteMessage[T, NotUsed]], Seq[WriteMessage[T, NotUsed]]](
_.asScala.toIndexedSeq)
.via(
scaladsl.SolrFlow
@@ -103,7 +102,7 @@ object SolrFlow {
SolrInputDocument, PT]], NotUsed] =
Flow
.fromFunction[java.util.List[WriteMessage[SolrInputDocument, PT]],
- immutable.Seq[WriteMessage[SolrInputDocument,
+ Seq[WriteMessage[SolrInputDocument,
PT]]](
_.asScala.toIndexedSeq)
.via(
@@ -124,7 +123,7 @@ object SolrFlow {
client: SolrClient,
clazz: Class[T]): javadsl.Flow[java.util.List[WriteMessage[T, PT]], java.util.List[WriteResult[T, PT]], NotUsed] =
Flow
- .fromFunction[java.util.List[WriteMessage[T, PT]], immutable.Seq[WriteMessage[T, PT]]](_.asScala.toIndexedSeq)
+ .fromFunction[java.util.List[WriteMessage[T, PT]], Seq[WriteMessage[T, PT]]](_.asScala.toIndexedSeq)
.via(
scaladsl.SolrFlow
.beansWithPassThrough[T, PT](collection, settings)(client))
@@ -144,7 +143,7 @@ object SolrFlow {
client: SolrClient,
clazz: Class[T]): javadsl.Flow[java.util.List[WriteMessage[T, PT]], java.util.List[WriteResult[T, PT]], NotUsed] =
Flow
- .fromFunction[java.util.List[WriteMessage[T, PT]], immutable.Seq[WriteMessage[T, PT]]](_.asScala.toIndexedSeq)
+ .fromFunction[java.util.List[WriteMessage[T, PT]], Seq[WriteMessage[T, PT]]](_.asScala.toIndexedSeq)
.via(
scaladsl.SolrFlow
.typedsWithPassThrough[T, PT](collection, settings, i => binder.apply(i))(client))
diff --git a/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/scaladsl/SolrFlow.scala b/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/scaladsl/SolrFlow.scala
index e7297ffc1..04f1ae1da 100644
--- a/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/scaladsl/SolrFlow.scala
+++ b/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/scaladsl/SolrFlow.scala
@@ -21,8 +21,6 @@ import pekko.stream.scaladsl.Flow
import org.apache.solr.client.solrj.SolrClient
import org.apache.solr.common.SolrInputDocument
-import scala.collection.immutable
-
/**
* Scala API
*/
@@ -34,8 +32,8 @@ object SolrFlow {
def documents(
collection: String,
settings: SolrUpdateSettings)(
- implicit client: SolrClient): Flow[immutable.Seq[WriteMessage[SolrInputDocument, NotUsed]],
- immutable.Seq[
+ implicit client: SolrClient): Flow[Seq[WriteMessage[SolrInputDocument, NotUsed]],
+ Seq[
WriteResult[SolrInputDocument, NotUsed]], NotUsed] =
Flow
.fromGraph(
@@ -53,7 +51,7 @@ object SolrFlow {
collection: String,
settings: SolrUpdateSettings)(
implicit client: SolrClient)
- : Flow[immutable.Seq[WriteMessage[T, NotUsed]], immutable.Seq[WriteResult[T, NotUsed]], NotUsed] =
+ : Flow[Seq[WriteMessage[T, NotUsed]], Seq[WriteResult[T, NotUsed]], NotUsed] =
Flow
.fromGraph(
new SolrFlowStage[T, NotUsed](
@@ -72,7 +70,7 @@ object SolrFlow {
settings: SolrUpdateSettings,
binder: T => SolrInputDocument)(
implicit client: SolrClient)
- : Flow[immutable.Seq[WriteMessage[T, NotUsed]], immutable.Seq[WriteResult[T, NotUsed]], NotUsed] =
+ : Flow[Seq[WriteMessage[T, NotUsed]], Seq[WriteResult[T, NotUsed]], NotUsed] =
Flow
.fromGraph(
new SolrFlowStage[T, NotUsed](
@@ -89,8 +87,8 @@ object SolrFlow {
def documentsWithPassThrough[PT](
collection: String,
settings: SolrUpdateSettings)(
- implicit client: SolrClient): Flow[immutable.Seq[WriteMessage[SolrInputDocument, PT]],
- immutable.Seq[WriteResult[SolrInputDocument, PT]], NotUsed] =
+ implicit client: SolrClient): Flow[Seq[WriteMessage[SolrInputDocument, PT]],
+ Seq[WriteResult[SolrInputDocument, PT]], NotUsed] =
Flow
.fromGraph(
new SolrFlowStage[SolrInputDocument, PT](
@@ -108,7 +106,7 @@ object SolrFlow {
def beansWithPassThrough[T, PT](
collection: String,
settings: SolrUpdateSettings)(implicit client: SolrClient)
- : Flow[immutable.Seq[WriteMessage[T, PT]], immutable.Seq[WriteResult[T, PT]], NotUsed] =
+ : Flow[Seq[WriteMessage[T, PT]], Seq[WriteResult[T, PT]], NotUsed] =
Flow
.fromGraph(
new SolrFlowStage[T, PT](
@@ -127,7 +125,7 @@ object SolrFlow {
collection: String,
settings: SolrUpdateSettings,
binder: T => SolrInputDocument)(implicit client: SolrClient)
- : Flow[immutable.Seq[WriteMessage[T, PT]], immutable.Seq[WriteResult[T, PT]], NotUsed] =
+ : Flow[Seq[WriteMessage[T, PT]], Seq[WriteResult[T, PT]], NotUsed] =
Flow
.fromGraph(
new SolrFlowStage[T, PT](
diff --git a/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/scaladsl/SolrSink.scala b/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/scaladsl/SolrSink.scala
index 386b48139..2a0320aa9 100644
--- a/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/scaladsl/SolrSink.scala
+++ b/solr/src/main/scala/org/apache/pekko/stream/connectors/solr/scaladsl/SolrSink.scala
@@ -20,7 +20,6 @@ import pekko.{ Done, NotUsed }
import org.apache.solr.client.solrj.SolrClient
import org.apache.solr.common.SolrInputDocument
-import scala.collection.immutable
import scala.concurrent.Future
/**
@@ -32,7 +31,7 @@ object SolrSink {
* Write `SolrInputDocument`s to Solr.
*/
def documents[T](collection: String, settings: SolrUpdateSettings)(
- implicit client: SolrClient): Sink[immutable.Seq[WriteMessage[SolrInputDocument, NotUsed]], Future[Done]] =
+ implicit client: SolrClient): Sink[Seq[WriteMessage[SolrInputDocument, NotUsed]], Future[Done]] =
SolrFlow
.documents(collection, settings)
.toMat(Sink.ignore)(Keep.right)
@@ -42,7 +41,7 @@ object SolrSink {
* The stream element classes must be annotated for use with [[org.apache.solr.client.solrj.beans.DocumentObjectBinder]] for conversion.
*/
def beans[T](collection: String, settings: SolrUpdateSettings)(
- implicit client: SolrClient): Sink[immutable.Seq[WriteMessage[T, NotUsed]], Future[Done]] =
+ implicit client: SolrClient): Sink[Seq[WriteMessage[T, NotUsed]], Future[Done]] =
SolrFlow
.beans[T](collection, settings)
.toMat(Sink.ignore)(Keep.right)
@@ -56,7 +55,7 @@ object SolrSink {
collection: String,
settings: SolrUpdateSettings,
binder: T => SolrInputDocument)(
- implicit client: SolrClient): Sink[immutable.Seq[WriteMessage[T, NotUsed]], Future[Done]] =
+ implicit client: SolrClient): Sink[Seq[WriteMessage[T, NotUsed]], Future[Done]] =
SolrFlow
.typeds[T](collection, settings, binder)
.toMat(Sink.ignore)(Keep.right)
diff --git a/solr/src/test/scala/docs/scaladsl/SolrSpec.scala b/solr/src/test/scala/docs/scaladsl/SolrSpec.scala
index 4d9c77485..13fcbc972 100644
--- a/solr/src/test/scala/docs/scaladsl/SolrSpec.scala
+++ b/solr/src/test/scala/docs/scaladsl/SolrSpec.scala
@@ -37,7 +37,6 @@ import org.scalatest.concurrent.ScalaFutures
import org.scalatest.BeforeAndAfterAll
import scala.annotation.nowarn
-import scala.collection.immutable
import scala.concurrent.duration._
import scala.concurrent.{ ExecutionContext, Future }
import org.scalatest.matchers.should.Matchers
@@ -277,7 +276,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
}
}
- case class CommittableOffsetBatch(offsets: immutable.Seq[CommittableOffset]) {
+ case class CommittableOffsetBatch(offsets: Seq[CommittableOffset]) {
def commitScaladsl(): Future[Done] = {
committedOffsets = committedOffsets ++ offsets
Future.successful(Done)
@@ -644,7 +643,7 @@ class SolrSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with Sca
}
}
- case class CommittableOffsetBatch(offsets: immutable.Seq[CommittableOffset]) {
+ case class CommittableOffsetBatch(offsets: Seq[CommittableOffset]) {
def commitScaladsl(): Future[Done] = {
committedOffsets = committedOffsets ++ offsets
Future.successful(Done)
diff --git a/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/SqsSourceSettings.scala b/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/SqsSourceSettings.scala
index 5a946b924..1fadeb38b 100644
--- a/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/SqsSourceSettings.scala
+++ b/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/SqsSourceSettings.scala
@@ -17,7 +17,6 @@ import java.time.temporal.ChronoUnit
import software.amazon.awssdk.services.sqs.model
-import scala.collection.immutable
import scala.concurrent.duration.FiniteDuration
import scala.jdk.CollectionConverters._
@@ -26,8 +25,8 @@ final class SqsSourceSettings private (
val maxBufferSize: Int,
val parallelRequests: Int,
val maxBatchSize: Int,
- val attributeNames: immutable.Seq[MessageSystemAttributeName],
- val messageAttributeNames: immutable.Seq[MessageAttributeName],
+ val attributeNames: Seq[MessageSystemAttributeName],
+ val messageAttributeNames: Seq[MessageAttributeName],
val closeOnEmptyReceive: Boolean,
val visibilityTimeout: Option[FiniteDuration]) {
require(maxBatchSize <= maxBufferSize, "maxBatchSize must be lower or equal than maxBufferSize")
@@ -79,8 +78,8 @@ final class SqsSourceSettings private (
def withMaxBatchSize(maxBatchSize: Int): SqsSourceSettings = copy(maxBatchSize = maxBatchSize)
def withAttribute(attribute: MessageSystemAttributeName): SqsSourceSettings =
- copy(attributeNames = immutable.Seq(attribute))
- def withAttributes(attributes: immutable.Seq[MessageSystemAttributeName]): SqsSourceSettings =
+ copy(attributeNames = Seq(attribute))
+ def withAttributes(attributes: Seq[MessageSystemAttributeName]): SqsSourceSettings =
copy(attributeNames = attributes)
/** Java API */
@@ -88,8 +87,8 @@ final class SqsSourceSettings private (
copy(attributeNames = attributes.asScala.toList)
def withMessageAttribute(attributes: MessageAttributeName): SqsSourceSettings =
- copy(messageAttributeNames = immutable.Seq(attributes))
- def withMessageAttributes(attributes: immutable.Seq[MessageAttributeName]): SqsSourceSettings =
+ copy(messageAttributeNames = Seq(attributes))
+ def withMessageAttributes(attributes: Seq[MessageAttributeName]): SqsSourceSettings =
copy(messageAttributeNames = attributes)
/** Java API */
@@ -119,8 +118,8 @@ final class SqsSourceSettings private (
maxBufferSize: Int = maxBufferSize,
parallelRequests: Int = parallelRequests,
maxBatchSize: Int = maxBatchSize,
- attributeNames: immutable.Seq[MessageSystemAttributeName] = attributeNames,
- messageAttributeNames: immutable.Seq[MessageAttributeName] = messageAttributeNames,
+ attributeNames: Seq[MessageSystemAttributeName] = attributeNames,
+ messageAttributeNames: Seq[MessageAttributeName] = messageAttributeNames,
closeOnEmptyReceive: Boolean = closeOnEmptyReceive,
visibilityTimeout: Option[FiniteDuration] = visibilityTimeout): SqsSourceSettings = new SqsSourceSettings(
waitTimeSeconds,
@@ -151,8 +150,8 @@ object SqsSourceSettings {
maxBufferSize = 100,
parallelRequests = 1,
maxBatchSize = 10,
- attributeNames = immutable.Seq.empty,
- messageAttributeNames = immutable.Seq.empty,
+ attributeNames = Seq.empty,
+ messageAttributeNames = Seq.empty,
closeOnEmptyReceive = false,
visibilityTimeout = None)
diff --git a/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsAckFlow.scala b/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsAckFlow.scala
index 8f4baea98..c32fed40d 100644
--- a/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsAckFlow.scala
+++ b/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsAckFlow.scala
@@ -27,7 +27,6 @@ import pekko.stream.scaladsl.{ Flow, GraphDSL, Merge, Partition }
import software.amazon.awssdk.services.sqs.SqsAsyncClient
import software.amazon.awssdk.services.sqs.model._
-import scala.collection.immutable
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.parasitic
import scala.jdk.CollectionConverters._
@@ -133,7 +132,7 @@ object SqsAckFlow {
.build()
}
.mapAsync(settings.concurrentRequests) {
- case (actions: immutable.Seq[Delete], request) =>
+ case (actions: Seq[Delete], request) =>
sqsClient
.deleteMessageBatch(request)
.asScala
diff --git a/sqs/src/test/scala/docs/scaladsl/SqsSourceSpec.scala b/sqs/src/test/scala/docs/scaladsl/SqsSourceSpec.scala
index 3ec5e4044..ded4a3c88 100644
--- a/sqs/src/test/scala/docs/scaladsl/SqsSourceSpec.scala
+++ b/sqs/src/test/scala/docs/scaladsl/SqsSourceSpec.scala
@@ -39,7 +39,6 @@ import software.amazon.awssdk.services.sqs.model.{
SendMessageRequest
}
-import scala.collection.immutable
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
@@ -203,7 +202,7 @@ class SqsSourceSpec extends AnyFlatSpec with ScalaFutures with Matchers with Def
.withWaitTime(20.seconds)
.withMaxBufferSize(100)
.withMaxBatchSize(10)
- .withAttributes(immutable.Seq(SenderId, SentTimestamp))
+ .withAttributes(Seq(SenderId, SentTimestamp))
.withMessageAttribute(MessageAttributeName.create("bar.*"))
.withCloseOnEmptyReceive(true)
.withVisibilityTimeout(10.seconds)
@@ -299,7 +298,7 @@ class SqsSourceSpec extends AnyFlatSpec with ScalaFutures with Matchers with Def
input.foreach(m => sqsClient.sendMessage(m).get(2, TimeUnit.SECONDS))
// #run
- val messages: Future[immutable.Seq[Message]] =
+ val messages: Future[Seq[Message]] =
SqsSource(
queueUrl,
SqsSourceSettings().withCloseOnEmptyReceive(true).withWaitTime(10.millis)).runWith(Sink.seq)
diff --git a/sse/src/test/scala/docs/scaladsl/EventSourceSpec.scala b/sse/src/test/scala/docs/scaladsl/EventSourceSpec.scala
index 34804f85f..fccb0cf0e 100644
--- a/sse/src/test/scala/docs/scaladsl/EventSourceSpec.scala
+++ b/sse/src/test/scala/docs/scaladsl/EventSourceSpec.scala
@@ -30,7 +30,6 @@ import pekko.testkit.SocketUtil
import pekko.{ Done, NotUsed }
import org.scalatest.BeforeAndAfterAll
-import scala.collection.immutable
import scala.concurrent.duration.DurationInt
import scala.concurrent.{ Await, ExecutionContext, Future }
//#event-source
@@ -168,7 +167,7 @@ final class EventSourceSpec extends AsyncWordSpec with Matchers with BeforeAndAf
// #event-source
// #consume-events
- val events: Future[immutable.Seq[ServerSentEvent]] =
+ val events: Future[Seq[ServerSentEvent]] =
eventSource
.throttle(elements = 1, per = 500.milliseconds, maximumBurst = 1, ThrottleMode.Shaping)
.take(nrOfSamples)
diff --git a/text/src/test/scala/docs/scaladsl/CharsetCodingFlowsDoc.scala b/text/src/test/scala/docs/scaladsl/CharsetCodingFlowsDoc.scala
index 0fd3d7373..0269c363d 100644
--- a/text/src/test/scala/docs/scaladsl/CharsetCodingFlowsDoc.scala
+++ b/text/src/test/scala/docs/scaladsl/CharsetCodingFlowsDoc.scala
@@ -27,7 +27,6 @@ import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
import org.scalatest.{ BeforeAndAfterAll, RecoverMethods }
-import scala.collection.immutable
import scala.concurrent.Future
class CharsetCodingFlowsDoc
@@ -90,7 +89,7 @@ class CharsetCodingFlowsDoc
.single(utf16bytes)
// #decoding
- val result: Future[immutable.Seq[String]] =
+ val result: Future[Seq[String]] =
byteStringSource
.via(TextFlow.decoding(StandardCharsets.UTF_16))
.runWith(Sink.seq)
diff --git a/text/src/test/scala/org/apache/pekko/stream/connectors/text/scaladsl/CharsetCodingFlowsSpec.scala b/text/src/test/scala/org/apache/pekko/stream/connectors/text/scaladsl/CharsetCodingFlowsSpec.scala
index 95d6e2676..1b733c6d0 100644
--- a/text/src/test/scala/org/apache/pekko/stream/connectors/text/scaladsl/CharsetCodingFlowsSpec.scala
+++ b/text/src/test/scala/org/apache/pekko/stream/connectors/text/scaladsl/CharsetCodingFlowsSpec.scala
@@ -29,7 +29,6 @@ import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
import org.scalatest.{ BeforeAndAfterAll, RecoverMethods }
-import scala.collection.immutable
import scala.concurrent.duration.DurationInt
import scala.concurrent.{ ExecutionContextExecutor, Future }
@@ -97,7 +96,7 @@ class CharsetCodingFlowsSpec
Source
.single(utf16bytes)
- val result: Future[immutable.Seq[String]] =
+ val result: Future[Seq[String]] =
byteStringSource
.via(TextFlow.decoding(StandardCharsets.UTF_16))
.runWith(Sink.seq)
diff --git a/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/Subslice.scala b/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/Subslice.scala
index bdd78f8fd..3fbc416e3 100644
--- a/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/Subslice.scala
+++ b/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/Subslice.scala
@@ -19,12 +19,10 @@ import pekko.stream.{ Attributes, FlowShape, Inlet, Outlet }
import pekko.stream.connectors.xml.{ EndElement, ParseEvent, StartElement }
import pekko.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler }
-import scala.collection.immutable
-
/**
* INTERNAL API
*/
-@InternalApi private[xml] class Subslice(path: immutable.Seq[String])
+@InternalApi private[xml] class Subslice(path: Seq[String])
extends GraphStage[FlowShape[ParseEvent, ParseEvent]] {
val in: Inlet[ParseEvent] = Inlet("XMLSubslice.in")
val out: Outlet[ParseEvent] = Outlet("XMLSubslice.out")
diff --git a/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/Subtree.scala b/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/Subtree.scala
index 1e64b2485..23f39164a 100644
--- a/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/Subtree.scala
+++ b/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/impl/Subtree.scala
@@ -21,12 +21,10 @@ import pekko.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler }
import javax.xml.parsers.DocumentBuilderFactory
import org.w3c.dom.Element
-import scala.collection.immutable
-
/**
* INTERNAL API
*/
-@InternalApi private[xml] class Subtree(path: immutable.Seq[String])
+@InternalApi private[xml] class Subtree(path: Seq[String])
extends GraphStage[FlowShape[ParseEvent, Element]] {
val in: Inlet[ParseEvent] = Inlet("XMLSubtree.in")
val out: Outlet[Element] = Outlet("XMLSubtree.out")
diff --git a/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/scaladsl/XmlParsing.scala b/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/scaladsl/XmlParsing.scala
index 06e12d33e..4e7a905b4 100644
--- a/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/scaladsl/XmlParsing.scala
+++ b/xml/src/main/scala/org/apache/pekko/stream/connectors/xml/scaladsl/XmlParsing.scala
@@ -25,8 +25,6 @@ import org.w3c.dom.Element
import javax.xml.stream.XMLInputFactory
import javax.xml.XMLConstants
-import scala.collection.immutable
-
object XmlParsing {
/**
@@ -88,14 +86,14 @@ object XmlParsing {
* a certain path in the XML document. Any event that is under the specified path (including subpaths) is passed
* through.
*/
- def subslice(path: immutable.Seq[String]): Flow[ParseEvent, ParseEvent, NotUsed] =
+ def subslice(path: Seq[String]): Flow[ParseEvent, ParseEvent, NotUsed] =
Flow.fromGraph(new impl.Subslice(path))
/**
* A Flow that transforms a stream of XML ParseEvents. This stage pushes elements of a certain path in
* the XML document as org.w3c.dom.Element.
*/
- def subtree(path: immutable.Seq[String]): Flow[ParseEvent, Element, NotUsed] =
+ def subtree(path: Seq[String]): Flow[ParseEvent, Element, NotUsed] =
Flow.fromGraph(new impl.Subtree(path))
}
diff --git a/xml/src/test/scala/docs/scaladsl/XmlProcessingSpec.scala b/xml/src/test/scala/docs/scaladsl/XmlProcessingSpec.scala
index 125e8e948..f76f00bce 100644
--- a/xml/src/test/scala/docs/scaladsl/XmlProcessingSpec.scala
+++ b/xml/src/test/scala/docs/scaladsl/XmlProcessingSpec.scala
@@ -72,7 +72,7 @@ class XmlProcessingSpec extends AnyWordSpec with Matchers with ScalaFutures with
"properly parse simple XML and read it" in {
// #parser-to-data
val doc = ByteString("<doc><elem>elem1</elem><elem>elem2</elem></doc>")
- val result: Future[immutable.Seq[String]] = Source
+ val result: Future[Seq[String]] = Source
.single(doc)
.via(XmlParsing.parser)
.statefulMap(() => new StringBuilder())((textBuffer, parseEvent) => {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org