Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Reactive: add Sentinel Reactor adapter module #545

Merged
merged 5 commits into from
Mar 8, 2019

Conversation

sczyh30
Copy link
Member

@sczyh30 sczyh30 commented Mar 5, 2019

Describe what this PR does / why we need it

Add Sentinel Reactor adapter module and implement reactive transformer.

Does this pull request fix one issue?

Resolves #544

Describe how you did it

API Design:

someService.doSomething()
   .transform(new SentinelReactorTransformer<>(resourceName))
   .subscribe();

Internal implementation:

Here I've implemented a SentinelReactorTransformer that will augment the original stream with the encapsulated Sentinel operator during the assembly stage. There are two kinds of operators:

  • MonoSentinelOperator implements MonoOperator
  • FluxSentinelOperator implements FluxOperator

Both will wrap the actual subscriber with the SentinelReactorSubscriber during subscribe. SentinelReactorSubscriber is derived from the Reactor BaseSubscriber (due to the fact that the original BaseSubscriber declares all event callbacks as final methods, I have to copy the original code and modify it with so that the methods are inheritable).

Entry logic of Sentinel is wrapped in each hook function (hookOnSubscribe, hookOnNext, hookOnError, hookOnComplete). The stage:

  • The hookOnSubscribe will be called during the subscribe stage. Here Sentinel will execute SphU.asyncEntry(xxx) and propagate the event. If the request is blocked, Sentinel will cancel the upstream and propagate the corresponding BlockException with onError callback.

  • The hookOnNext will be called when receiving data. If the upstream has been canceled, Sentinel will just try to exit the entry.

    • For unary publishers (Mono), the onNext should be called at most once, followed by the terminal signal. There may be such kind of event stream: onSubscribe() -> onNext() -> cancel() -> onComplete(). In that case, the onComplete event would be dropped. So for Mono we need to tryCompleteEntry in both hookOnNext and hookOnComplete.
    • For Flux, we just call onNext to propagate the event and data.
  • In hookOnComplete we just tryCompleteEntry and call actual.onComplete() to propagate the event.

  • In hookOnError, if the error is biz-exception (not a BlockException), we'll record the exception within the entry. Then we'll tryCompleteEntry and propogate the error.

  • How to carry Context of Sentinel to build the invocation chain: via Reactor Context (override the actualContext)

  • The response time: from onSubscribe to terminal state

Describe how to verify it

Run the test cases.

Special notes for reviews

The TCK did not work well so here I removed tests with TCK. Please check whether the SentinelReactorSubscriber obeys the Reactive Streams specification.

- Initial work for common Mono/Flux

Signed-off-by: Eric Zhao <[email protected]>
- Add a `InheritableBaseSubscriber` that derives from the original BaseSubscriber of reactor-core

Signed-off-by: Eric Zhao <[email protected]>
Signed-off-by: Eric Zhao <[email protected]>
@sczyh30 sczyh30 added the to-review To review label Mar 5, 2019
Copy link
Contributor

@CarpenterLee CarpenterLee left a comment

Choose a reason for hiding this comment

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

LGTM

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

[Feature] Add Project Reactor adapter support
2 participants