Add MoreGatherers.distinctUntilChanged()

README.md

@@ -22,4 +22,6 @@ Provided `Gatherers`:
 - `MoreGatherers.zip(Collection<T2>, BiFunction<T1,T2>)`
 - `MoreGatherers.zip(Iterator<T2>, BiFunction<T1,T2>)`
 - `MoreGatherers.zip(Stream<T2>, BiFunction<T1,T2>)`
+- `MoreGatherers.distinctUntilChanged()`
+- `MoreGatherers.distinctUntilChanged(Function<T, R>)`
 - `MoreGatherers.distinctBy(Function<T, R>)`

src/main/java/com/pivovarit/gatherers/MoreGatherers.java

@@ -50,6 +50,30 @@ private MoreGatherers() {
         );
     }
 
+    public static <T> Gatherer<T, ?, T> distinctUntilChanged() {
+        return distinctUntilChanged(Function.identity());
+    }
+
+    public static <T, U> Gatherer<T, ?, T> distinctUntilChanged(Function<? super T, ? extends U> keyExtractor) {
+        Objects.requireNonNull(keyExtractor, "keyExtractor can't be null");
+        class State {
+            U value;
+            boolean hasValue;
+        }
+        return Gatherer.ofSequential(
+          State::new,
+          (state, element, downstream) -> {
+              U key = keyExtractor.apply(element);
+              if (!state.hasValue || !Objects.equals(state.value, key)) {
+                  state.value = key;
+                  state.hasValue = true;
+                  downstream.push(element);
+              }
+              return true;
+          }
+        );
+    }
+
     public static <T1, T2> Gatherer<T1, ?, Map.Entry<T1, T2>> zip(Stream<T2> other) {
         Objects.requireNonNull(other, "other can't be null");
         return zip(other.iterator());

src/test/java/com/pivovarit/gatherers/DistinctUntilChangedMapperTest.java

@@ -0,0 +1,29 @@
+package com.pivovarit.gatherers;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.stream.Stream;
+
+import static com.pivovarit.gatherers.MoreGatherers.distinctUntilChanged;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class DistinctUntilChangedMapperTest {
+
+    @Test
+    void shouldDistinctUntilChangedEmptyStream() {
+        assertThat(Stream.empty().gather(distinctUntilChanged(i -> i))).isEmpty();
+    }
+
+    @Test
+    void shouldDistinctUntilChangedMapper() {
+        assertThat(Stream.of("a", "b", "bb", "cc", "ddd", "eee", "ffff", "ggggg", "hhhhh", "iiiii", "j", "k")
+          .gather(distinctUntilChanged(s -> s.length())))
+          .containsExactly("a", "bb", "ddd", "ffff", "ggggg", "j");
+    }
+
+    @Test
+    void shouldRejectNullMapper() {
+        assertThatThrownBy(() -> distinctUntilChanged(null)).isInstanceOf(NullPointerException.class);
+    }
+}

src/test/java/com/pivovarit/gatherers/DistinctUntilChangedTest.java

@@ -0,0 +1,23 @@
+package com.pivovarit.gatherers;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.stream.Stream;
+
+import static com.pivovarit.gatherers.MoreGatherers.distinctUntilChanged;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class DistinctUntilChangedTest {
+
+    @Test
+    void shouldDistinctUntilChangedEmptyStream() {
+        assertThat(Stream.empty().gather(distinctUntilChanged())).isEmpty();
+    }
+
+    @Test
+    void shouldDistinctUntilChanged() {
+        assertThat(Stream.of(1, 1, 2, 2, 3, 3, 4, 5, 5, 5, 1, 1)
+          .gather(distinctUntilChanged()))
+          .containsExactly(1, 2, 3, 4, 5, 1);
+    }
+}