Now : Reactor Core 3.3.0.RELEASE - Reactor Netty 0.9.0.RELEASE - BOM : DYSPROSIUM-RELEASE
override fun getAttributeInformation(attribute: Attribute): Mono<AttributeInformation> {
val input = data()
.doOnSubscribe { logger.warn { attribute.id } }
.doOnNext { logger.info { "ID: ${it.id}" } }
.publish()
.autoConnect(2)
val date = input
.doOnNext { logger.info { "ID: ${it.id}" } }
.map { it.lastModifiedDate }
.max()
.doOnNext { logger.info { "date is $it" } }
val relatedInformation = input
.doOnNext { logger.info { "ID: ${it.id}" } }
.flatMapIterable { it.toPairs(attribute.id) }
.groupBy({ it.first.value }) { it.second }
.flatMap { g -> g.collect(Collectors.toSet()).map { g.key()!! to it } }
.collectMap({ it.first }) { it.second }
.map { RelatedInformation(byValue = it) }
.doOnNext { logger.info { "related is done" } }
return Mono.zip(date, relatedInformation)
.map { AttributeInformation(attributeId = attribute.id, lastModifiedDate = it.t1, related = it.t2) }
.doOnSuccess { logger.warn { attribute.id } }
.doOnError { logger.error(it) { attribute.id } }data() and I get a lot of IDs in my log that seem not to be consumed by date or relatedInformation
I have the following issue - to invoke a web service, I need to get some params from Mongo. I don't want to use blocking IO and use subscribe. But there is a problem to get the value from subscribe. Could anyone tell me a good solution how it's possible to get value using subscribe method?
How can I make sure to clean up the resources (InputStream and ByteBuf) if a TimeoutException is thrown in this scenario?
@Test
public void testReleasingResources() throws InterruptedException {
final String result =
HttpClient.create(ConnectionProvider.fixed("dummy", 10))
.protocol(HttpProtocol.HTTP11)
.keepAlive(true)
.request(HttpMethod.GET)
.uri("http://localhost:" + mockServer.port() + "/hello")
.responseSingle((response, bytes) -> bytes
.asInputStream()
.map(is -> new SomeClass(is))
.flatMap(sc -> slowMethod(sc))
)
.timeout(Duration.ofMillis(500))
.map(someClass -> convertToString(someClass))
.block();
Assert.assertEquals("Hello World!", result);
}block etc is used just to simplify the test.
@violetagg @simonbasle We discussed about this during SpringOne Platform and you suggested using doOnDiscard and provide Consumer for specific types, e.g., InputStream, ByteBuf or SomeClass. I added bunch of doOnDiscard for different types, but, I do not see doOnDiscard being called at all. I assume this is because there is no element in the buffer of any operator? Timeout happens while element is being processed in one of the operators?
I will very much appreciate any suggestion here.
You can find the full dummy test code here https://gist.github.com/anilgursel/301efbac56f1a6755956da6f63d8a0df
Be aware that combineLatest will not emit an initial value until each observable emits at least one value. This is the same behavior as withLatestFrom and can be a gotcha as there will be no output and no error but one (or more) of your inner observables is likely not functioning as intended, or a subscription is late.
InputStream is already emitted to the map by the time timeout occurs? doOnDiscard's main intent is that the operators that buffer elements as part of their natural operation will pass these to the consumer(s) IF they haven't emitted said elements
SomeClass that would be passed to the consumer. flatMap is supposed to apply doOnDiscard eg. if it held an element for backpressure purpose
doOnDiscard for SomeClass, which did not help neither. When timeout is triggered, the element is being processed inside one of the operators, it is not in buffer (and not back pressured). What is the best way to deal with this? Do we need to add a try/catch inside all such operators? I was looking for a hook (like doOnDiscard, doOnError) which would be called when an error/cancellation happens, but also allows me to access the resource to clean up.
what if the cancellation happens right in the middle of your mapping function for instance?
When an exception happens and the stream is canceled, wouldn’t it cancel the execution on all operators? Is there any possibility that the stream ends with error and that operator keeps running?
If not, then having a doOnEror which provides access to the element would help, IMO. At that point, one can check the state and determine what to do.
onNext/onError)
doOnDiscard is using the Context to approximate that
onNext|onComplete|onError|cancel signals. it just puts your Consumer(s) in the Context under a key that each supporting operators has to explicitly look for in their implementations. not 100% of operators support it (because it only makes sense for a subset of operators to pay that overhead)
using ? I wonder if it could fit the bill here
InputStream/ SomeClass conversion and its usage, and allowing to call close() on it when the downstream cancels
using in more detail, but from a quick glance, it looked to me for being used for the resources that are created outside of the stream. In my case, the resource is created inside the stream.. the resource will potentially continue to exist outside of the stream in happy path, but I would like to make sure it gets cleaned up in case of an exception/cancellation.
I am getting this error java.lang.ClassCastException: reactor.core.publisher.FluxMapFuseable incompatible with java.lang.String
do you have any idea
plugins {
id("org.springframework.boot") version "2.2.0.RELEASE"
id("io.spring.dependency-management") version "1.0.8.RELEASE"
kotlin("jvm") version "1.3.50"
kotlin("plugin.spring") version "1.3.50"
}
group = "com.aiod"
version = "0.0.1-SNAPSHOT"
java.sourceCompatibility = JavaVersion.VERSION_1_8
repositories {
mavenCentral()
maven { url = uri("https://repo.spring.io/milestone") }
}
val springCloudVersion = "Hoxton.M3"
Execution failed for task ':compileKotlin'.
Could not resolve all files for configuration ':compileClasspath'.
Could not find org.springframework.cloud:spring-cloud-function-kotlin:.
Required by:
project :
Hey all - I'm building out a webflux app and trying to push/evangelize it to the wider org, but there's some 'simple' things that I can't help feel I'm not doing right. The main one is 'domain' validation - i.e. throwing an error from a service class. Currently I'm doing something like this:
Mono<Boolean> isValidFuture = ContextUtils.getUserJwt()
.map(JwtUtils::getUserId)
.zipWith(itemFuture, validItemUser())
.filter(Boolean::booleanValue)
.switchIfEmpty(Mono.error(new RuntimeException("User can't act on this item")));I then have to wire this isValidFuture in with the code that should do something after validation has been performed. Is there a simpler way/more idiomatic to achieve this?
i.e. I end up doing something like:
return isValidFuture.flatMap(x -> Mono.zip(itemUpdateFuture, statusUpdateFuture))
.map(tuple2 -> new NiceResponseObject(tuple2.getT2()));it's this bit ^ that I feel is suboptimal, but can't think of a way to have the framework subscribe to the isValidFuture any other way.
