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 2a3c48e66 switch to lambdas (#1777)
2a3c48e66 is described below
commit 2a3c48e66437dfb2d0223fbec066e9252f1d0f97
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Jul 15 09:50:42 2026 +0100
switch to lambdas (#1777)
---
.../connectors/azure/storagequeue/javadsl/AzureQueueSink.scala | 10 +++++-----
.../azure/storagequeue/javadsl/AzureQueueSource.scala | 8 ++++----
sqs/src/test/scala/docs/scaladsl/SqsAckSpec.scala | 7 +++----
.../stream/connectors/sqs/scaladsl/SqsPublishSinkSpec.scala | 10 +++-------
.../unixdomainsocket/impl/UnixDomainSocketImpl.scala | 10 ++--------
5 files changed, 17 insertions(+), 28 deletions(-)
diff --git
a/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSink.scala
b/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSink.scala
index 8bb0360db..890dcbc8d 100644
---
a/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSink.scala
+++
b/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSink.scala
@@ -13,15 +13,15 @@
package org.apache.pekko.stream.connectors.azure.storagequeue.javadsl
-import com.microsoft.azure.storage.queue.{ CloudQueue, CloudQueueMessage }
-import org.apache.pekko
-import pekko.stream.connectors.azure.storagequeue.impl.AzureQueueSinkFunctions
-import pekko.stream.javadsl.Sink
-import pekko.Done
import java.util.concurrent.CompletionStage
import java.util.function.Supplier
+import com.microsoft.azure.storage.queue.{ CloudQueue, CloudQueueMessage }
+import org.apache.pekko
+import pekko.Done
import pekko.stream.connectors.azure.storagequeue.DeleteOrUpdateMessage
+import pekko.stream.connectors.azure.storagequeue.impl.AzureQueueSinkFunctions
+import pekko.stream.javadsl.Sink
object AzureQueueSink {
diff --git
a/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSource.scala
b/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSource.scala
index acdfefe19..1468fe1f0 100644
---
a/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSource.scala
+++
b/azure-storage-queue/src/main/scala/org/apache/pekko/stream/connectors/azure/storagequeue/javadsl/AzureQueueSource.scala
@@ -13,14 +13,14 @@
package org.apache.pekko.stream.connectors.azure.storagequeue.javadsl
+import java.util.function.Supplier
+
import com.microsoft.azure.storage.queue.{ CloudQueue, CloudQueueMessage }
import org.apache.pekko
-import pekko.stream.connectors.azure.storagequeue.AzureQueueSourceSettings
-import pekko.stream.javadsl.Source
import pekko.NotUsed
-import java.util.function.Supplier
-
+import pekko.stream.connectors.azure.storagequeue.AzureQueueSourceSettings
import pekko.stream.connectors.azure.storagequeue.impl.AzureQueueSourceStage
+import pekko.stream.javadsl.Source
object AzureQueueSource {
diff --git a/sqs/src/test/scala/docs/scaladsl/SqsAckSpec.scala
b/sqs/src/test/scala/docs/scaladsl/SqsAckSpec.scala
index 485fa32a0..82597febc 100644
--- a/sqs/src/test/scala/docs/scaladsl/SqsAckSpec.scala
+++ b/sqs/src/test/scala/docs/scaladsl/SqsAckSpec.scala
@@ -14,7 +14,7 @@
package docs.scaladsl
import java.util.concurrent.{ CompletableFuture, TimeUnit }
-import java.util.function.Supplier
+
import org.apache.pekko
import pekko.Done
import pekko.stream.connectors.sqs.scaladsl._
@@ -279,9 +279,8 @@ class SqsAckSpec extends AnyFlatSpec with Matchers with
DefaultTestContext with
when(mockAwsSqsClient.deleteMessageBatch(any[DeleteMessageBatchRequest]))
.thenReturn(
- CompletableFuture.supplyAsync[DeleteMessageBatchResponse](new
Supplier[DeleteMessageBatchResponse] {
- override def get(): DeleteMessageBatchResponse = throw new
RuntimeException("Fake client error")
- }))
+ CompletableFuture.supplyAsync[DeleteMessageBatchResponse](() =>
+ throw new RuntimeException("Fake client error")))
val future = Source(messages)
.take(10)
diff --git
a/sqs/src/test/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsPublishSinkSpec.scala
b/sqs/src/test/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsPublishSinkSpec.scala
index 854bcd3bb..b6a5c2eb5 100644
---
a/sqs/src/test/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsPublishSinkSpec.scala
+++
b/sqs/src/test/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsPublishSinkSpec.scala
@@ -15,7 +15,6 @@ package org.apache.pekko.stream.connectors.sqs.scaladsl
import java.util.UUID
import java.util.concurrent.{ CompletableFuture, TimeUnit }
-import java.util.function.Supplier
import org.apache.pekko
import pekko.Done
@@ -54,9 +53,7 @@ class SqsPublishSinkSpec extends AnyFlatSpec with Matchers
with DefaultTestConte
when(sqsClient.sendMessage(any[SendMessageRequest]()))
.thenReturn(
- CompletableFuture.supplyAsync[SendMessageResponse](new
Supplier[SendMessageResponse] {
- override def get(): SendMessageResponse = throw new
RuntimeException("Fake client error")
- }))
+ CompletableFuture.supplyAsync[SendMessageResponse](() => throw new
RuntimeException("Fake client error")))
val (probe, future) =
TestSource[String]().toMat(SqsPublishSink("notused"))(Keep.both).run()
probe.sendNext("notused").sendComplete()
@@ -221,9 +218,8 @@ class SqsPublishSinkSpec extends AnyFlatSpec with Matchers
with DefaultTestConte
implicit val sqsClient: SqsAsyncClient = mock[SqsAsyncClient]
when(
sqsClient.sendMessageBatch(any[SendMessageBatchRequest]())).thenReturn(
- CompletableFuture.supplyAsync[SendMessageBatchResponse](new
Supplier[SendMessageBatchResponse] {
- override def get(): SendMessageBatchResponse = throw new
RuntimeException("Fake client error")
- }))
+ CompletableFuture.supplyAsync[SendMessageBatchResponse](() => throw new
RuntimeException("Fake client error"))
+ )
val settings = SqsPublishGroupedSettings().withMaxBatchSize(5)
val (probe, future) =
TestSource[String]().toMat(SqsPublishSink.grouped("notused",
settings))(Keep.both).run()
diff --git
a/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala
b/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala
index a3b5da2f0..040c8bcd7 100644
---
a/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala
+++
b/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala
@@ -472,14 +472,8 @@ private[unixdomainsocket] abstract class
UnixDomainSocketImpl(system: ExtendedAc
val connectionFinished = Promise[Done]()
val cancellable =
connectTimeout match {
- case d: FiniteDuration =>
- Some(system.scheduler.scheduleOnce(d,
- new Runnable {
- override def run(): Unit =
- channel.close()
- }))
- case _ =>
- None
+ case d: FiniteDuration => Some(system.scheduler.scheduleOnce(d, ()
=> channel.close()))
+ case _ => None
}
val (context, connectionFlow) = sendReceiveStructures(sel,
receiveBufferSize, sendBufferSize, halfClose)
val ra = new JnrUnixSocketAddress(remoteAddress.path.toFile)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
