Replies: 2 comments 2 replies
-
|
Sounds like an interesting case! Indeed you're quite correct that when the client disconnects, I'd expect it to send a RST on the response socket, which should convert to a stream cancellation / Ideally I think you'd want to override the It's interesting that you note that it does get cancelled as expected with a Could you share your async attempt as well? Of course it'd be interesting if you can reproduce this problem in more isolation, for example in a (unit) test where you cancel the stream 'directly' instead of relying on Pekko HTTP and Play. |
Beta Was this translation helpful? Give feedback.
1 reply
-
|
Thanks for your message @raboof . Here's a (not so) minified version of the code I was using. However, as often , now that I'm sharing this with you, I was able to make the async variant work. I'll need to review the more complex real code to see if I missed something and async does work or if there's something else preventing async to work. Sync version object PekkoMain {
private val logger = org.slf4j.LoggerFactory.getLogger(this.getClass)
def main(args: Array[String]): Unit = {
logger.info("Hello, Pekko!")
implicit val system = ActorSystem("PekkoSystem")
implicit val ec = system.dispatcher
val ks = KillSwitches.shared("kill-switch")
// KillSwitch is not able to stop the flow
val source: Source[String, _] = CustomSource.source()
// KillSwitch does stop the flow
// val source: Source[String, _] = Source.repeat("Repeat").initialDelay(15.seconds).delay(1.second).take(20)
val done: Future[Done] = source
.via(ks.flow)
.watchTermination() { (_, f) =>
f.onComplete { _ =>
logger.info(s"onComplete")
}
NotUsed
}
.runForeach(logger.info)
Future {
Thread.sleep(10 * 1000)
logger.info("Trigger KillSwitch")
ks.shutdown()
}
Await.result(done, 60.seconds)
Await.result(system.terminate(), 60.seconds)
()
}
}
object CustomSource {
private val logger = org.slf4j.LoggerFactory.getLogger(this.getClass)
def source(): Source[String, Future[Int]] = Source.fromGraph(new ResultSource())
private class ResultSource() extends GraphStageWithMaterializedValue[SourceShape[String], Future[Int]] {
val out: Outlet[String] = Outlet(s"$toString.out")
val shape: SourceShape[String] = SourceShape(out)
override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Future[Int]) = {
val result = Promise[Int]()
val logic = new GraphStageLogic(shape) with OutHandler {
private var cursor: Option[String] = None
private var counter: Int = 0
private var cancel = false
override def postStop() = release()
private def release(): Unit = {
logger.info("Releasing resources")
}
private def nextCursor(): Unit = {
if (cancel) {
logger.info("nextCursor -> cancelled")
throw new RuntimeException("Cancelled")
}
cursor = if (counter == 0) {
logger.info("nextCursor -> Sleeping for 15 seconds")
(1 to 15).foreach { _ =>
if (!cancel) {
logger.info("nextCursor -> Sleeping")
Thread.sleep(1000)
} else {
logger.info("nextCursor -> cancelled")
throw new RuntimeException("Cancelled")
}
}
Some(s"Item $counter")
} else if (counter < 50) {
logger.info("nextCursor -> Immediate result")
Some(s"Item $counter")
} else {
logger.info("nextCursor -> No more results")
None
}
}
def onPull(): Unit = {
nextCursor()
cursor match {
case Some(c) =>
counter += 1
push(out, c)
case _ =>
result.success(counter)
complete(out)
}
}
@nowarn
override def onDownstreamFinish() = {
logger.info("onDownstreamFinish")
result.tryFailure(new InterruptedException("Downstream finished"))
release()
cancel = true
super.onDownstreamFinish(): @nowarn
}
setHandler(out, this)
}
logic -> result.future
}
}
}Gives this output: Async version object PekkoMain {
private val logger = org.slf4j.LoggerFactory.getLogger(this.getClass)
def main(args: Array[String]): Unit = {
// Same as for sync with:
val blockingEc = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(4))
val source: Source[String, _] = CustomSource.asyncSource(blockingEc)
}
}
object CustomSource {
private val logger = org.slf4j.LoggerFactory.getLogger(this.getClass)
def asyncSource(blockingEc: ExecutionContext): Source[String, Future[Int]] = Source.fromGraph(new AsyncResultSource(blockingEc))
private class AsyncResultSource(blockingEc: ExecutionContext) extends GraphStageWithMaterializedValue[SourceShape[String], Future[Int]] {
val out: Outlet[String] = Outlet(s"$toString.out")
val shape: SourceShape[String] = SourceShape(out)
override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Future[Int]) = {
val result = Promise[Int]()
val logic = new GraphStageLogic(shape) with OutHandler {
private var counter: Int = 0
private var cancel = false
val cb: AsyncCallback[Try[Option[String]]] = getAsyncCallback[Try[Option[String]]] { itemOptTry =>
itemOptTry match {
case Success(itemOpt) =>
itemOpt match {
case Some(c) =>
counter += 1
push(out, c)
case _ =>
result.success(counter)
complete(out)
}
case Failure(ex) => failStage(ex)
}
}
override def postStop() = release()
private def release(): Unit = {
logger.info("Releasing resources")
}
private def nextCursor(): Unit = {
if (cancel) {
logger.info("nextCursor -> cancelled")
throw new RuntimeException("Cancelled")
}
val f = Future {
blocking {
if (counter == 0) {
logger.info("nextCursor -> Sleeping for 15 seconds")
logger.info("nextCursor -> Sleeping")
Thread.sleep(15000)
Some(s"Item $counter")
} else if (counter < 50) {
logger.info("nextCursor -> Immediate result")
Some(s"Item $counter")
} else {
logger.info("nextCursor -> No more results")
None
}
}
}(blockingEc)
f.onComplete(cb.invoke)(ExecutionContext.parasitic)
}
def onPull(): Unit = {
nextCursor()
}
@nowarn
override def onDownstreamFinish() = {
logger.info("onDownstreamFinish")
result.tryFailure(new InterruptedException("Downstream finished"))
release()
cancel = true
super.onDownstreamFinish(): @nowarn
}
setHandler(out, this)
}
logic -> result.future
}
}
}Gives the following output: |
Beta Was this translation helpful? Give feedback.
-
|
At first, I would suggest you delegate the blocking call to a VirtualThread with Java 21, and how about make use of the Source.unfoldAsync ? For the networking , where you may not been notified, so I would suggest you attach a keepalive to get notified if you don't get the RST. |
Beta Was this translation helpful? Give feedback.
1 reply
-
Could you expand on this? I'm not sure how keep alive helps to get notified? |
Beta Was this translation helpful? Give feedback.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
-
Hello folks,
I've a situation where I can't find what's the idiomatic way to handle it.
Here's my use case:
ResultSet... (blocking API) to HTTP responseTransfer-Encoding: chunked(Play Framework for the record, Pekko HTTP under the hood)ResultSetonDownstreamFinishbeing called until 1 minute(*)(*): with another simpler
Sourcelike aSource.repeat, I can see the stream being stopped shortly after interruption on the client side.The current stream implementation relies on a custom graph that can be resumed as:
I was thinking this is because of the blocking call in
onPull, so I tried to come up with something async but it didn't change anything. I also tried to useunfoldResourcethat is very close to what I'm doing in the custom graph but with no improvement either.My knowledge of Pekko Streams is limited, hence I was wondering if anyone ever had a similar use case and was able to come up with something.
(I can provide a more concrete code sample if needed but didn't want to go too much in the details yet).
Beta Was this translation helpful? Give feedback.
All reactions