本文整理了Java中reactor.core.publisher.Hooks.onEachOperator()
方法的一些代码示例,展示了Hooks.onEachOperator()
的具体用法。这些代码示例主要来源于Github
/Stackoverflow
/Maven
等平台,是从一些精选项目中提取出来的代码,具有较强的参考意义,能在一定程度帮忙到你。Hooks.onEachOperator()
方法的具体详情如下:
包路径:reactor.core.publisher.Hooks
类名称:Hooks
方法名:onEachOperator
[英]Add or replace a named Publisher operator interceptor for each operator created ( Flux or Mono). The passed function is applied to the original operator Publisher and can return a different Publisher, on the condition that it generically maintains the same data type as the original. Use of the Flux/ Mono APIs is discouraged as it will recursively call this hook, leading to StackOverflowError.
Note that sub-hooks are cumulative. Invoking this method twice with the same key will replace the old sub-hook with that name, but keep the execution order (eg. A-h1, B-h2, A-h3 will keep A-B execution order, leading to hooks h3 then h2 being executed). Removing a particular key using #resetOnEachOperator(String) then adding it back will result in the execution order changing (the later sub-hook being executed last). Can be fully reset via #resetOnEachOperator().
This pointcut function cannot make use of Flux, Mono or ParallelFlux APIs as it would lead to a recursive call to the hook: the operator calls would effectively invoke onEachOperator from onEachOperator.
[中]为创建的每个操作符(Flux或Mono)添加或替换命名的发布者操作符拦截器。传递的函数应用于原始运算符发布服务器,并且可以返回不同的发布服务器,条件是它通常保持与原始服务器相同的数据类型。不鼓励使用Flux/MonoAPI,因为它会递归调用此钩子,从而导致StackOverflowerError。
请注意,子挂钩是累积的。使用同一个键调用此方法两次将用该名称替换旧的子钩子,但保留执行顺序(例如,A-h1、B-h2、A-h3将保留A-B执行顺序,导致钩子h3然后执行h2)。使用#resetOnEachOperator(String)删除一个特定的键,然后将其添加回去,将导致执行顺序发生变化(最后执行的是后面的子钩子)。可通过#resetOnEachOperator()完全重置。
这个切入点函数不能使用Flux、Mono或ParallelFlux API,因为它会导致对钩子的递归调用:操作符调用将有效地从onEachOperator调用onEachOperator。
代码示例来源:origin: reactor/reactor-core
/**
* Add a {@link Publisher} operator interceptor for each operator created
* ({@link Flux} or {@link Mono}). The passed function is applied to the original
* operator {@link Publisher} and can return a different {@link Publisher},
* on the condition that it generically maintains the same data type as the original.
* Use of the {@link Flux}/{@link Mono} APIs is discouraged as it will recursively
* call this hook, leading to {@link StackOverflowError}.
* <p>
* Note that sub-hooks are cumulative, but invoking this method twice with the same instance
* (or any instance that has the same `toString`) will result in only a single instance
* being applied. See {@link #onEachOperator(String, Function)} for a variant that
* allows you to name the sub-hooks (and thus replace them or remove them individually
* later on). Can be fully reset via {@link #resetOnEachOperator()}.
* <p>
* This pointcut function cannot make use of {@link Flux}, {@link Mono} or
* {@link ParallelFlux} APIs as it would lead to a recursive call to the hook: the
* operator calls would effectively invoke onEachOperator from onEachOperator.
*
* @param onEachOperator the sub-hook: a function to intercept each operation call
* (e.g. {@code map (fn)} and {@code map(fn2)} in {@code flux.map(fn).map(fn2).subscribe()})
*
* @see #onEachOperator(String, Function)
* @see #resetOnEachOperator(String)
* @see #resetOnEachOperator()
* @see #onLastOperator(Function)
*/
public static void onEachOperator(Function<? super Publisher<Object>, ? extends Publisher<Object>> onEachOperator) {
onEachOperator(onEachOperator.toString(), onEachOperator);
}
代码示例来源:origin: reactor/reactor-core
@Test
public void onEachOperatorSameLambdaDifferentNamesAppliedTwice() {
AtomicInteger applied = new AtomicInteger();
Function<? super Publisher<Object>, ? extends Publisher<Object>> hook = p -> {
applied.incrementAndGet();
return p;
};
Hooks.onEachOperator(hook);
Hooks.onEachOperator("other", hook);
Hooks.onEachOperatorHook.apply(s -> {});
assertThat(applied.get()).isEqualTo(2);
}
代码示例来源:origin: reactor/reactor-core
@Test
public void onEachOperatorSameLambdaSameNameAppliedOnce() {
AtomicInteger applied = new AtomicInteger();
Function<? super Publisher<Object>, ? extends Publisher<Object>> hook = p -> {
applied.incrementAndGet();
return p;
};
Hooks.onEachOperator(hook);
Hooks.onEachOperator(hook);
Hooks.onEachOperatorHook.apply(s -> {});
assertThat(applied.get()).isEqualTo(1);
}
代码示例来源:origin: reactor/reactor-core
@Test
public void onEachOperatorOneHookNoComposite() {
Function<? super Publisher<Object>, ? extends Publisher<Object>> hook = p -> p;
Hooks.onEachOperator(hook);
assertThat(Hooks.onEachOperatorHook).isSameAs(hook);
}
代码示例来源:origin: reactor/reactor-core
@Test
public void onEachOperatorResetSpecific() {
List<String> applied = new ArrayList<>(3);
Function<? super Publisher<Object>, ? extends Publisher<Object>> hook1 = p -> {
applied.add("h1");
return p;
};
Function<? super Publisher<Object>, ? extends Publisher<Object>> hook2 = p -> {
applied.add("h2");
return p;
};
Hooks.onEachOperator("1", hook1);
Hooks.onEachOperator(hook2);
Hooks.onEachOperatorHook.apply(s -> {});
assertThat(Hooks.getOnEachOperatorHooks()).hasSize(2);
assertThat(applied).containsExactly("h1", "h2");
applied.clear();
Hooks.resetOnEachOperator("1");
Hooks.onEachOperatorHook.apply(s -> {});
assertThat(Hooks.getOnEachOperatorHooks()).hasSize(1);
assertThat(applied).containsExactly("h2");
}
代码示例来源:origin: reactor/reactor-core
};
Hooks.onEachOperator("1", hook1);
Hooks.onEachOperator("2", hook2);
Hooks.onEachOperatorHook.apply(s -> {});
Hooks.onEachOperator("1", hook3);
Hooks.onEachOperatorHook.apply(s -> {});
代码示例来源:origin: reactor/reactor-core
@Test
public void onEachOperatorReset() {
Hooks.onEachOperator("some", p -> p);
assertThat(Hooks.onEachOperatorHook).isNotNull();
assertThat(Hooks.getOnEachOperatorHooks()).hasSize(1);
Hooks.resetOnEachOperator();
assertThat(Hooks.onEachOperatorHook).isNull();
assertThat(Hooks.getOnEachOperatorHooks()).isEmpty();
}
代码示例来源:origin: reactor/reactor-core
@Test
public void onEachOperatorClearByName() {
Hooks.onEachOperator("some", p -> p);
Hooks.onEachOperator("other", p -> p);
assertThat(Hooks.onEachOperatorHook).isNotNull();
assertThat(Hooks.getOnEachOperatorHooks()).hasSize(2);
Hooks.resetOnEachOperator("some");
assertThat(Hooks.onEachOperatorHook).isNotNull();
assertThat(Hooks.getOnEachOperatorHooks())
.hasSize(1)
.containsOnlyKeys("other");
Hooks.resetOnEachOperator("other");
assertThat(Hooks.onEachOperatorHook).isNull();
assertThat(Hooks.getOnEachOperatorHooks()).isEmpty();
}
代码示例来源:origin: reactor/reactor-core
Hooks.onEachOperator(h -> {
Flux<Object> publisher = TestPublisher.create()
.flux();
return publisher;
});
Hooks.onEachOperator(h -> {
hook2.set(h);
return h;
代码示例来源:origin: reactor/reactor-core
@Test
public void verboseExtension() {
Queue<String> q = new LinkedTransferQueue<>();
Hooks.onEachOperator(p -> {
q.offer(p.toString());
return p;
Hooks.resetOnEachOperator();
Hooks.onEachOperator(p -> {
q.offer(p.toString());
return p;
代码示例来源:origin: reactor/reactor-core
Hooks.onOperatorDebug();
Hooks.onEachOperator(p -> {
System.out.println(Scannable.from(p).stepName());
return p;
代码示例来源:origin: reactor/reactor-core
@Test
public void eachOperatorTest() {
Hooks.onEachOperator(Operators.lift((sc, sub) ->
new CoreSubscriber<Object>(){
@Override
代码示例来源:origin: reactor/reactor-core
@Test
public void eachOperatorTest() {
Hooks.onEachOperator(Operators.lift((sc, sub) ->
new CoreSubscriber<Object>(){
@Override
代码示例来源:origin: io.projectreactor/reactor-core
/**
* Add a {@link Publisher} operator interceptor for each operator created
* ({@link Flux} or {@link Mono}). The passed function is applied to the original
* operator {@link Publisher} and can return a different {@link Publisher},
* on the condition that it generically maintains the same data type as the original.
* Use of the {@link Flux}/{@link Mono} APIs is discouraged as it will recursively
* call this hook, leading to {@link StackOverflowError}.
* <p>
* Note that sub-hooks are cumulative, but invoking this method twice with the same instance
* (or any instance that has the same `toString`) will result in only a single instance
* being applied. See {@link #onEachOperator(String, Function)} for a variant that
* allows you to name the sub-hooks (and thus replace them or remove them individually
* later on). Can be fully reset via {@link #resetOnEachOperator()}.
* <p>
* This pointcut function cannot make use of {@link Flux}, {@link Mono} or
* {@link ParallelFlux} APIs as it would lead to a recursive call to the hook: the
* operator calls would effectively invoke onEachOperator from onEachOperator.
*
* @param onEachOperator the sub-hook: a function to intercept each operation call
* (e.g. {@code map (fn)} and {@code map(fn2)} in {@code flux.map(fn).map(fn2).subscribe()})
*
* @see #onEachOperator(String, Function)
* @see #resetOnEachOperator(String)
* @see #resetOnEachOperator()
* @see #onLastOperator(Function)
*/
public static void onEachOperator(Function<? super Publisher<Object>, ? extends Publisher<Object>> onEachOperator) {
onEachOperator(onEachOperator.toString(), onEachOperator);
}
内容来源于网络,如有侵权,请联系作者删除!