Return Mono.empty() if a value is found but without executing other steps

Question

This problem is very hard to describe as text, so if the title doesn't fit the requirement, sorry for that.

I would like to achieve a specific goal with Project Reactor flux and mono, which seems to be pretty simple at first look.

A code example, in the "blocking-way" will be better than a long description:

    fun findGroupToCreateBlocking(userId: UUID, groupLabel: String): Optional<LinkUserToGroup> {
        val group = lib.findGroupsOfUser(userId)
                .flatMapIterable { it.items }
                .filter { it.label == groupLabel }
                .toMono()
                .blockOptional()

        if(group.isPresent) {
            return Optional.empty()
        }

        return lib.searchGroups(groupLabel)
                .flatMapIterable { it.items }
                .filter { it.label == groupLabel }
                .toMono()
                .map { LinkUserToGroup(userId, it.id) }
                .switchIfEmpty { IllegalStateException("Group $groupLabel not found").toMono() }
                .blockOptional()
    }

I try to achieve the same thing without the block part of course. I ended up with the following code:

fun findGroupToCreateReactive(userId: UUID, groupLabel: String): Mono<LinkUserToGroup> =
            lib.findGroupsOfUser(userId)
                    .flatMapIterable { it.items }
                    .filter { it.label == groupLabel }
                    .toMono()
                    .map { Optional.of(it) }
                    .defaultIfEmpty(Optional.empty())
                    .filter { g -> g.isEmpty }
                    .flatMap { lib.searchGroups(groupLabel)
                            .flatMapIterable { it.items }
                            .toMono()
                            .map { LinkUserToGroup(userId, it.id) }
                            .switchIfEmpty { IllegalStateException("Group $groupLabel not found").toMono() }
                    }

I think (and I'm not the only one 😇) we can do better and not rely on the Optional usage in the middle of the stream... but I didn't find any other solution.

This is the fourth time I fight against this "pattern", so some help would be welcomed!

I've generated a demo project on Gitlab (here) with unit tests for both reactive and blocking implementation to see if the proposition match the requirement. If you want, you can fork and use the project.

Answer 1

Instead of using a Mono.empty, I used the Flux.hasElement method (like @yossarian), but adding a negation filter. It seems to work as the unit test still pass.

    fun findGroupToCreateReactive(userId: UUID, groupLabel: String): Mono<LinkUserToGroup> =
            lib.findGroupsOfUser(userId)
                    .flatMapIterable { it.items }
                    .map { it.label }
                    .hasElement(groupLabel)
                    .filter { g -> !g }
                    .flatMap { lib.searchGroups(groupLabel)
                            .flatMapIterable { it.items }
                            .toMono()
                            .map { LinkUserToGroup(userId, it.id) }
                            .switchIfEmpty { IllegalStateException("Group $groupLabel not found").toMono() }
                    }

Since we want to search groups only if the user doesn't belong to a group, then this is made more explicit with the negation filter.