pjfanning commented on issue #3107:
URL: https://github.com/apache/pekko/issues/3107#issuecomment-4762547446
Copilot recommends an agnostic API for pekko-streams.
```
package org.apache.pekko.stream.impl
import org.apache.pekko.annotation.InternalStableApi
/**
* INTERNAL API
*
* Manages context propagation for stream elements during execution.
*
* This trait supports two main use cases:
* 1. Context suspension/resumption for elements flowing through stages
* 2. Tracing lifecycle hooks for external observability integration (e.g.,
OpenTelemetry)
*
* The tracing methods provide extension points for external libraries to
implement distributed tracing
* without requiring Pekko to depend on any specific tracing framework.
*
* External integration libraries can:
* - Implement this trait with real tracing logic
* - Capture trace context from incoming elements via `captureTraceContext()`
* - Emit trace spans on element push/pop via `onPushElement()` and
`onPopElement()`
* - Propagate context across async boundaries via `onAsyncBoundary()`
*
* When no tracing is configured, all methods remain no-ops with negligible
overhead.
*/
@InternalStableApi trait ContextPropagation {
/**
* Suspend the current context, typically called when an element cannot be
immediately processed downstream.
*/
def suspendContext(): Unit
/**
* Resume a previously suspended context.
*/
def resumeContext(): Unit
/**
* Get the current context object. Returns an opaque AnyRef that can be
later restored.
*/
def currentContext(): AnyRef
/**
* Resume a specific context that was previously captured via
`currentContext()`.
*/
def resumeContext(context: AnyRef): Unit
/**
* INTERNAL API
*
* Called when an element is about to be pushed through a stage output
port.
*
* External tracing libraries can use this hook to:
* - Start a span for element processing in this stage
* - Record metrics about element flow
* - Link to parent traces
*
* @param stageName The name/type of the stage (e.g., "map", "filter",
"mapAsync")
* @param portId The output port identifier through which the element is
being pushed
*/
def onPushElement(stageName: String, portId: Int): Unit = ()
/**
* INTERNAL API
*
* Called when an element is about to be popped from a stage input port.
*
* External tracing libraries can use this hook to:
* - End/close spans created in `onPushElement`
* - Record completion of element processing
* - Capture any error states
*
* @param stageName The name/type of the stage
* @param portId The input port identifier through which the element is
being popped
*/
def onPopElement(stageName: String, portId: Int): Unit = ()
/**
* INTERNAL API
*
* Called when an element crosses an async boundary (e.g., through an
actor or thread pool).
*
* External tracing libraries can use this hook to:
* - Capture the trace context before crossing the boundary
* - Ensure proper context restoration after async operations
* - Handle context propagation across threads or actor messages
*
* @param stageName The name/type of the stage performing the async
operation
*/
def onAsyncBoundary(stageName: String): Unit = ()
/**
* INTERNAL API
*
* Capture the current trace context for propagation across async
boundaries.
*
* This is called before an element is sent to an async operation (e.g.,
via Future or actor message).
* The returned context should be lightweight and serializable across
threads.
*
* @return An opaque trace context object that can be later restored via
`restoreTraceContext()`,
* or null if no tracing context is available
*/
def captureTraceContext(): AnyRef = null
/**
* INTERNAL API
*
* Restore a trace context that was captured before an async boundary.
*
* This is called after an element returns from an async operation to
restore the original
* trace context and allow proper span linking.
*
* @param context The trace context captured via `captureTraceContext()`
*/
def restoreTraceContext(context: AnyRef): Unit = ()
}
/**
* INTERNAL API
*/
@InternalStableApi object ContextPropagation {
/**
* INTERNAL API
*/
@InternalStableApi def apply(): ContextPropagation = new
ContextPropagationImpl
}
private[pekko] final class ContextPropagationImpl extends ContextPropagation
{
def suspendContext(): Unit = ()
def resumeContext(): Unit = ()
def currentContext(): AnyRef = null
def resumeContext(context: AnyRef): Unit = ()
}
```
This would be wired into pekko-streams existing lib.
We could then create an OpenTelemetry based implementation in a separate lib.
--
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]
