-
Notifications
You must be signed in to change notification settings - Fork 41
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #3 from netifi/fix-fuseable-metrics-issue
fix issue with fuseable metrics subscriber
- Loading branch information
Showing
13 changed files
with
426 additions
and
33 deletions.
There are no files selected for viewing
This file contains 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
This file contains 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
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,2 +1,2 @@ | ||
| group=io.rsocket.rpc | ||
| version=0.2.0 | ||
| version=0.2.1 |
This file contains 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
This file contains 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
This file contains 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
94 changes: 94 additions & 0 deletions
94
rsocket-rpc-core/src/main/java/io/rsocket/rpc/metrics/MetricsConditionalSubscriber.java
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| package io.rsocket.rpc.metrics; | ||
|
|
||
| import io.micrometer.core.instrument.Counter; | ||
| import io.micrometer.core.instrument.Timer; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import org.reactivestreams.Subscription; | ||
| import reactor.core.Fuseable.ConditionalSubscriber; | ||
| import reactor.core.publisher.Operators; | ||
| import reactor.util.context.Context; | ||
|
|
||
| public class MetricsConditionalSubscriber<T> extends AtomicBoolean | ||
| implements Subscription, ConditionalSubscriber<T> { | ||
| private final ConditionalSubscriber<? super T> actual; | ||
| private final Counter next, complete, error, cancelled; | ||
| private final Timer timer; | ||
|
|
||
| private Subscription s; | ||
| private long start; | ||
|
|
||
| MetricsConditionalSubscriber( | ||
| ConditionalSubscriber<? super T> actual, | ||
| Counter next, | ||
| Counter complete, | ||
| Counter error, | ||
| Counter cancelled, | ||
| Timer timer) { | ||
| this.actual = actual; | ||
| this.next = next; | ||
| this.complete = complete; | ||
| this.error = error; | ||
| this.cancelled = cancelled; | ||
| this.timer = timer; | ||
| } | ||
|
|
||
| @Override | ||
| public void onSubscribe(Subscription s) { | ||
| if (Operators.validate(this.s, s)) { | ||
| this.s = s; | ||
| this.start = System.nanoTime(); | ||
|
|
||
| actual.onSubscribe(this); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onNext(T t) { | ||
| next.increment(); | ||
| actual.onNext(t); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean tryOnNext(T t) { | ||
| next.increment(); | ||
| return actual.tryOnNext(t); | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable t) { | ||
| if (compareAndSet(false, true)) { | ||
| error.increment(); | ||
| timer.record(System.nanoTime() - start, TimeUnit.NANOSECONDS); | ||
| } | ||
| actual.onError(t); | ||
| } | ||
|
|
||
| @Override | ||
| public void onComplete() { | ||
| if (compareAndSet(false, true)) { | ||
| complete.increment(); | ||
| timer.record(System.nanoTime() - start, TimeUnit.NANOSECONDS); | ||
| } | ||
| actual.onComplete(); | ||
| } | ||
|
|
||
| @Override | ||
| public void request(long n) { | ||
| s.request(n); | ||
| } | ||
|
|
||
| @Override | ||
| public void cancel() { | ||
| if (compareAndSet(false, true)) { | ||
| cancelled.increment(); | ||
| timer.record(System.nanoTime() - start, TimeUnit.NANOSECONDS); | ||
| } | ||
| s.cancel(); | ||
| } | ||
|
|
||
| @Override | ||
| public Context currentContext() { | ||
| return actual.currentContext(); | ||
| } | ||
| } |
144 changes: 144 additions & 0 deletions
144
...t-rpc-core/src/main/java/io/rsocket/rpc/metrics/MetricsFuseableConditionalSubscriber.java
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,144 @@ | ||
| package io.rsocket.rpc.metrics; | ||
|
|
||
| import static reactor.core.Fuseable.ASYNC; | ||
|
|
||
| import io.micrometer.core.instrument.Counter; | ||
| import io.micrometer.core.instrument.Timer; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import org.reactivestreams.Subscription; | ||
| import reactor.core.Fuseable; | ||
| import reactor.core.Fuseable.ConditionalSubscriber; | ||
| import reactor.core.Fuseable.QueueSubscription; | ||
| import reactor.core.publisher.Operators; | ||
| import reactor.util.annotation.Nullable; | ||
| import reactor.util.context.Context; | ||
|
|
||
| public class MetricsFuseableConditionalSubscriber<T> extends AtomicBoolean | ||
| implements QueueSubscription<T>, ConditionalSubscriber<T> { | ||
| private final ConditionalSubscriber<? super T> actual; | ||
| private final Counter next, complete, error, cancelled; | ||
| private final Timer timer; | ||
|
|
||
| private QueueSubscription<T> s; | ||
| private int sourceMode; | ||
|
|
||
| private long start; | ||
|
|
||
| MetricsFuseableConditionalSubscriber( | ||
| ConditionalSubscriber<? super T> actual, | ||
| Counter next, | ||
| Counter complete, | ||
| Counter error, | ||
| Counter cancelled, | ||
| Timer timer) { | ||
| this.actual = actual; | ||
| this.next = next; | ||
| this.complete = complete; | ||
| this.error = error; | ||
| this.cancelled = cancelled; | ||
| this.timer = timer; | ||
| } | ||
|
|
||
| @Override | ||
| @SuppressWarnings("unchecked") | ||
| public void onSubscribe(Subscription s) { | ||
| if (Operators.validate(this.s, s)) { | ||
| this.s = (QueueSubscription<T>) s; | ||
| this.start = System.nanoTime(); | ||
|
|
||
| actual.onSubscribe(this); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onNext(T t) { | ||
| if (sourceMode == ASYNC) { | ||
| actual.onNext(null); | ||
| } else { | ||
| next.increment(); | ||
| actual.onNext(t); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public boolean tryOnNext(T t) { | ||
| next.increment(); | ||
| return actual.tryOnNext(t); | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable t) { | ||
| if (compareAndSet(false, true)) { | ||
| error.increment(); | ||
| timer.record(System.nanoTime() - start, TimeUnit.NANOSECONDS); | ||
| } | ||
| actual.onError(t); | ||
| } | ||
|
|
||
| @Override | ||
| public void onComplete() { | ||
| if (compareAndSet(false, true)) { | ||
| complete.increment(); | ||
| timer.record(System.nanoTime() - start, TimeUnit.NANOSECONDS); | ||
| } | ||
| actual.onComplete(); | ||
| } | ||
|
|
||
| @Override | ||
| public void request(long n) { | ||
| s.request(n); | ||
| } | ||
|
|
||
| @Override | ||
| public void cancel() { | ||
| if (compareAndSet(false, true)) { | ||
| cancelled.increment(); | ||
| timer.record(System.nanoTime() - start, TimeUnit.NANOSECONDS); | ||
| } | ||
| s.cancel(); | ||
| } | ||
|
|
||
| @Override | ||
| public Context currentContext() { | ||
| return actual.currentContext(); | ||
| } | ||
|
|
||
| @Override | ||
| public int requestFusion(int requestedMode) { | ||
| int m; | ||
| if ((requestedMode & Fuseable.THREAD_BARRIER) != 0) { | ||
| return Fuseable.NONE; | ||
| } else { | ||
| m = s.requestFusion(requestedMode); | ||
| } | ||
| sourceMode = m; | ||
| return m; | ||
| } | ||
|
|
||
| @Override | ||
| @Nullable | ||
| public T poll() { | ||
| T v = s.poll(); | ||
| if (v != null) { | ||
| next.increment(); | ||
| return v; | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isEmpty() { | ||
| return s.isEmpty(); | ||
| } | ||
|
|
||
| @Override | ||
| public void clear() { | ||
| s.clear(); | ||
| } | ||
|
|
||
| @Override | ||
| public int size() { | ||
| return s.size(); | ||
| } | ||
| } |
Oops, something went wrong.