All generic questions around Reactor. For advanced questions you can also try #reactor-core and #reactor-netty
@KATKrazy
Hi,
I'm trying to figure out how to ensure thread is finished when I use publishOn;
Flux.just(o)
.doOnNext(System.out::println)
.publishOn(Schedulers.single())
.doOnNext(o -> process(o))
.subscribe();
I was wondering how to make sure the process function is finished when I use subscribe() instead of block()
another doOnNext after .doOnNext(o -> process(o)) would do it. isn't it?
Here are netty startup logs
2021-01-02 22:34:37.861 DEBUG 29069 --- [ main] r.netty.resources.DefaultLoopEpoll : Default Epoll support : false
2021-01-02 22:34:37.863 DEBUG 29069 --- [ main] r.netty.resources.DefaultLoopKQueue : Default KQueue support : false
2021-01-02 22:34:37.871 DEBUG 29069 --- [ main] i.n.channel.MultithreadEventLoopGroup : -Dio.netty.eventLoopThreads: 24
2021-01-02 22:34:37.899 DEBUG 29069 --- [ main] i.n.u.internal.InternalThreadLocalMap : -Dio.netty.threadLocalMap.stringBuilder.initialSize: 1024
2021-01-02 22:34:37.899 DEBUG 29069 --- [ main] i.n.u.internal.InternalThreadLocalMap : -Dio.netty.threadLocalMap.stringBuilder.maxSize: 4096
2021-01-02 22:34:37.908 DEBUG 29069 --- [ main] io.netty.channel.nio.NioEventLoop : -Dio.netty.noKeySetOptimization: false
2021-01-02 22:34:37.908 DEBUG 29069 --- [ main] io.netty.channel.nio.NioEventLoop : -Dio.netty.selectorAutoRebuildThreshold: 512
2021-01-02 22:34:37.919 DEBUG 29069 --- [ main] i.netty.util.internal.PlatformDependent : org.jctools-core.MpscChunkedArrayQueue: available
2021-01-02 22:34:37.987 DEBUG 29069 --- [ main] io.netty.util.ResourceLeakDetector : -Dio.netty.leakDetection.level: simple
2021-01-02 22:34:37.987 DEBUG 29069 --- [ main] io.netty.util.ResourceLeakDetector : -Dio.netty.leakDetection.targetRecords: 4
2021-01-02 22:34:38.124 DEBUG 29069 --- [ main] io.netty.channel.DefaultChannelId : -Dio.netty.processId: 29069 (auto-detected)
2021-01-02 22:34:38.130 DEBUG 29069 --- [ main] io.netty.channel.DefaultChannelId : -Dio.netty.machineId: 38:f9:d3:ff:fe:22:64:e8 (auto-detected)
2021-01-02 22:34:38.166 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.numHeapArenas: 24
2021-01-02 22:34:38.166 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.numDirectArenas: 24
2021-01-02 22:34:38.166 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.pageSize: 8192
2021-01-02 22:34:38.166 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.maxOrder: 11
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.chunkSize: 16777216
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.tinyCacheSize: 512
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.smallCacheSize: 256
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.normalCacheSize: 64
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.maxCachedBufferCapacity: 32768
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.cacheTrimInterval: 8192
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.cacheTrimIntervalMillis: 0
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.useCacheForAllThreads: true
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.maxCachedByteBuffersPerChunk: 1023
2021-01-02 22:34:38.175 DEBUG 29069 --- [ main] io.netty.buffer.ByteBufUtil : -Dio.netty.allocator.type: pooled
2021-01-02 22:34:38.175 DEBUG 29069 --- [ main] io.netty.buffer.ByteBufUtil : -Dio.netty.threadLocalDirectBufferSize: 0
2021-01-02 22:34:38.176 DEBUG 29069 --- [ main] io.netty.buffer.ByteBufUtil : -Dio.netty.maxThreadLocalCharBufferSize: 16384
2021-01-02 22:34:38.203 DEB
I want to access value generated in Mono later in the reactive pipeline
Mono.just(nameService.getName()).map(name -> addressService.getAddress(name)).map(address ->postService.postLetter(name,address))
how can I access name to pass for postLetter, which is generated from nameService without calling it second time ?
4 replies
1 reply
1 reply
Hi, @joshlong
I want to build a kind of priority based message exchange mechanism in Reactor. Messages arrive in a form of infinite stream. I'd like messages to have different limitRate() applied depending on the priority they carry. Sum of pressure on the consumer that is generated by all priorities should be constant. I managed to achieve a half-baked solution (see below) with use of window but actually waiting for the completion of window seems to slow down performance.
Three questions here:
- Is there a way to conditionally apply different limitrate() without using window and all the modifications around it?
- Do you have any advice on testing these parallel processes execution time - is logging accurate enough?
- If something feels incredibly wrong here - shoot!
public Flux<ResponseMessage> applyPrio(Flux<RequestMessage> inflow) {
return inflow.window(Duration.ofMillis(windowDurationMillis))
.flatMap(windowFlux -> windowFlux
.sort(Comparator.comparing(RequestMessage::getPriority))
.groupBy(RequestMessage::getPriority)
.flatMap(gf -> gf
.publishOn(Schedulers.parallel())
.onBackpressureBuffer()
.flatMap(exchangeFunction::apply)
.limitRate(this.priorityToRate.apply(gf.key()), this.priorityToRate.apply(gf.key())))
);
}
Things that everyone should take care of...
Mono.just(X).subscribe(new HookSubscriber());
public class HookSubscriber extends BaseSubscriber {
@Overrite
public void hookOnNext(T value){
//do something
}
}
Given a cold TestPublisher which is created with e.g. 100 items available and then completes.
Is there a way to leverage TestPublisher and StepVerifier to assert the number of items requested from this TestPublisher` source?
I see there are the assertMinRequested(min) and assertMaxRequested(max) methods, however my first impression is these indicate the number of "open" requests and is therefore always 0 in my setup with pre-created source.
Example code:
@Test
public void test() {
TestPublisher<Integer> publisher = TestPublisher.createCold();
Flux.range(1, 100).doOnNext(publisher::emit).subscribe();
publisher.complete();
StepVerifier.create(publisher)
.thenRequest(1)
.expectNextCount(1)
// .then(() -> publisher.assertMinRequested(1)) // FAILS
.thenRequest(10)
.expectNextCount(10)
// .then(() -> publisher.assertMinRequested(10)) // FAILS
.thenRequest(Long.MAX_VALUE)
.expectNextCount(89)
.expectComplete()
.verify();
}My goal here is to be able to test that upstream is not consumed in an unbound way when I .transform(...) it with a custom component.
28 replies
Flux<List<T>> into a Flux<T>, with respect for backpressure? ("split" / inverse buffer operation)I've been trying with the
flatMap / flatMapIterable / concatMapIterable operations but my testing (see previous question) suggests they all consume the upstream in an unbound way...flux.concatMapIterable(Function.identity(), 1);
flux
.log()
.flatMap(FluxMetrics::complexMono1)
.name("pipeline")
.tag("operation","complexMono1")
.metrics()
.flatMap(FluxMetrics::complexMono2)
.name("pipeline")
.tag("operation","complexMono2")
.metrics()
.flatMap(FluxMetrics::complexMono3)
.name("pipeline")
.tag("operation","complexMono2")
.metrics()
.subscribe(System.out::println);
I have metrics configured for my reactive stream in the following manner. I am observing that only the first metrics are getting publish. Could someone help understand if this is expected? Here is the structure of my reactive sequence.
flux .log() .flatMap(FluxMetrics::complexMono1) .name("pipeline") .tag("operation","complexMono1") .metrics() .flatMap(FluxMetrics::complexMono2) .name("pipeline") .tag("operation","complexMono2") .metrics() .flatMap(FluxMetrics::complexMono3) .name("pipeline") .tag("operation","complexMono2") .metrics() .subscribe(System.out::println);
Could someone please help with understanding if it is expected to see only the first metrics? My assumption was first metrics after first flatmap gives insight on the first flatmap operator and so on. What would be the right place to have this metrics if not here?
Could someone please help
name ? in that case, the metrics would be at best mixed together
tags() with metrics(): it walks the whole line of parents to the metrics operator and finds ALL the above tags, from downmost to topmost. so eg. the third metrics() sees tag values complextMono3 then complextMono2 then complexMono1 and ends up using the later :(
I am using async http clients which require passing an executor to which the result will be dispatched to. What executor type is recommended here. When I use work stealing, I see the threads getting blocked under high TPS. I am not doing any heavy lifting except for logging (log4j2 async). I used this with the thought that this would be similar to parallel. Should I switch to Fixed executor pool or something similar?
1 reply