add reactive stream support

[zigzago]

https://mongodb.github.io/mongo-java-driver-reactivestreams/

as async driver is now deprecated

[zigzago]

target 3.9.2 because more impacts than expected

[jyemin]

Can this be made transparent to applications via the coroutine support, or is the callback-based API exposed to applications?

[zigzago]

KMongo supports 3 versions of async extensions:

1. Callback style (kmongo-async)
2. RxJava2 (kmongo-rxjava2)
3. Coroutine (kmongo-coroutine)

My assumption is that for each version, it is doable to:

1. async : Duplicate the extensions with reactivestreams base classes -> ok
2. rxjava2 -> made it transparent -> seems ok
3. coroutine -> made it transparent (using https://github.com/Kotlin/kotlinx.coroutines/blob/master/reactive/coroutines-guide-reactive.md and replacing current xIterable extensions with XPublisher extensions ) -> seems ok too

But I think I need more time to stabilize the whole thing. The new plan is to push the feature in a few days and release it at the end of January.

[BKaraargirov]

You have probably seen it, but Kotlin have released very good extension functions, to transform rxjava and reactive streams into coroutines:
https://github.com/Kotlin/kotlinx.coroutines/blob/master/reactive/coroutines-guide-reactive.md

I played around with the reactive stream one (https://github.com/Kotlin/kotlinx.coroutines/tree/master/reactive/kotlinx-coroutines-reactive) and it works pretty well.

I was going to try migrate the async driver and make pull request, but I haven't used Flapdoodle. And changing out the async driver hits it and some related custom abstractions. Unfortunately I don't have enough time right now to dig deeper

[zigzago]

The feature is pushed. Async & rxjava2 modules looks good to me, but a little issue remains with coroutine.

This is because of the KMongo design, that use heavily extensions. The original idea was to add additional behavior to the java driver, and not to hide its methods. It is great because we avoid the problem that have all other frameworks: we don't need to duplicate all the new features of the mongo driver (and they advance rather quickly ;) )

But with coroutine, we change the behavior of existing methods. It was not a problem when extending the java async driver, as there was almost no signature clash. But with reactivestreams + coroutine, we have many signature clashes. For example, with the insertOne method:

Publisher<Success> insertOne(TDocument document)

We would like to write this extension method:

suspend fun <T> MongoCollection<T>.insertOne(document: T): Success 
 = insertOne(document).awaitSingle()

But as the signature is the same, the extension is shadowed. So the compiler will always use the java driver version. I have renamed the extensions using a 'AndAwait' suffix pattern, but this is quite error prone. I you write:

col.insertOneAndAwait(doc)

The insertion is applied. But it you write

col.insertOne(doc)

Nothing happens (of course).

I'm currently thinking of introducing a dedicated MongoCollection interface in the coroutine module, in order to avoid this kind of issue.

[phiSgr]

I feel strongly that the right way out is to argue against the deprecation of the callback API.