He-Pin opened a new pull request, #3179:
URL: https://github.com/apache/pekko/pull/3179

   ### Motivation
   
   Per the [stream error handling 
docs](https://pekko.apache.org/docs/pekko/current/stream/stream-error.html#supervision-strategies),
 operators that apply user-provided functions should consult the configured 
`SupervisionStrategy`. The `doOnFirst` operator runs a user side-effect 
function `f` on the **first** element, but it was called without a try-catch, 
so any exception failed the stream **unconditionally** — even under 
`Supervision.Resume`/`Restart`.
   
   This is part of the meta-issue #3110 (add supervisor strategy support to 
stream operators that accept user functions). One operator per PR.
   
   ### Modification
   
   - Wrap `f(elem)` in `DoOnFirst` with a `try`/`catch` (`NonFatal`) that 
consults the `SupervisionStrategy` decider:
     - **Stop** → fail the stage (unchanged fail-fast behavior)
     - **Resume** → drop the failing first element and switch to pass-through; 
`f` is **not** retried (the one-shot is consumed)
     - **Restart** → re-arm first-element detection so the **next** element is 
treated as the first (`f` is retried on it)
   - The first-element `InHandler` was extracted into a named `val 
firstHandler` so it can be re-armed on Restart. On success, the switch to 
pass-through now happens after `f` (behaviorally identical on the happy path — 
`push` does not synchronously re-enter `onPush`).
   - `decider` is a `lazy val` → zero overhead on the happy path.
   - Fix a fully-qualified type reference in the `createLogic` signature to use 
the project's two-line import style.
   - Document supervision adherence (including the Resume/Restart semantics) in 
the Scala/Java DSL scaladoc and the operator reference page.
   
   ### Result
   
   `doOnFirst` now adheres to the `SupervisionStrategy` attribute with clear, 
distinguishable Resume (drop first, no retry) and Restart (next becomes first) 
semantics.
   
   ### Tests
   
   - `sbt "stream-tests/Test/testOnly 
org.apache.pekko.stream.scaladsl.FlowDoOnFirstSpec"` — **6/6 passed**, 
including 4 new directional tests:
     - fails on `f` throw with `Supervision.Stop`
     - fails with no supervision attribute set (default Stop) — back-compat 
guard
     - `Resume` drops the first element, does not retry `f`, passes the rest 
through
     - `Restart` re-arms so `f` is retried on each subsequent "first" element
   - `sbt "stream/mimaReportBinaryIssues"` — clean (binary compatible)
   
   ### References
   
   Refs #3110
   
   This is a clean-room implementation written directly for Apache Pekko.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to