### Question:
DeduplicatingTransformer implements Function<Publisher<T>, Publisher<T>> { public static <T> DeduplicatingTransformer<T> identity(DeduplicationConfig config, Deduplication<T> deduplication) { return new DeduplicatingTransformer<>(config, deduplication); } private  DeduplicatingTransformer(DeduplicationConfig config, Deduplication<T> deduplication); static DeduplicatingTransformer<T> identity(DeduplicationConfig config, Deduplication<T> deduplication); static DeduplicatingTransformer<Acknowledgeable<T>> acknowledgeable(DeduplicationConfig config, Deduplication<T> deduplication); @Override Publisher<T> apply(Publisher<T> publisher);  }### Answer:
@Test public void dataAreSequentiallyProcessed() { Flux<String> invertedDownstream = processor.transform(DeduplicatingTransformer.identity(CONFIG, new InvertedReducerDeduplication())); StepVerifier.create(invertedDownstream) .expectSubscription() .then(() -> consumer.accept("ONE")) .thenAwait(Duration.ofMillis(200)) .then(() -> { consumer.accept("TWO"); consumer.accept("TWO"); consumer.accept("ONE"); }) .thenAwait(CONFIG.getDeduplicationDuration()) .expectNext("ONE") .expectNext("TWO") .expectNoEvent(CONFIG.getDeduplicationDuration()) .thenCancel() .verify(); }