#17 - fixed write greed behavior

src/main/java/org/cqfn/rio/file/WriteGreed.java

@@ -96,15 +96,10 @@ public Constant(final long amount, final long shift) {
 
         @Override
         public boolean request(final Subscription sub) {
-            // @checkstyle AvoidInlineConditionalsCheck (1 line)
-            final long pos = this.cnt.updateAndGet(prev -> prev <= this.shift ? this.amount : prev);
-            final boolean result;
-            if (pos == this.amount) {
+            final long pos = this.cnt.getAndIncrement();
+            final boolean result = pos == 0 || pos % (this.amount - this.shift + 1) == 0;
+            if (result) {
                 sub.request(this.amount);
-                result = true;
-            } else {
-                this.cnt.decrementAndGet();
-                result = false;
             }
             return result;
         }

src/test/java/org/cqfn/rio/file/FileTest.java

@@ -34,6 +34,7 @@
 import java.security.NoSuchAlgorithmException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.cqfn.rio.ext.BufferSource;
 import org.cqfn.rio.ext.BufferSourceExtension;
 import org.hamcrest.MatcherAssert;
@@ -148,6 +149,26 @@ void copySingleThread(@TempDir final Path tmp) throws Exception {
         );
     }
 
+    @Test
+    void requestNextItemsOnlyOnDemand(@TempDir final Path tmp,
+        @BufferSource(buffers = 5) final Publisher<ByteBuffer> source) throws Exception {
+        final WriteGreed.Constant greed = new WriteGreed.Constant(3, 1);
+        final AtomicInteger requests = new AtomicInteger();
+        new File(tmp.resolve("output1")).write(
+            source,
+            sub -> {
+                final boolean requested = greed.request(sub);
+                if (requested) {
+                    requests.incrementAndGet();
+                }
+                return requested;
+            }
+        ).toCompletableFuture().get();
+        MatcherAssert.assertThat(
+            requests.get(), Matchers.is(3)
+        );
+    }
+
     @Test
     @Timeout(2)
     @EnabledIfSystemProperty(named = "test.hugeFiles", matches = "true|yes|on|1")

src/test/java/org/cqfn/rio/file/WriteGreedTest.java

@@ -0,0 +1,75 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 cqfn.org
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining
+ * a copy of this software and associated documentation files
+ * (the "Software"), to deal in the Software without restriction,
+ * including without limitation the rights * to use, copy, modify,
+ * merge, publish, distribute, sublicense, and/or sell copies of the Software,
+ * and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+ * ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+ * OTHER DEALINGS IN THE SOFTWARE.
+ */
+package org.cqfn.rio.file;
+
+import org.hamcrest.MatcherAssert;
+import org.hamcrest.Matchers;
+import org.junit.jupiter.api.Test;
+import org.reactivestreams.Subscription;
+
+/**
+ * Test case for {@link WriteGreed}.
+ *
+ * @since 1.0
+ * @checkstyle JavadocMethodCheck (500 lines)
+ */
+public final class WriteGreedTest {
+
+    /**
+     * Dummy subscription.
+     */
+    private static final Subscription SUB_DUMMY = new Subscription() {
+        @Override
+        public void request(final long items) {
+            // nothing
+        }
+
+        @Override
+        public void cancel() {
+            // nothing
+        }
+    };
+
+    @Test
+    void constantGreedRespectsAmountAndShift() {
+        final WriteGreed.Constant greed = new WriteGreed.Constant(3, 1);
+        // @checkstyle MagicNumberCheck (1 line)
+        final boolean[] results = new boolean[10];
+        for (int pos = 0; pos < results.length; ++pos) {
+            results[pos] = greed.request(WriteGreedTest.SUB_DUMMY);
+        }
+        MatcherAssert.assertThat(
+            results,
+            Matchers.equalTo(
+                new boolean[]{
+                    true, false, false,
+                    true, false, false,
+                    true, false, false,
+                    true,
+                }
+            )
+        );
+    }
+}