Merge pull request #50 from nstdio/micro-ops

chore: Micro optimizations.

src/main/java/io/github/nstdio/http/ext/Buffers.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright (C) 2022 Edgar Asatryan
+ *
+ * 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 io.github.nstdio.http.ext;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+class Buffers {
+    private Buffers() {
+    }
+
+    static ByteBuffer duplicate(ByteBuffer buf) {
+        var dup = buf.asReadOnlyBuffer();
+        return dup.hasRemaining() ? dup : dup.flip();
+    }
+
+    static List<ByteBuffer> duplicate(List<ByteBuffer> item) {
+        int s = item.size();
+        switch (s) {
+            case 0:
+                return Collections.emptyList();
+            case 1:
+                return List.of(duplicate(item.get(0)));
+            case 2:
+                return List.of(duplicate(item.get(0)), duplicate(item.get(1)));
+            default: {
+                List<ByteBuffer> list = new ArrayList<>(s);
+                for (ByteBuffer byteBuffer : item) {
+                    list.add(duplicate(byteBuffer));
+                }
+
+                return Collections.unmodifiableList(list);
+            }
+        }
+    }
+}

src/main/java/io/github/nstdio/http/ext/ByteBufferInputStream.java

@@ -177,15 +177,8 @@ private void ensureOpen() throws IOException {
     }
 
     void add(ByteBuffer b) {
-        if (!closed) {
-            b = b.duplicate().asReadOnlyBuffer();
-            if (!b.hasRemaining()) {
-                b.flip();
-            }
-
-            if (b.hasRemaining()) {
-                buffers.offer(b);
-            }
+        if (!closed && (b = Buffers.duplicate(b)).hasRemaining()) {
+            buffers.offer(b);
         }
     }
 
@@ -197,6 +190,7 @@ List<ByteBuffer> drainToList() {
 
         var l = List.copyOf(buffs);
         buffs.clear();
+        mark = null;
 
         return l;
     }

src/main/java/io/github/nstdio/http/ext/CacheEntryMetadata.java

@@ -36,7 +36,7 @@
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
-public class CacheEntryMetadata {
+public final class CacheEntryMetadata {
     private final HttpRequest request;
     private final Clock clock;
 

src/main/java/io/github/nstdio/http/ext/CachingBodySubscriber.java

@@ -18,8 +18,6 @@
 
 import java.net.http.HttpResponse.BodySubscriber;
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Flow;
@@ -58,22 +56,7 @@ public void onSubscribe(Flow.Subscription subscription) {
     @Override
     public void onNext(List<ByteBuffer> item) {
         cachingSub.onNext(item);
-        originalSub.onNext(dup(item));
-    }
-
-    private List<ByteBuffer> dup(List<ByteBuffer> item) {
-        List<ByteBuffer> list = new ArrayList<>(item.size());
-
-        for (ByteBuffer byteBuffer : item) {
-            ByteBuffer duplicate = byteBuffer.duplicate();
-            if (!duplicate.hasRemaining()) {
-                duplicate.flip();
-            }
-
-            list.add(duplicate);
-        }
-
-        return Collections.unmodifiableList(list);
+        originalSub.onNext(Buffers.duplicate(item));
     }
 
     @Override

src/main/java/io/github/nstdio/http/ext/PathReadingSubscription.java

@@ -23,13 +23,14 @@
 import java.nio.channels.ReadableByteChannel;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.Flow.Subscriber;
 import java.util.concurrent.Flow.Subscription;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 class PathReadingSubscription implements Subscription {
-    public static final int DEFAULT_BUFF_CAPACITY = 8192;
+    private static final int DEFAULT_BUFF_CAPACITY = 1 << 14;
     private final Subscriber<List<ByteBuffer>> subscriber;
     private final AtomicBoolean completed = new AtomicBoolean(false);
     private final Path path;
@@ -56,15 +57,17 @@ public void request(long n) {
                 channel = Files.newByteChannel(path);
             }
 
+            var chan = channel;
+            var sub = subscriber;
             while (n-- > 0) {
                 ByteBuffer buff = ByteBuffer.allocate(DEFAULT_BUFF_CAPACITY);
-                int r = channel.read(buff);
-                if (r > 0) {
+                int r = chan.read(buff);
+                if (r != -1) {
                     buff.flip();
-                    subscriber.onNext(List.of(buff));
+                    sub.onNext(Collections.singletonList(buff));
                 } else {
                     completed.set(true);
-                    subscriber.onComplete();
+                    sub.onComplete();
                     break;
                 }
             }

src/test/java/io/github/nstdio/http/ext/BuffersTest.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright (C) 2022 Edgar Asatryan
+ *
+ * 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 io.github.nstdio.http.ext;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.commons.lang3.RandomUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.nio.Buffer;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.stream.Stream;
+
+class BuffersTest {
+
+    static Stream<List<ByteBuffer>> listBuffersData() {
+        return Stream.of(
+                List.of(),
+                List.of(ByteBuffer.wrap("abcde".repeat(16).getBytes(UTF_8))),
+                List.of(ByteBuffer.wrap("ab".repeat(16).getBytes(UTF_8)), ByteBuffer.wrap("cd".repeat(16).getBytes(UTF_8))),
+                Helpers.toBuffers(RandomUtils.nextBytes(96), true)
+        );
+    }
+
+    @ParameterizedTest
+    @MethodSource("listBuffersData")
+    void shouldDuplicatedBufferList(List<ByteBuffer> buffers) {
+        //when
+        List<ByteBuffer> actual = Buffers.duplicate(buffers);
+
+        //then
+        assertThat(actual)
+                .isNotSameAs(buffers)
+                .hasSameSizeAs(buffers)
+                .allMatch(Buffer::isReadOnly, "Expecting duplicated buffet to be read-only")
+                .containsExactlyElementsOf(buffers);
+    }
+
+    @Test
+    void shouldDuplicatedSingleBuffer() {
+        //given
+        ByteBuffer buffer = ByteBuffer.wrap(RandomUtils.nextBytes(16));
+
+        //when
+        ByteBuffer actual = Buffers.duplicate(buffer);
+        buffer.get();
+
+        //then
+        assertThat(actual.position()).isZero();
+    }
+}

src/test/java/io/github/nstdio/http/ext/ByteBufferInputStreamTest.java

@@ -18,6 +18,7 @@
 
 import static io.github.nstdio.http.ext.Helpers.toBuffers;
 import static org.apache.commons.lang3.RandomUtils.nextBytes;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIOException;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -35,6 +36,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.List;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
@@ -199,4 +201,21 @@ void shouldSupportMark() {
         //given + when + then
         assertTrue(new ByteBufferInputStream().markSupported());
     }
+
+    @Test
+    void shouldDumpBuffersToList() throws IOException {
+        //given
+        var is = new ByteBufferInputStream();
+        List<ByteBuffer> buffers = toBuffers(nextBytes(96), false);
+        buffers.forEach(is::add);
+
+        //when
+        List<ByteBuffer> actual = is.drainToList();
+
+        //then
+        assertEquals(-1, is.read());
+        assertThat(actual)
+                .hasSameSizeAs(buffers)
+                .containsExactlyElementsOf(buffers);
+    }
 }

src/test/java/io/github/nstdio/http/ext/ExtendedHttpClientContract.java

@@ -503,18 +503,19 @@ default void shouldUpdateExistingCacheWithNoCacheProvided() throws Exception {
                         .withBody("abc: Updated")
                 )
         );
+        HttpRequest request = requestBuilder().build();
 
         //when + then
-        var r1 = send(requestBuilder().build());
+        var r1 = send(request);
         assertThat(r1).isNotCached().hasBody("abc");
 
         var r2 = send(requestBuilder().header(HEADER_CACHE_CONTROL, "no-cache").build());
         assertThat(r2).isNotCached().hasBody("abc: Updated");
 
-        awaitFor(() -> {
-            var r3 = send(requestBuilder().build());
-            assertThat(r3).isCached().hasBody("abc: Updated");
-        });
+        await().until(() -> cache().get(request) != null);
+
+        var r3 = send(request);
+        assertThat(r3).isCached().hasBody("abc: Updated");
     }
 
     @Test

src/test/java/io/github/nstdio/http/ext/PlainSubscription.java

@@ -67,7 +67,7 @@ private void request0(long n) {
         }
     }
 
-    void execute(Runnable cmd) {
+    private void execute(Runnable cmd) {
         executor.ifPresentOrElse(service -> service.execute(cmd), cmd);
     }
 
@@ -90,7 +90,7 @@ private void clean() {
             service.shutdown();
             try {
                 //noinspection ResultOfMethodCallIgnored
-                service.awaitTermination(1, TimeUnit.SECONDS);
+                service.awaitTermination(5, TimeUnit.SECONDS);
             } catch (InterruptedException e) {
                 // noop
             }