Add flush functionality to distinctUntilChanged

[mpodlasin]

Use case:
We are implementing custom notification system for our web messenger app. If message comes, user should see notification with details of who wrote a message. Notification disappears after 3s. If user A wrote to me twice within these 3s, I should see only one notification, 3s after his first message arrived to me, with the last message which he sent me. If however some other user B writes to me within these 3s, I should see new notification, this time counting from the moment I got first message from B.

We wanted to write something like this in redux-observable epic:

    action$
        .ofType(OPEN_NOTIFICATION)
        .distinctUntilChanged(
            notificationIsFromTheSameUser, 
            action$.ofType(REMOVE_NOTIFICATION) // imaginary flush argument
        )
        .switchMap(({payload}) => 
            Observable.of(removeNotification(payload)).delay(3000)
        )

which would start counting 3s when user A writes, and would ignore subsequent messages from him, but AFTER 3s would again listen to messages from him. Without flush functionality, after first display of notification, we will never close notification from A, since subsequent events from A will be ignored, unless some other user writes to me.

We ended up with:

    action$
        .ofType(OPEN_NOTIFICATION)
        .windowTime(3000)
        .mergeMap(openNotifs$) => {
             return openNotifs$.
                     .distinctUntilChanged(
                            notificationIsFromTheSameUser, 
                            action$.ofType(REMOVE_NOTIFICATION)
                     )
                     .switchMap(({payload}) => 
                             Observable.of(removeNotification(payload)).delay(3000)
                     )
        })

which works, but this is much more complicated and less readable, since it utilizes higher order observable. I have difficulties myself wrapping my head around what is happening in that code.

Proposal
distinct operator already has flush argument, which controls cache of that operator. We could add similar argument to distinctUntilChanged where operator would "reset" whenever flush observable would complete or emit an event.

PS
I will gladly accept better solutions to that problem. ;)

[mpodlasin]

Actually I just realized that windowTime solution will not work properly, in case where users A and B send respectively messages in first window and then user B sends message in second window, thus causing REMOVE_NOTIFICATION event to be sent twice.

[kievsash]

I would do it this way:

let usersToDelay = {}

action$
    .ofType(OPEN_NOTIFICATION)
    .distinctUntilChanged( (prev, next) => {
        let userName = next.userName; 
        if (usersToDelay[userName]) {
          return true; // do not emit notification
        }
        setTimeout(() => {delete usersToDelay[userName];}, 3000);
        return false; //emit if another use;
     }
    )