This is an automated email from the ASF dual-hosted git repository.
mdedetrich 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 9160702d7 Missing death watch in UDP.bindFlow(...)
9160702d7 is described below
commit 9160702d77008e280fcafebe551fb6de42035ad7
Author: sdeigm <46...@users.noreply.github.com>
AuthorDate: Mon Mar 31 13:37:31 2025 +0200
Missing death watch in UDP.bindFlow(...)
---
.../pekko/stream/connectors/udp/impl/UdpBind.scala | 16 +++++++++++-----
1 file changed, 11 insertions(+), 5 deletions(-)
diff --git a/udp/src/main/scala/org/apache/pekko/stream/connectors/udp/impl/UdpBind.scala b/udp/src/main/scala/org/apache/pekko/stream/connectors/udp/impl/UdpBind.scala
index f285662b2..2eea60802 100644
--- a/udp/src/main/scala/org/apache/pekko/stream/connectors/udp/impl/UdpBind.scala
+++ b/udp/src/main/scala/org/apache/pekko/stream/connectors/udp/impl/UdpBind.scala
@@ -16,7 +16,7 @@ package org.apache.pekko.stream.connectors.udp.impl
import java.net.InetSocketAddress
import org.apache.pekko
-import pekko.actor.{ ActorRef, ActorSystem }
+import pekko.actor.{ ActorRef, ActorSystem, Terminated }
import pekko.annotation.InternalApi
import pekko.io.{ IO, Udp }
import pekko.io.Inet.SocketOption
@@ -24,6 +24,7 @@ import pekko.stream.{ Attributes, FlowShape, Inlet, Outlet }
import pekko.stream.connectors.udp.Datagram
import pekko.stream.stage._
+import java.io.IOException
import scala.collection.immutable.Iterable
import scala.concurrent.{ Future, Promise }
@@ -42,7 +43,7 @@ import scala.concurrent.{ Future, Promise }
private var listener: ActorRef = _
override def preStart(): Unit = {
- implicit val sender = getStageActor(processIncoming).ref
+ implicit val sender: ActorRef = getStageActor(processIncoming).ref
IO(Udp) ! Udp.Bind(sender, localAddress, options)
}
@@ -53,6 +54,7 @@ import scala.concurrent.{ Future, Promise }
case (sender, Udp.Bound(boundAddress)) =>
boundPromise.success(boundAddress)
listener = sender
+ stageActor.watch(listener)
pull(in)
case (_, Udp.CommandFailed(cmd: Udp.Bind)) =>
val ex = new IllegalArgumentException(s"Unable to bind to [${cmd.localAddress}]")
@@ -62,10 +64,13 @@ import scala.concurrent.{ Future, Promise }
if (isAvailable(out)) {
push(out, Datagram(data, sender))
}
+ case (_, Terminated(_)) =>
+ listener = null
+ failStage(new IOException("UDP listener terminated unexpectedly"))
case _ =>
}
- private def unbindListener() =
+ private def unbindListener(): Unit =
if (listener != null) {
listener ! Udp.Unbind
}
@@ -73,7 +78,7 @@ import scala.concurrent.{ Future, Promise }
setHandler(
in,
new InHandler {
- override def onPush() = {
+ override def onPush(): Unit = {
val msg = grab(in)
listener ! Udp.Send(msg.data, msg.remote)
pull(in)
@@ -96,7 +101,8 @@ import scala.concurrent.{ Future, Promise }
val out: Outlet[Datagram] = Outlet("UdpBindFlow.in")
val shape: FlowShape[Datagram, Datagram] = FlowShape.of(in, out)
- override def createLogicAndMaterializedValue(inheritedAttributes: Attributes) = {
+ override def createLogicAndMaterializedValue(
+ inheritedAttributes: Attributes): (UdpBindLogic, Future[InetSocketAddress]) = {
val boundPromise = Promise[InetSocketAddress]()
(new UdpBindLogic(localAddress, options, boundPromise)(shape), boundPromise.future)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org