Make SnapshotStream public (#6784)

This PR is a subset of changes for #6781

Motivation:

The `SnapshotStream` reactive infrastructure in the `xds` module 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 like `CredentialInjectorFilterFactory` that need
to subscribe to SDS secrets reactively.

Modifications:

- Moved the `SnapshotStream` reactive stream infrastructure from
package-private `com.linecorp.armeria.xds` to a new public
`com.linecorp.armeria.xds.stream` package.
- `SnapshotStream`, `RefCountedStream`, `Subscription` (formerly
`SnapshotStream.Subscription`), and all operators (`MapStream`,
`SwitchMapEagerStream`, `CombineLatest2Stream`, `CombineLatest3Stream`,
`CombineNLatestStream`, `StaticSnapshotStream`, `TriFunction`).
- Made `SnapshotWatcher` public so it can be used as the subscriber type
in the public stream API.
- Updated all internal stream consumers (`ClusterStream`,
`EndpointStream`, `ListenerStream`, `RouteStream`, `SecretStream`,
`DataSourceStream`, `TransportSocketStream`, etc.) to import from the
new `xds.stream` package.
- Simplified `ResourceNodeMeterBinderFactory.ResourceNodeMeterBinder` to
no longer implement `ResourceWatcher`, using plain methods instead.
- Moved stream tests to the `com.linecorp.armeria.xds.stream` package.

Result:

- The `SnapshotStream` API is now public and available for xDS extension
factories to build reactive filter pipelines.
- No behavioral changes — this is a pure refactoring with no changes to
existing functionality.

Author: jrhee17

xds/src/main/java/com/linecorp/armeria/xds/CertificateValidationContextStream.java

@@ -24,6 +24,9 @@
 
 import com.linecorp.armeria.common.util.Exceptions;
 import com.linecorp.armeria.internal.common.util.CertificateUtil;
+import com.linecorp.armeria.xds.stream.RefCountedStream;
+import com.linecorp.armeria.xds.stream.SnapshotStream;
+import com.linecorp.armeria.xds.stream.Subscription;
 
 import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.CertificateValidationContext;
 

xds/src/main/java/com/linecorp/armeria/xds/ClusterRoot.java

@@ -18,7 +18,7 @@
 
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.annotation.UnstableApi;
-import com.linecorp.armeria.xds.SnapshotStream.Subscription;
+import com.linecorp.armeria.xds.stream.Subscription;
 
 import io.envoyproxy.envoy.config.cluster.v3.Cluster;
 

xds/src/main/java/com/linecorp/armeria/xds/ClusterStream.java

@@ -27,6 +27,9 @@
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.xds.client.endpoint.XdsLoadBalancer;
 import com.linecorp.armeria.xds.client.endpoint.XdsLoadBalancerFactory;
+import com.linecorp.armeria.xds.stream.RefCountedStream;
+import com.linecorp.armeria.xds.stream.SnapshotStream;
+import com.linecorp.armeria.xds.stream.Subscription;
 
 import io.envoyproxy.envoy.config.cluster.v3.Cluster;
 import io.envoyproxy.envoy.config.cluster.v3.Cluster.EdsClusterConfig;

xds/src/main/java/com/linecorp/armeria/xds/DataSourceStream.java

@@ -30,6 +30,9 @@
 import com.linecorp.armeria.common.Cancellable;
 import com.linecorp.armeria.common.CommonPools;
 import com.linecorp.armeria.common.file.PathWatcher;
+import com.linecorp.armeria.xds.stream.RefCountedStream;
+import com.linecorp.armeria.xds.stream.SnapshotStream;
+import com.linecorp.armeria.xds.stream.Subscription;
 
 import io.envoyproxy.envoy.config.core.v3.DataSource;
 import io.envoyproxy.envoy.config.core.v3.DataSource.SpecifierCase;

xds/src/main/java/com/linecorp/armeria/xds/EndpointStream.java

@@ -17,6 +17,8 @@
 package com.linecorp.armeria.xds;
 
 import com.linecorp.armeria.common.annotation.Nullable;
+import com.linecorp.armeria.xds.stream.RefCountedStream;
+import com.linecorp.armeria.xds.stream.Subscription;
 
 import io.envoyproxy.envoy.config.core.v3.ConfigSource;
 

xds/src/main/java/com/linecorp/armeria/xds/ListenerManager.java

@@ -26,7 +26,7 @@
 import java.util.Map;
 
 import com.linecorp.armeria.common.util.SafeCloseable;
-import com.linecorp.armeria.xds.SnapshotStream.Subscription;
+import com.linecorp.armeria.xds.stream.Subscription;
 
 import io.envoyproxy.envoy.config.bootstrap.v3.Bootstrap;
 import io.envoyproxy.envoy.config.bootstrap.v3.Bootstrap.StaticResources;

xds/src/main/java/com/linecorp/armeria/xds/ListenerRoot.java

@@ -18,7 +18,7 @@
 
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.annotation.UnstableApi;
-import com.linecorp.armeria.xds.SnapshotStream.Subscription;
+import com.linecorp.armeria.xds.stream.Subscription;
 
 import io.envoyproxy.envoy.config.listener.v3.Listener;
 

xds/src/main/java/com/linecorp/armeria/xds/ListenerStream.java

@@ -19,6 +19,9 @@
 import static com.linecorp.armeria.xds.XdsType.LISTENER;
 
 import com.linecorp.armeria.common.annotation.Nullable;
+import com.linecorp.armeria.xds.stream.RefCountedStream;
+import com.linecorp.armeria.xds.stream.SnapshotStream;
+import com.linecorp.armeria.xds.stream.Subscription;
 
 import io.envoyproxy.envoy.config.core.v3.ConfigSource;
 import io.envoyproxy.envoy.config.route.v3.RouteConfiguration;

xds/src/main/java/com/linecorp/armeria/xds/RawBufferTransportSocketFactory.java

@@ -21,6 +21,7 @@
 import com.google.common.collect.ImmutableList;
 
 import com.linecorp.armeria.common.annotation.Nullable;
+import com.linecorp.armeria.xds.stream.SnapshotStream;
 
 import io.envoyproxy.envoy.config.core.v3.ConfigSource;
 import io.envoyproxy.envoy.config.core.v3.TransportSocket;

xds/src/main/java/com/linecorp/armeria/xds/ResourceNodeAdapter.java

@@ -16,6 +16,9 @@
 
 package com.linecorp.armeria.xds;
 
+import com.linecorp.armeria.xds.stream.RefCountedStream;
+import com.linecorp.armeria.xds.stream.Subscription;
+
 import io.envoyproxy.envoy.config.core.v3.ConfigSource;
 
 final class ResourceNodeAdapter<T extends XdsResource> extends RefCountedStream<T> implements ResourceNode<T> {
@@ -59,13 +62,13 @@ public void onChanged(T update) {
 
     @Override
     public void onError(XdsType type, String resourceName, Throwable t) {
-        resourceNodeMeterBinder.onError(type, resourceName, t);
+        resourceNodeMeterBinder.onError();
         emit(null, XdsResourceException.maybeWrap(type, resourceName, t));
     }
 
     @Override
     public void onResourceDoesNotExist(XdsType type, String resourceName) {
-        resourceNodeMeterBinder.onResourceDoesNotExist(type, resourceName);
+        resourceNodeMeterBinder.onResourceDoesNotExist();
         emit(null, new MissingXdsResourceException(type, resourceName));
     }