reactor-core/src/main/java/reactor/core/publisher/Flux.java
public final <V, C> Flux<T> distinct(
			Function<? super T, ? extends V> keySelector,
			Supplier<C> distinctStoreSupplier,
			BiPredicate<C, V> distinctPredicate,
			Consumer<C> cleanup) {
		if (this instanceof Fuseable) {
			return onAssembly(new FluxDistinctFuseable<>(this, keySelector,
					distinctStoreSupplier, distinctPredicate, cleanup));
		}
		return onAssembly(new FluxDistinct<>(this, keySelector, distinctStoreSupplier, distinctPredicate, cleanup));
	}
public final <V, C> Flux<T> distinct(
			Function<? super T, ? extends V> keySelector,
			Supplier<C> distinctStoreSupplier,
			BiPredicate<C, V> distinctPredicate,
			Consumer<C> cleanup) {
		if (this instanceof Fuseable) {
			return onAssembly(new FluxDistinctFuseable<>(this, keySelector,
					distinctStoreSupplier, distinctPredicate, cleanup));
		}
		return onAssembly(new FluxDistinct<>(this, keySelector, distinctStoreSupplier, distinctPredicate, cleanup));
	}
For each {@link Subscriber}, track elements from this {@link Flux} that have been
seen and filter out duplicates, as compared by applying a {@link BiPredicate} on
an arbitrary user-supplied {@code <C>} store and a key extracted through the user
provided {@link Function}. The BiPredicate should typically add the key to the
arbitrary store for further comparison. A cleanup callback is also invoked on the
store upon termination of the sequence.

<p>
<img class="marble" src="doc-files/marbles/distinctWithKey.svg" alt="">

@reactor.discard This operator discards elements that don't match the distinct predicate,
but you should use the {@code cleanup} as well if you need discarding of keys
categorized by the operator as "seen".

@param keySelector function to compute comparison key for each element
@param distinctStoreSupplier supplier of the arbitrary store object used in distinct
checks along the extracted key.
@param distinctPredicate the {@link BiPredicate} to apply to the arbitrary store +
extracted key to perform a distinct check. Since nothing is assumed of the store,
this predicate should also add the key to the store as necessary.
@param cleanup the cleanup callback to invoke on the store upon termination.

@param <V> the type of the key extracted from each value in this sequence
@param <C> the type of store backing the {@link BiPredicate}

@return a filtering {@link Flux} only emitting values with distinct keys