Cache all last distinct values

I have a stream of different values with key (which I can group by key - character here):

XY - where X is a key and Y is some other data

** - reference point (see latter)

streamInput: -A1--B1----C1--A2-**-A3-A4--B2--->

What I want is some combination of Flux.cache() and Flux.distinct(keySelector) let call this wanted method/pattern cacheDistinct().

So when new stream will subscribe at reference point ** I will get all distinct values before ** and all new.

For example:

var cachedStream = streamInput.cacheDistinct();

//will print: -A1--B1----C1--A2--A3-A4--B2--->
cacheDistinct.subscribe(i -> print(i)) 

//user/event click and we open new subscription at reference time **

//will print: -B1-C1-A2--> (all last distinct before ** and all new) ---A3-A4--B2--->
//order before ** doesn't matter
cacheDistinct.subscribe(i -> print(i))

How to achieve this cacheDistinct() method/pattern?

PS. I use project react, but I need an idea here, so you can show how you do it in other languages/frameworks reactively.