Skip to content

Expose CachingStream as a public API - #6831

Merged
jrhee17 merged 1 commit into
line:mainfrom
jrhee17:feat/xds-caching-stream
Jun 29, 2026
Merged

Expose CachingStream as a public API#6831
jrhee17 merged 1 commit into
line:mainfrom
jrhee17:feat/xds-caching-stream

Conversation

@jrhee17

@jrhee17 jrhee17 commented Jun 24, 2026

Copy link
Copy Markdown
Contributor

Motivation:

CachingStream provides a useful pattern for deduplicating SnapshotStream subscriptions by key using reference counting. Currently it is package-private in com.linecorp.armeria.xds, making it unavailable to external modules that build on the SnapshotStream API (e.g. Central Dogma had to fork it).

Modifications:

  • Added SnapshotStream.caching(Function) static factory method that returns a Function<K, SnapshotStream<T>> for keyed, ref-counted stream deduplication
  • Moved CachingStream to com.linecorp.armeria.xds.stream as a package-private implementation detail
  • Updated RouteStream to use the new SnapshotStream.caching() API
  • Deleted the old com.linecorp.armeria.xds.CachingStream

Result:

  • Users can now deduplicate SnapshotStream subscriptions by key via SnapshotStream.caching(factory) without forking internal classes

@jrhee17 jrhee17 added this to the 1.41.0 milestone Jun 24, 2026
@coderabbitai

coderabbitai Bot commented Jun 24, 2026

Copy link
Copy Markdown
Contributor

Review Change Stack

📝 Walkthrough

Walkthrough

Introduces a SnapshotStream.caching(...) static factory method that wraps a CachingStream as a Function<K, SnapshotStream<T>>. CachingStream is moved to the stream package with widened generic bounds. FilterCaches in RouteStream is updated to store and invoke these functions instead of directly calling subscribe.

Changes

CachingStream API and RouteStream migration

Layer / File(s) Summary
CachingStream bounds widening and SnapshotStream.caching factory
xds/.../stream/CachingStream.java, xds/.../stream/SnapshotStream.java
CachingStream moves to the stream package, widens its factory field to Function<? super K, ? extends SnapshotStream<T>>, and switches computeIfAbsent to a constructor reference. SnapshotStream gains a static caching(factory) method that null-checks the factory, constructs a CachingStream, and returns cachingStream::subscribe as a Function<K, SnapshotStream<T>> with Javadoc describing per-key reference-counted deduplication.
FilterCaches migrated to function-based API
xds/.../xds/RouteStream.java
FilterCaches fields change from CachingStream instances to Function<Map<String, Any>, SnapshotStream<...>> values initialized via SnapshotStream.caching(...). RouteEntryStream.onStart calls filterCaches.<field>.apply(filterConfigs) instead of subscribe(filterConfigs). Adds java.util.function.Function import.

Estimated code review effort

🎯 2 (Simple) | ⏱️ ~10 minutes

Poem

🐇 A function hops in where a stream once lived,
caching(factory) — the shortcut is givened!
Per-key deduplication, reference-counted tight,
apply replaces subscribe with delight.
The bounds grow wider, the lambda grows terse —
A cleaner stream API, for better or verse! 🌿

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name Status Explanation Resolution
Docstring Coverage ⚠️ Warning Docstring coverage is 28.57% which is insufficient. The required threshold is 80.00%. Write docstrings for the functions missing them to satisfy the coverage threshold.
✅ Passed checks (4 passed)
Check name Status Explanation
Title check ✅ Passed The title is related to the new caching API, though it overstates exposing the class itself.
Description check ✅ Passed The description accurately matches the added API, the stream move, and the RouteStream update.
Linked Issues check ✅ Passed Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check ✅ Passed Check skipped because no linked issues were found for this pull request.

✏️ Tip: You can configure your own custom pre-merge checks in the settings.

✨ Finishing Touches
🧪 Generate unit tests (beta)
  • Create PR with unit tests

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

Comment @coderabbitai help to get the list of available commands.

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@xds/src/main/java/com/linecorp/armeria/xds/stream/SnapshotStream.java`:
- Around line 216-241: The caching API in SnapshotStream.caching is stateful and
backed by CachingStream’s HashMap and ref-counting, so it needs an explicit
single-threaded contract. Update the Javadoc on caching (and, if applicable,
CachingStream.subscribe/close) to state that all apply/subscribe/close calls
must be serialized on the same event-loop thread; alternatively, make
CachingStream thread-safe if that contract cannot be guaranteed. Use the
existing SnapshotStream and CachingStream symbols to keep the documentation
aligned with the public API.
🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

  • Push a commit to this branch (recommended)
  • Create a new PR with the fixes

ℹ️ Review info
⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: b183e00c-1f57-48ec-9629-02fa306e9574

📥 Commits

Reviewing files that changed from the base of the PR and between 5567e0d and a6c10db.

📒 Files selected for processing (3)
  • xds/src/main/java/com/linecorp/armeria/xds/RouteStream.java
  • xds/src/main/java/com/linecorp/armeria/xds/stream/CachingStream.java
  • xds/src/main/java/com/linecorp/armeria/xds/stream/SnapshotStream.java

@codecov

codecov Bot commented Jun 24, 2026

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 75.07%. Comparing base (8150425) to head (a6c10db).
⚠️ Report is 516 commits behind head on main.

Additional details and impacted files
@@             Coverage Diff              @@
##               main    #6831      +/-   ##
============================================
+ Coverage     74.46%   75.07%   +0.61%     
- Complexity    22234    25162    +2928     
============================================
  Files          1963     2239     +276     
  Lines         82437    93431   +10994     
  Branches      10764    12187    +1423     
============================================
+ Hits          61385    70147    +8762     
- Misses        15918    17462    +1544     
- Partials       5134     5822     +688     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

@jrhee17
jrhee17 marked this pull request as ready for review June 24, 2026 08:54
@mergify

mergify Bot commented Jun 24, 2026

Copy link
Copy Markdown

Tick the box to add this pull request to the merge queue (same as @mergifyio queue).

  • Queue this pull request

@ikhoon ikhoon left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍 👍

@minwoox minwoox left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍 👍 👍

@jrhee17
jrhee17 merged commit 8de1a03 into line:main Jun 29, 2026
17 of 19 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants