Revert to use processor #2893
Revert to use processor #2893
Conversation
until we figure out what to do. fixes micrometer-metricsgh-2880
related: micrometer-metricsgh-2880 Co-Authored-By: Denis Khitrik <[email protected]>
![sonatype-lift[bot]](https://avatars.githubusercontent.com/in/9843?s=60&v=4){kind=small}
| @@ -143,11 +146,11 @@ private StatsdMeterRegistry(StatsdConfig config, | |||
| ); | |||
|
|
|||
| if (config.enabled()) { | |||
| this.sink = Sinks.many().multicast().directBestEffort(); | |||
| this.sink = processor.sink(); | |||
There was a problem hiding this comment.
deprecation: sink() in FluxProcessor has been deprecated (details)
(at-me in a reply with help or ignore)
| @@ -80,7 +82,8 @@ | |||
| private final HierarchicalNameMapper nameMapper; | |||
| private final Map<Meter.Id, StatsdPollable> pollableMeters = new ConcurrentHashMap<>(); | |||
| private final AtomicBoolean started = new AtomicBoolean(); | |||
| Sinks.Many<String> sink = new NoopManySink(); | |||
| DirectProcessor<String> processor = DirectProcessor.create(); | |||
There was a problem hiding this comment.
deprecation: create() in DirectProcessor has been deprecated (details)
(at-me in a reply with help or ignore)
| @@ -80,7 +82,8 @@ | |||
| private final HierarchicalNameMapper nameMapper; | |||
| private final Map<Meter.Id, StatsdPollable> pollableMeters = new ConcurrentHashMap<>(); | |||
| private final AtomicBoolean started = new AtomicBoolean(); | |||
| Sinks.Many<String> sink = new NoopManySink(); | |||
| DirectProcessor<String> processor = DirectProcessor.create(); | |||
There was a problem hiding this comment.
deprecation: DirectProcessor in reactor.core.publisher has been deprecated (details)
(at-me in a reply with help or ignore)
| @@ -33,12 +33,14 @@ | |||
| import org.reactivestreams.Subscription; | |||
| import reactor.core.Disposable; | |||
| import reactor.core.Disposables; | |||
| import reactor.core.publisher.DirectProcessor; | |||
There was a problem hiding this comment.
deprecation: DirectProcessor in reactor.core.publisher has been deprecated (details)
(at-me in a reply with help or ignore)
| @@ -46,16 +46,16 @@ private BufferingFlux() { | |||
| final AtomicInteger byteSize = new AtomicInteger(); | |||
| final AtomicLong lastTime = new AtomicLong(); | |||
|
|
|||
| final Sinks.Empty<Void> intervalEnd = Sinks.empty(); | |||
| final DirectProcessor<Void> intervalEnd = DirectProcessor.create(); | |||
There was a problem hiding this comment.
deprecation: create() in DirectProcessor has been deprecated (details)
(at-me in a reply with help or ignore)
| @@ -46,16 +46,16 @@ private BufferingFlux() { | |||
| final AtomicInteger byteSize = new AtomicInteger(); | |||
| final AtomicLong lastTime = new AtomicLong(); | |||
|
|
|||
| final Sinks.Empty<Void> intervalEnd = Sinks.empty(); | |||
| final DirectProcessor<Void> intervalEnd = DirectProcessor.create(); | |||
There was a problem hiding this comment.
deprecation: DirectProcessor in reactor.core.publisher has been deprecated (details)
(at-me in a reply with help or ignore)
| @@ -15,8 +15,8 @@ | |||
| */ | |||
| package io.micrometer.statsd.internal; | |||
|
|
|||
| import reactor.core.publisher.DirectProcessor; | |||
There was a problem hiding this comment.
deprecation: DirectProcessor in reactor.core.publisher has been deprecated (details)
(at-me in a reply with help or ignore)
Reverts the change to the new API until we figure out how to use it in a way that doesn't cause the regression describe in #2880. It is less than ideal to use the deprecated API, but this is a shaded dependency so the deprecation does not leak to user code.
Resolves #2880