package com.benbarron.react.operator;

import com.benbarron.react.Observer;
import com.benbarron.react.function.Action2;

import java.util.Collections;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class Distinct<T> implements Action2<T, Observer<T>> {

    private final Set<T> previousItems = Collections.newSetFromMap(new ConcurrentHashMap<>());

    @Override
    public void run(T item, Observer<T> observer) {
        if (previousItems.add(item)) {
            observer.onNext(item);
        }
    }
}

@@ -278,16 +278,52 @@
 %3CT%3E%3E %7B%0A%0A
+    // TODO: this will need cleanup%0A
     priv

Add todo