+ every, +repeat, repeatCallable

Author: akarnokd

README.md

@@ -688,6 +688,42 @@ Flowable.intervalRange(1, 5, 100, 100, TimeUnit.MILLISECONDS)
         .assertResult();
 ```
 
+### Flowables.repeat
+
+Repeats a scalar value indefinitely (until the downstream actually cancels), honoring backpressure and supporting synchronous fusion and/or conditional fusion.
+
+```java
+Flowable.repeat("doesn't matter")
+.map(v -> ThreadLocalRandom.current().nextDouble())
+.take(100)
+.all(v -> v < 1d)
+.test()
+.assertResult(true);
+```
+
+### Flowables.repeatCallable
+
+Repeatedly calls a callable, indefinitely (until the downstream actually cancels) or if the callable throws or returns null (when it signals `NullPointerException`), honoring backpressure and supporting synchronous fusion and/or conditional fusion.
+
+```java
+Flowable.repeatCallable(() -> ThreadLocalRandom.current().nextDouble())
+.take(100)
+.all(v -> v < 1d)
+.test()
+.assertResult(true);
+```
+
+### FlowableTransformers.every
+
+Relays every Nth item from upstream (skipping the in-between items).
+
+```java
+Flowable.range(1, 5)
+.compose(FlowableTransformers.<Integer>every(2))
+.test()
+.assertResult(2, 4)
+```
+
 ## Special Publisher implementations
 
 ### Nono - 0-error publisher

gradle.properties

@@ -1 +1 @@
-version=0.14.1
+version=0.14.2

src/main/java/hu/akarnokd/rxjava2/operators/FlowableEvery.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2016 David Karnok
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package hu.akarnokd.rxjava2.operators;
+
+import org.reactivestreams.*;
+
+import io.reactivex.*;
+import io.reactivex.internal.subscriptions.SubscriptionHelper;
+import io.reactivex.internal.util.BackpressureHelper;
+
+/**
+ * Emits only every Nth item.
+ *
+ * @param <T> the value type
+ *
+ * @since 0.14.2
+ */
+final class FlowableEvery<T> extends Flowable<T> implements FlowableTransformer<T, T> {
+
+    final Publisher<T> source;
+
+    final long keep;
+
+    FlowableEvery(Publisher<T> source, long keep) {
+        this.source = source;
+        this.keep = keep;
+    }
+
+    @Override
+    protected void subscribeActual(Subscriber<? super T> s) {
+        source.subscribe(new EverySubscriber<T>(s, keep));
+    }
+
+    @Override
+    public Publisher<T> apply(Flowable<T> upstream) {
+        return new FlowableEvery<T>(upstream, keep);
+    }
+
+    static final class EverySubscriber<T> implements Subscriber<T>, Subscription {
+
+        final Subscriber<? super T> actual;
+
+        final long keep;
+
+        long index;
+
+        Subscription s;
+
+        public EverySubscriber(Subscriber<? super T> actual, long keep) {
+            this.actual = actual;
+            this.keep = keep;
+        }
+
+        @Override
+        public void onSubscribe(Subscription s) {
+            if (SubscriptionHelper.validate(this.s, s)) {
+                this.s = s;
+
+                actual.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onNext(T t) {
+            long i = index + 1;
+            if (i == keep) {
+                index = 0;
+                actual.onNext(t);
+            } else {
+                index = i;
+            }
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            actual.onError(t);
+        }
+
+        @Override
+        public void onComplete() {
+            actual.onComplete();
+        }
+
+        @Override
+        public void request(long n) {
+            if (SubscriptionHelper.validate(n)) {
+                long u = BackpressureHelper.multiplyCap(n, keep);
+                s.request(u);
+            }
+        }
+
+        @Override
+        public void cancel() {
+            s.cancel();
+        }
+    }
+}