We have a mix of reactor types (Mono/Flux) and Kotlin Flow<> types in our codebase, and trying...

# coroutines

[martypitt]

We have a mix of reactor types (Mono/Flux) and Kotlin Flow<> types in our codebase, and trying to standardize on one.

Is there anything in Flow<> that can be used to indicate "This returns a single item" as a Mono does?

[kevin.cianfarini]

A Mono is conceptually equivalent to a suspending function which returns a single value.

[martypitt]

Yeah ... but when we're doing things like serialization, the difference becomes important.

Flow appears to always be serialized as an array - wheras a Mono indicates that the item should be serialized as an object.

[kevin.cianfarini]

I’m not sure of your specific use case and I don’t know Mono/Flux. I find it surprising that you’re serializing types meant to facilitate asynchronous computation, though.

[martypitt]

This is a pretty common use-case:

@GetMapping("/api/customers")
fun getCustomer(id:String):Mono<Customer>

It returns exactly one Customer, so is serialized to JSON as { ... }

The equivalent on Flow would be:

@GetMapping("/api/customers")
fun getCustomer(id:String):Flow<Customer>

Which serializes as [ { ... } ], which isn't what we need to return.

[streetsofboston]

Not using a Flow.
For syntactically and semantically indicating one async/emitted item, use a suspend function.

[kevin.cianfarini]

I am unsure if there’s more native integrations than this, but you’ll likely want to use the mono builder for this. https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-reactor/kotlinx.coroutines.reactor/mono.html

@GetMapping("/api/customers")
fun getCustomer(id:String):Mono<Customer> = mono {
  mySuspendingFunction()
}

[kevin.cianfarini]

I presume this is Spring webflux integration. They may have native support for declaring suspending @GetMapping functions so you don’t have to use the mono builder everywhere

[kevin.cianfarini]

It looks like they do if you toggle the Kotlin section on these docs https://docs.spring.io/spring-framework/reference/web/webflux-functional.html

[kevin.cianfarini]

val route = coRouter { 
	accept(APPLICATION_JSON).nest {
		GET("/person/{id}", handler::getPerson)
		GET("/person", handler::listPeople)
	}
	POST("/person", handler::createPerson)
}


class PersonHandler(private val repository: PersonRepository) {

	// ...

	suspend fun listPeople(request: ServerRequest): ServerResponse {
		// ...
	}

	suspend fun createPerson(request: ServerRequest): ServerResponse {
		// ...
	}

	suspend fun getPerson(request: ServerRequest): ServerResponse {
		// ...
	}
}

[martypitt]

Thanks Kevin / Anton.