Rewritten more to graphstage

[He-Pin]

I think we should rewrite more stream operators to graphstages, there are some are not.

[He-Pin]

Analysis: Legacy Stream Components That Can Be Refactored to GraphStage

After analyzing the codebase, here is a structured inventory of components still using the legacy Pump/InputBunch/OutputBunch/TransferPhase infrastructure that could be refactored to modern GraphStage equivalents.

---

Component Analysis Table

#	Component	File	Current Pattern	Legacy Dependencies	Shape	Refactor Priority	Complexity	Fusion Benefit
1	TLSActor	impl/io/TLSActor.scala	Actor + Pump	InputBunch (2), OutputBunch (2), TransferPhase, custom TransferState	BidiShape	High	Very High	Low
2	Unzip (Legacy)	impl/FanOut.scala:328	FanOut + Pump	OutputBunch, BatchingInputBuffer, TransferPhase	FanOutShape2	High	Low	High
3	FanoutProcessorImpl	impl/FanoutProcessor.scala:135	ActorProcessorImpl + Pump	BatchingInputBuffer, FanoutOutputs, TransferPhase	FlowShape	High	Medium	Medium
4	FanoutPublisherSink	impl/Sinks.scala:114	SinkModule → creates FanoutProcessorImpl	Depends on #3	SinkShape	Medium	Medium	Medium
5	FanIn (Base Class)	impl/FanIn.scala	Actor + Pump	InputBunch, SimpleOutputs	Variable	Low	Medium	High
6	FanOut (Base Class)	impl/FanOut.scala:273	Actor + Pump	OutputBunch, BatchingInputBuffer	Variable	Low	Medium	High
7	ActorProcessorImpl	impl/ActorProcessor.scala:272	Actor + Pump	BatchingInputBuffer, SimpleOutputs	FlowShape	Low	Medium	Low
8	TlsModule	impl/io/TlsModule.scala	AtomicModule + custom Island	TlsModuleIslandTag, spawns TLSActor	BidiShape	Done (reference)	High	Low

---

Detailed Breakdown

1. TLSActor (Highest Priority)

The most complex legacy component. Handles bidirectional SSL/TLS encryption/decryption with a sophisticated state machine using TransferPhase patterns (bidirectional, flushingOutbound, awaitingClose, outboundClosed, inboundClosed). Uses custom TransferState implementations like ChoppingBlock for byte chunking.

Key complexity: SSL handshake state management, half-closed connection handling, buffer chopping/put-back logic, and the maxTLSIterations loop guard for unwrapping.

2. Unzip (Legacy) (High Priority, Low Complexity)

Simple fan-out that splits Tuple2/Pair elements into two output streams. Already has a modern GraphStage counterpart in UnzipWith2. The legacy version can be directly replaced.

3. FanoutProcessorImpl (High Priority)

Reactive Streams processor supporting multiple downstream subscribers. Includes SubscriptionTimeout handling. Used by BroadcastHub and similar fan-out patterns.

---

Legacy Infrastructure Dependency Map

Pump (trait)
├── FanIn (Actor) ─── InputBunch ─── BatchingInputBuffer
│   └── (no direct subclasses in use)
│
├── FanOut (Actor) ─── OutputBunch ─── FanoutOutputs
│   └── Unzip ← Component #2
│
├── ActorProcessorImpl (Actor) ─── SimpleOutputs
│   └── FanoutProcessorImpl ← Component #3
│       └── FanoutPublisherSink ← Component #4
│
└── TLSActor (Actor) ─── InputBunch + OutputBunch ← Component #1

---

Advantages of Refactoring to GraphStage

1. Eliminate Legacy Infrastructure: Each refactored component removes dependency on the Pump/InputBunch/OutputBunch/TransferPhase system, which is the old Akka Streams architecture predating GraphStage. This reduces the overall codebase maintenance burden.

2. Materializer Simplification: Components with special Island tags (TlsModule, SourceModule, SinkModule, ProcessorModule) require separate materialization phases. Refactoring to GraphStage means fewer islands, simpler traversal, and less special-case handling in PhasedFusingActorMaterializer.

3. Uniform Architecture: GraphStage is the modern, well-understood abstraction. Having all operators use the same model makes the codebase more consistent, easier to understand, and simpler for new contributors.

4. Better Testability: GraphStage logic (GraphStageLogic) is more testable in isolation compared to actor-based implementations that require actor system setup and message passing.

5. Future-Proofing: The legacy Pump infrastructure is technical debt from early Akka Streams. Removing it incrementally (one component at a time, opt-in via config like the TlsModule approach) allows safe migration without breaking changes.

6. Fusion Opportunities: Some legacy components (like Unzip) could potentially be fused with adjacent stages when converted to GraphStage, reducing the number of async boundaries and improving throughput. Note: TLSActor likely won't benefit from fusion (must run as its own island due to SSLEngine state management), but still benefits from uniform architecture.

7. Reduced Actor Overhead: Each legacy component spawns a dedicated actor. GraphStage components can be fused into a single ActorGraphInterpreter actor, reducing actor count, memory overhead, and context switching.

---

Recommended Order

Phase	Components	Rationale
Phase 1	Unzip (#2)	Lowest complexity, highest confidence gain. Proves the migration pattern.
Phase 2	TLSActor + TlsModule (#1)	Highest impact. Follow the opt-in pattern (system property → config) described in the original proposal.
Phase 3	FanoutProcessorImpl + FanoutPublisherSink (#3, #4)	Medium complexity, enables removal of ProcessorModuleIslandTag.
Phase 4	FanIn/FanOut base classes (#5, #6)	Clean up after #2 is done. Evaluate if they have remaining users.
Phase 5	ActorProcessorImpl (#7)	deepest integration with Reactive Streams adapter, handle last.

[He-Pin]

Based on the current apache/pekko codebase, I re-validated the scope of #2860 and would propose splitting the work into a sequence of small PRs.

Scope correction

Unzip is no longer a public GraphStage migration target: the public DSL scaladsl.Unzip is already GraphStage-based. What remains is an internal legacy impl.FanOut.Unzip, which appears unused and is a good first cleanup candidate.

Proposed PR order

1. PR-1: remove internal legacy impl.FanOut.Unzip
   • small, low-risk cleanup
   • if needed, add directional regression coverage around the current public Unzip behavior
2. PR-2: migrate the Sink.asPublisher(fanout = true) path
   • covers FanoutPublisherSink / FanoutProcessorImpl / required replacement plumbing
   • behavior-sensitive and TCK-sensitive, so this should stay separate from PR-1
3. PR-3: add a GraphStage-backed TLS path behind a safe switch
   • keep the existing TLS actor/island path as fallback initially
4. PR-4: TLS cutover and retirement of old TLS actor/island infrastructure
5. PR-5: final cleanup of now-unused legacy substrate
   • only after grep + compile + MiMa prove there are no remaining users

Notes on dependencies

• Each change should be a separate PR.
• If a PR depends on an earlier one, the PR description should explicitly list the required merge order.
• TLS should not be the first migration PR; it is the highest-risk area because of custom materialization/island behavior, SSLEngine state handling, and compatibility concerns.

Validation focus

• PR-1: GraphUnzipSpec, stream compile, MiMa
• PR-2: FanoutProcessorSpec, TimeoutsSpec, fanout-related TCK, MiMa
• PR-3 / PR-4: TlsSpec, TcpSpec, MaterializerStateSpec, remote TlsTcpSpec, relevant artery TLS coverage, MiMa

I have not started code changes yet in this thread; this comment is only to align the execution plan before taking the first slice.

[He-Pin]

Progress update:

Two review-complete draft PR slices are now open for issue #2860:

1. PR chore(stream): remove internal FanOut Unzip #2873 - chore(stream): remove internal FanOut Unzip

   • Removes the dead internal org.apache.pekko.stream.impl.Unzip implementation from FanOut.scala
   • Adds the matching 2.0.x MiMa excludes
   • Validation: stream/test:compile, stream/mimaReportBinaryIssues, GraphUnzipSpec

2. PR feat(stream): replace fanout publisher runtime with GraphStage bridge #2874 - feat(stream): replace fanout publisher runtime with GraphStage bridge

   • Replaces the actor-backed FanoutProcessorImpl runtime behind Sink.asPublisher(fanout = true) with FanoutPublisherBridgeStage
   • Deletes FanoutProcessor.scala
   • Replaces the old implementation-bound spec with FanoutPublisherBehaviorSpec
   • Adds the matching 2.0.x MiMa excludes
   • Validation: stream/test:compile, stream/mimaReportBinaryIssues, FanoutPublisherBehaviorSpec, TimeoutsSpec, FlowSpec, SinkSpec, SinkTest, FanoutPublisherTest

Both slices were independently cross-reviewed locally (code/RS semantics/safety/test adequacy) before opening the draft PRs.

Next up: continue with the TLS scope from #2860. The current plan is to treat TLS as its own focused migration area and keep the performance-sensitive actor/island constraints explicit during the design and review work.

[He-Pin]

PR-3 Progress Update

PR #2877 has been opened as a draft: feat(stream): add GraphStage-backed TLS path behind feature switch

What is in PR-3

A new TlsGraphStage (pure BidiGraphStage implementation of TLS) is added behind a runtime feature switch:

• Config: pekko.stream.materializer.tls.use-legacy-actor = off
• JVM flag: -Dpekko.stream.materializer.tls.use-legacy-actor=false
• Default: legacy TLSActor path unchanged

Test status

Suite	Status
TlsSpec (legacy path)	✅ 111/111
TlsGraphStageSpec (GraphStage path)	✅ 107 pass, 4 pending*
MaterializerStateSpec	✅ 6/6

*Four "reliably cancel subscriptions" tests are marked pending with detailed Scaladoc. Root cause: async boundary ordering — demand (1 inter-actor hop) consistently wins over failure (2 hops) in the TLS actor mailbox. The async boundary itself is intentional (SSLEngine thread-safety + per-connection actor isolation). See class Scaladoc in TlsGraphStageSpec for full explanation.

PRs so far

• PR-1: chore(stream): remove internal FanOut Unzip #2873 (remove dead internal impl.FanOut.Unzip)
• PR-2: feat(stream): replace fanout publisher runtime with GraphStage bridge #2874 (migrate Sink.asPublisher(fanout=true) to GraphStage)

Coming up

• PR-4: TLS cutover and legacy TLSActor / TlsModule retirement (after PR-3 is reviewed/merged)
• PR-5: Final cleanup of unused FanIn.scala / FanOut.scala substrate

[He-Pin]

 旧基础设施依赖全景分析
  一、依赖关系图谱
  ┌─────────────────────────────────────────────────────────────────┐
  │                    Legacy Pump Infrastructure                     │
  │                                                                 │
  │  Transfer.scala                          FanIn.scala            │
  │  ├─ Pump (trait)                         ├─ InputBunch          │
  │  ├─ TransferState                        ├─ SubInput            │
  │  ├─ TransferPhase                        └─ FanIn (abstract)    │
  │  ├─ Inputs / DefaultInputTransferStates      └── extends Actor + Pump
  │  ├─ Outputs / DefaultOutputTransferStates                       │
  │  ├─ BatchingInputBuffer                                         │
  │  └─ SimpleOutputs                                               │
  │                                                                 │
  │  FanOut.scala                          ActorProcessor.scala     │
  │  ├─ OutputBunch                          ├─ ActorProcessor      │
  │  ├─ FanoutOutputs                        ├─ BatchingInputBuffer │
  │  ├─ FanOut (abstract)                    ├─ SimpleOutputs       │
  │  └─ Unzip                                └─ ActorProcessorImpl  │
  │      └── extends Actor + Pump                └── extends Actor + Pump
  │                                                                 │
  │  FanoutProcessor.scala                                          │
  │  └─ FanoutProcessorImpl                                         │
  │      └── extends ActorProcessorImpl                            │
  └─────────────────────────────────────────────────────────────────┘

  二、谁在用这些基础设施？
   组件                                                 │ 文件                                                 │ 直接依赖                                                 │ 间接依赖
  ──────────────────────────────────────────────────────┼──────────────────────────────────────────────────────┼──────────────────────────────────────────────────────────┼──────────────────────────────────────────────────────────────────
   TLSActor                                             │ TLSActor.scala                                       │ Pump, InputBunch, OutputBunch                            │ TransferState, TransferPhase, BatchingInputBuffer, SimpleOutputs
   FanIn                                                │ FanIn.scala                                          │ Pump, InputBunch, SimpleOutputs                          │ TransferState, BatchingInputBuffer
   FanOut                                               │ FanOut.scala                                         │ Pump, OutputBunch, BatchingInputBuffer                   │ TransferState, SimpleOutputs
   Unzip (impl)                                         │ FanOut.scala:317                                     │ extends FanOut                                           │ 全部间接依赖
   ActorProcessorImpl                                   │ ActorProcessor.scala:262                             │ Pump, BatchingInputBuffer, SimpleOutputs                 │ TransferState, TransferPhase
   FanoutProcessorImpl                                  │ FanoutProcessor.scala:123                            │ extends ActorProcessorImpl                               │ 全部间接依赖
   FanoutPublisherSink                                  │ Sinks.scala:104                                      │ 使用 FanoutProcessorImpl                                 │ 全部间接依赖
   TlsModulePhase                                       │ PhasedFusingActorMaterializer.scala:965              │ ActorPublisher, FanIn.SubInput, FanOut.ExposedPublishers │ 全部间接依赖

  三、关键发现
  3.1 ActorSubscriberMessage 不能删
  ActorSubscriberMessage.OnNext / OnComplete / OnError 被以下**现代**代码使用：
  - stage/GraphStage.scala:1423  ← GraphStage 的内部实现
  - impl/fusing/StreamOfStreams.scala  ← Source.fromAsync / Source.queue 等现代 API

  结论：ActorSubscriberMessage.scala 必须保留，它已经是现代 GraphStage 基础设施的一部分。

  3.2 ActorPublisher 被 TlsModulePhase 直接实例化
  // PhasedFusingActorMaterializer.scala:984
  def factory(id: Int) = new ActorPublisher[Any](tlsActor) { ... }

  这是 materializer 层将 TLSActor 的输出端口包装为 ReactiveStreams Publisher 的桥接代码。

  3.3 Unzip (impl) 是死代码
  公共 API scaladsl.Unzip 继承自 UnzipWith2（GraphStage 实现），从未引用 impl.Unzip。impl.Unzip 只在自己文件内被 Unzip.props 引用，没有任何外部调用者。

  3.4 FanIn (abstract) 和 FanOut (abstract) 没有独立的具体子类
  除了 Unzip extends FanOut 外，没有其它 concrete 子类。FanIn 甚至没有任何子类。

  四、文件级删除依赖顺序
  Phase 0: 新实现可用 (已完成)
    └─ TlsStage (GraphStage) 已通过 opt-in 配置提供

  Phase 1: 移除 TLSActor 及其专属 materializer 代码
    ├─ 删除: TLSActor.scala
    ├─ 删除: TlsModule.scala (data class)
    ├─ 删除: PhasedFusingActorMaterializer.scala 中的 TlsModuleIslandTag / TlsModulePhase
    ├─ 修改: TLS.scala → 始终走 GraphStage 路径
    └─ 可保留: TlsUtils.scala (已移到独立文件，新实现也在用)

  Phase 2: 移除 FanIn / FanOut 及其子类
    ├─ 删除: FanIn.scala (整个文件，无外部使用者)
    ├─ 删除: FanOut.scala 中的 FanOut class + Unzip class + OutputBunch
    │   注意: FanOut.scala 中的 FanoutOutputs 被 FanoutProcessorImpl 使用
    └─ 保留: FanoutOutputs 类（Phase 3 处理）

  Phase 3: 移除 ActorProcessorImpl / FanoutProcessorImpl
    ├─ 删除: FanoutProcessor.scala
    ├─ 修改: Sinks.scala 中的 FanoutPublisherSink → 用 GraphStage 实现替代
    ├─ 删除: ActorProcessor.scala 中的 ActorProcessorImpl, BatchingInputBuffer, SimpleOutputs
    └─ 保留: ActorProcessor class (ReactiveStreams Processor 桥接，仍有用途)

  Phase 4: 清理核心 Pump 基础设施
    ├─ 删除: Transfer.scala (Pump, TransferState, TransferPhase, Inputs, Outputs)
    ├─ 删除: ActorSubscriberMessage.scala → **不能删！** GraphStage 仍在使用
    └─ 验证: 无残留引用

  Phase 5: 清理 ActorPublisher
    └─ 评估是否仍有使用者（FanoutPublisherSink 移除后可能可删）

  五、风险提示
   风险项                                                                                   │ 说明
  ──────────────────────────────────────────────────────────────────────────────────────────┼────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
   FanoutPublisherSink                                                                      │ 用于 Sink.publisher()，必须先用 GraphStage 实现替代才能删除依赖链
   ActorProcessor                                                                           │ 作为 ReactiveStreams Processor[I,O] 的桥接层，FanoutPublisherSink 和潜在用户代码仍在使用
   TlsUtils.cloneParameters                                                                 │ 在 TLSActor 中使用，新 TlsStage 中未用到，可随 TLSActor 一起删除
   SubReceive                                                                               │ Transfer.scala 中的可变 receive 切换器，被 InputBunch/OutputBunch/BatchingInputBuffer/SimpleOutputs 全部使用。删除这些后 SubReceive 也无人使用
   MIMA 兼容性                                                                              │ 所有类型均为 @InternalApi，删除不会破坏公共 API 二进制兼容性，但 mima-filters 需更新

  六、建议执行路径
  最安全的路径是：

  1. 先切掉 TLSActor（PR #32914 已经铺好了路）— 将 opt-in 改为默认
  2. 评估 FanoutPublisherSink 是否有用户真正在使用 — 如果有，用 GraphStage 重写
  3. 批量删除 — Phase 1 完成后，Phase 2/3/4 可以合并为一个大 PR，因为中间状态没有用户价值
  4. 保留 ActorSubscriberMessage 和 ActorProcessor — 这两个已是现代基础设施的组成部分
  5. ```

[pjfanning]

TLSActor is used in pekko-http. So we need to retain that class in its current form for a period of a few weeks. So we could add its replacement but keep the old code for a few weeks to keep our builds working. We can update pekko-http but let's not break it and force a panicked update.

[He-Pin]

I think.we may need keep it in 2.20.0 and remove it in 2.1.0？