Make SnapshotStream public#6784
Merged
Merged
Conversation
Contributor
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughExtracts XDS stream primitives into com.linecorp.armeria.xds.stream (new interfaces and package annotations), introduces public SnapshotStream APIs and exposes RefCountedStream, moves concrete stream implementations and tests into the new package, updates consumer imports/signatures, and refactors resource meter-binder callbacks. ChangesXDS Stream Infrastructure Refactoring
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Warning Review ran into problems🔥 ProblemsGit: Failed to clone repository. Please run the 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. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
xds/src/main/java/com/linecorp/armeria/xds/stream/RefCountedStream.java (1)
37-50:⚠️ Potential issue | 🟠 Major | ⚡ Quick winFail fast on null watcher in public subscription API.
Line 48 should explicitly reject
nullwatcher; currently null can flow into callbacks and fail later with a less clear NPE.Suggested fix
+import static java.util.Objects.requireNonNull; + @@ `@Override` public final Subscription subscribe(SnapshotWatcher<? super T> watcher) { + requireNonNull(watcher, "watcher"); if (latestValue != null) { watcher.onUpdate(latestValue, null); }As per coding guidelines, "for any user-facing public methods, add explicit null checks; use Objects.requireNonNull(param, "name") for required params".
🤖 Prompt for 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. In `@xds/src/main/java/com/linecorp/armeria/xds/stream/RefCountedStream.java` around lines 37 - 50, The public subscribe(SnapshotWatcher<? super T> watcher) method in RefCountedStream should immediately validate its watcher parameter to fail fast on nulls; add Objects.requireNonNull(watcher, "watcher") at the top of RefCountedStream.subscribe and ensure java.util.Objects is imported so a clear NPE is thrown immediately rather than letting null flow into watcher.onUpdate or later callbacks.xds/src/main/java/com/linecorp/armeria/xds/SnapshotWatcher.java (1)
2-2:⚠️ Potential issue | 🟠 Major | ⚡ Quick winUpdate this modified file to the LY copyright header.
Line 2 still uses
LINE Corporation; this file was modified and should use the LY header format.As per coding guidelines, "every modified source file must start with the LY copyright header".
🤖 Prompt for 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. In `@xds/src/main/java/com/linecorp/armeria/xds/SnapshotWatcher.java` at line 2, Replace the existing copyright header in the modified SnapshotWatcher.java with the standard LY header format: update the top-of-file header (the one above the SnapshotWatcher class definition) so it uses the LY copyright text required by the project guidelines, ensuring the new header replaces the current "LINE Corporation" line and matches the project's canonical LY header exactly.
🤖 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 67-167: All public/default API methods should validate their
parameters and fail fast: add Objects.requireNonNull(...) checks at the start of
map (check mapper), switchMapEager (check mapper), combineNLatest (check streams
and iterate to requireNonNull each element), combineLatest(a,b,combiner) (check
a, b, combiner), combineLatest(a,b,c,combiner) (check a, b, c, combiner), just
(check value), and error (check error); use Objects.requireNonNull(param,
"paramName") and throw early rather than relying on downstream NPEs.
---
Outside diff comments:
In `@xds/src/main/java/com/linecorp/armeria/xds/SnapshotWatcher.java`:
- Line 2: Replace the existing copyright header in the modified
SnapshotWatcher.java with the standard LY header format: update the top-of-file
header (the one above the SnapshotWatcher class definition) so it uses the LY
copyright text required by the project guidelines, ensuring the new header
replaces the current "LINE Corporation" line and matches the project's canonical
LY header exactly.
In `@xds/src/main/java/com/linecorp/armeria/xds/stream/RefCountedStream.java`:
- Around line 37-50: The public subscribe(SnapshotWatcher<? super T> watcher)
method in RefCountedStream should immediately validate its watcher parameter to
fail fast on nulls; add Objects.requireNonNull(watcher, "watcher") at the top of
RefCountedStream.subscribe and ensure java.util.Objects is imported so a clear
NPE is thrown immediately rather than letting null flow into watcher.onUpdate or
later callbacks.
🪄 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: 3b37515e-e436-414c-a75c-0ac8bea5cc1c
📒 Files selected for processing (34)
xds/src/main/java/com/linecorp/armeria/xds/CertificateValidationContextStream.javaxds/src/main/java/com/linecorp/armeria/xds/ClusterRoot.javaxds/src/main/java/com/linecorp/armeria/xds/ClusterStream.javaxds/src/main/java/com/linecorp/armeria/xds/DataSourceStream.javaxds/src/main/java/com/linecorp/armeria/xds/EndpointStream.javaxds/src/main/java/com/linecorp/armeria/xds/ListenerManager.javaxds/src/main/java/com/linecorp/armeria/xds/ListenerRoot.javaxds/src/main/java/com/linecorp/armeria/xds/ListenerStream.javaxds/src/main/java/com/linecorp/armeria/xds/RawBufferTransportSocketFactory.javaxds/src/main/java/com/linecorp/armeria/xds/ResourceNodeAdapter.javaxds/src/main/java/com/linecorp/armeria/xds/ResourceNodeMeterBinderFactory.javaxds/src/main/java/com/linecorp/armeria/xds/RouteStream.javaxds/src/main/java/com/linecorp/armeria/xds/SecretStream.javaxds/src/main/java/com/linecorp/armeria/xds/SnapshotStream.javaxds/src/main/java/com/linecorp/armeria/xds/SnapshotWatcher.javaxds/src/main/java/com/linecorp/armeria/xds/TlsCertificateStream.javaxds/src/main/java/com/linecorp/armeria/xds/TransportSocketFactory.javaxds/src/main/java/com/linecorp/armeria/xds/TransportSocketStream.javaxds/src/main/java/com/linecorp/armeria/xds/UpstreamTlsTransportSocketFactory.javaxds/src/main/java/com/linecorp/armeria/xds/XdsClusterManager.javaxds/src/main/java/com/linecorp/armeria/xds/stream/CombineLatest2Stream.javaxds/src/main/java/com/linecorp/armeria/xds/stream/CombineLatest3Stream.javaxds/src/main/java/com/linecorp/armeria/xds/stream/CombineNLatestStream.javaxds/src/main/java/com/linecorp/armeria/xds/stream/MapStream.javaxds/src/main/java/com/linecorp/armeria/xds/stream/RefCountedStream.javaxds/src/main/java/com/linecorp/armeria/xds/stream/SnapshotStream.javaxds/src/main/java/com/linecorp/armeria/xds/stream/StaticSnapshotStream.javaxds/src/main/java/com/linecorp/armeria/xds/stream/Subscription.javaxds/src/main/java/com/linecorp/armeria/xds/stream/SwitchMapEagerStream.javaxds/src/main/java/com/linecorp/armeria/xds/stream/TriFunction.javaxds/src/main/java/com/linecorp/armeria/xds/stream/package-info.javaxds/src/test/java/com/linecorp/armeria/xds/stream/CombineNLatestStreamTest.javaxds/src/test/java/com/linecorp/armeria/xds/stream/RefCountedStreamTest.javaxds/src/test/java/com/linecorp/armeria/xds/stream/StreamSwitchMapEagerTest.java
💤 Files with no reviewable changes (1)
- xds/src/main/java/com/linecorp/armeria/xds/SnapshotStream.java
Contributor
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #6784 +/- ##
============================================
- Coverage 74.46% 0 -74.47%
============================================
Files 1963 0 -1963
Lines 82437 0 -82437
Branches 10764 0 -10764
============================================
- Hits 61385 0 -61385
+ Misses 15918 0 -15918
+ Partials 5134 0 -5134 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
jrhee17
added a commit
that referenced
this pull request
Jun 4, 2026
This PR should be reviewed after #6784 This PR is a subset of #6781 Motivation: xDS extension factories (e.g. `HttpFilterFactory`) currently receive only an `XdsResourceValidator`, which is insufficient for filters that depend on external xDS resources like SDS secrets. The `CredentialInjectorFilterFactory` (`envoy.filters.http.credential_injector`) needs to reactively subscribe to generic secrets via SDS and inject credentials into outgoing requests, requiring access to the event loop, metrics, and secret streams at construction time. Modifications: - Added `FactoryContext` interface providing runtime infrastructure to extension factories (`eventLoop()`, `meterRegistry()`, `meterIdPrefix()`, `validator()`, `genericSecretStream()`). - Changed `HttpFilterFactory.create()` to accept `FactoryContext` instead of `XdsResourceValidator`. - Added `HttpFilterFactory.createStream()` that returns a `SnapshotStream<XdsHttpFilter>` for reactive filter lifecycle management. - Added `XdsHttpFilter.NOOP` static constant for no-op filter instances. - Added `GenericSecretSnapshot` and `GenericSecretStream` for resolving generic secrets via SDS. - Made the filter pipeline reactive: `FilterUtil` now returns `SnapshotStream<ClientPreprocessors>` / `SnapshotStream<ClientDecoration>`, and `RouteStream` composes cluster, downstream, and upstream filter streams via `combineLatest`. - Simplified `RouteEntry` to accept pre-built filter configs instead of building them internally. - Updated `ListenerResourceParser` / `ListenerXdsResource` to parse and carry the `Router` HTTP filter for upstream filter extraction. - Made `SubscriptionContext` extend `FactoryContext` for internal use. - Added `CredentialInjectorFilterFactory` implementing `envoy.filters.http.credential_injector`, registered via service loader. - Annotated `TypedExtensionConfig`, `CredentialInjector`, `Generic`, and `Secret.generic_secret` proto fields as supported. - Updated `RouterFilterFactory` and `IstioFilterFactories` to use `FactoryContext`. Result: - xDS extension factories can now build reactive filter pipelines that depend on external resources like SDS secrets. - The `CredentialInjectorFilterFactory` injects credentials from SDS-backed generic secrets into outgoing HTTP requests, supporting the `envoy.filters.http.credential_injector` filter type.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR is a subset of changes for #6781
Motivation:
The
SnapshotStreamreactive infrastructure in thexdsmodule is currently package-private, which prevents xDS extension factories (e.g.HttpFilterFactory) from using reactive streams to model filters that depend on external resources. Making this API public is a prerequisite for upcoming features likeCredentialInjectorFilterFactorythat need to subscribe to SDS secrets reactively.Modifications:
SnapshotStreamreactive stream infrastructure from package-privatecom.linecorp.armeria.xdsto a new publiccom.linecorp.armeria.xds.streampackage.SnapshotStream,RefCountedStream,Subscription(formerlySnapshotStream.Subscription), and all operators (MapStream,SwitchMapEagerStream,CombineLatest2Stream,CombineLatest3Stream,CombineNLatestStream,StaticSnapshotStream,TriFunction).SnapshotWatcherpublic so it can be used as the subscriber type in the public stream API.ClusterStream,EndpointStream,ListenerStream,RouteStream,SecretStream,DataSourceStream,TransportSocketStream, etc.) to import from the newxds.streampackage.ResourceNodeMeterBinderFactory.ResourceNodeMeterBinderto no longer implementResourceWatcher, using plain methods instead.com.linecorp.armeria.xds.streampackage.Result:
SnapshotStreamAPI is now public and available for xDS extension factories to build reactive filter pipelines.