apply formatter

Author: khadiwala

.gitignore

@@ -20,3 +20,4 @@
 
 # virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
 hs_err_pid*
+/target/

src/main/java/com/ibm/async_util/iteration/AsyncChannel.java

@@ -24,35 +24,40 @@
 /**
  * An unbounded async multi-producer-single-consumer channel.
  *
- * <p>This class provides a channel abstraction that allows multiple senders to place values into
- * the channel synchronously, and a single consumer to consume values as they become available
- * asynchronously. You can construct an {@link AsyncChannel} with the static methods on {@link
- * AsyncChannels}.
- *
- * <p>This interface represents an <i> unbounded </i> queue, meaning there is no mechanism to notify
+ * <p>
+ * This class provides a channel abstraction that allows multiple senders to place values into the
+ * channel synchronously, and a single consumer to consume values as they become available
+ * asynchronously. You can construct an {@link AsyncChannel} with the static methods on
+ * {@link AsyncChannels}.
+ *
+ * <p>
+ * This interface represents an <i> unbounded </i> queue, meaning there is no mechanism to notify
  * the sender that the queue is "full" (nor is there a notion of the queue being full to begin
  * with). The channel will continue to accept values as fast as the senders can {@link #send} them,
  * regardless of the rate at which the values are being consumed. If senders produce a lot of values
  * much faster than the consumption rate, it will lead to an out of memory, so users are responsible
  * for enforcing that the channel does not grow too large. If you would like a channel abstraction
  * that provides backpressure, see {@link BoundedAsyncChannel}.
  *
- * <p>This channel can be terminated by someone calling {@link #terminate()}, it can be called by
+ * <p>
+ * This channel can be terminated by someone calling {@link #terminate()}, it can be called by
  * consumers or senders. It is strongly recommended that all instances of this class eventually be
- * terminated. Mose terminal operations on {@link AsyncIterator} return {@link
- * java.util.concurrent.CompletionStage CompletionStages} that whose stage will not complete until
- * the channel is terminated. After the channel is terminated, subsequent {@link #send}s are
+ * terminated. Mose terminal operations on {@link AsyncIterator} return
+ * {@link java.util.concurrent.CompletionStage CompletionStages} that whose stage will not complete
+ * until the channel is terminated. After the channel is terminated, subsequent {@link #send}s are
  * rejected, though consumers of the channel will still receive any values that were sent before the
  * termination.
  *
- * <p>Typically you'll want to use a channel when you have some "source" of items, and want to
- * consume them asynchronously as the become available. Some examples of sources could be a
- * collection of {@link java.util.concurrent.CompletionStage CompletionStages}, bytes off of a
- * socket, results produced by dedicated worker threads, etc. Suppose you had scenario where you had
- * many threads doing some CPU intensive computation, and you'd send their answers off to some
- * server somewhere one at a time.
+ * <p>
+ * Typically you'll want to use a channel when you have some "source" of items, and want to consume
+ * them asynchronously as the become available. Some examples of sources could be a collection of
+ * {@link java.util.concurrent.CompletionStage CompletionStages}, bytes off of a socket, results
+ * produced by dedicated worker threads, etc. Suppose you had scenario where you had many threads
+ * doing some CPU intensive computation, and you'd send their answers off to some server somewhere
+ * one at a time.
  *
- * <pre>{@code
+ * <pre>
+ * {@code
  * AsyncChannel<Integer> channel = AsyncChannels.unbounded();
  * for (i = 0; i < numThreads; i++) {
  *   // spawn threads that send results to channel
@@ -77,15 +82,18 @@
  * // terminate the channel, done computing
  * channel.terminate();
  *
- * }</pre>
+ * }
+ * </pre>
  *
- * <p>It is also convenient to use a channel to merge many {@link AsyncIterator}s together. Think if
- * we were the destination server in the previous example, and we had many compute servers sending
- * us numbers they were computing. If we used {@link AsyncIterator#concat} in the following example,
- * we would wait until we got all the work from the first iterator to move onto the next. With a
+ * <p>
+ * It is also convenient to use a channel to merge many {@link AsyncIterator}s together. Think if we
+ * were the destination server in the previous example, and we had many compute servers sending us
+ * numbers they were computing. If we used {@link AsyncIterator#concat} in the following example, we
+ * would wait until we got all the work from the first iterator to move onto the next. With a
  * channel we process each number as soon as it becomes available.
  *
- * <pre>{@code
+ * <pre>
+ * {@code
  * AsyncIterator<Integer> getNumbersFrom(ServerLocation ip);
  * AsyncChannel channel = AsyncChannels.unbounded();
  * futures = ips.stream()
@@ -106,16 +114,17 @@
  *  channel
  *    .forEach(num -> System.out.println(num))
  *    .thenAccept(ig -> System.out.println("finished getting all numbers")));
- * }</pre>
+ * }
+ * </pre>
  *
- * <p>A reminder, all topics addressed in the documentation of {@link AsyncIterator} apply to this
+ * <p>
+ * A reminder, all topics addressed in the documentation of {@link AsyncIterator} apply to this
  * interface as well. Most importantly this means:
  *
  * <ul>
- *   <li>Consumption of an AsyncIterator is <b> not </b> thread safe
- *   <li>Lazy methods on AsyncIterator like map/flatMap don't consume anything. Make sure you
- *       actually use a consumption operation somewhere, otherwise no one will ever read what was
- *       sent
+ * <li>Consumption of an AsyncIterator is <b> not </b> thread safe
+ * <li>Lazy methods on AsyncIterator like map/flatMap don't consume anything. Make sure you actually
+ * use a consumption operation somewhere, otherwise no one will ever read what was sent
  * </ul>
  *
  * @param <T> The type of the elements in this channel
@@ -126,37 +135,41 @@ public interface AsyncChannel<T> extends AsyncIterator<T> {
   /**
    * Sends a value into this channel that can be consumed via the {@link AsyncIterator} interface.
    *
-   * <p>This method is thread safe - multiple threads can send values into this channel
-   * concurrently. This channel is unbounded, so it will continue accept new items immediately and
-   * store them in memory until they can be consumed. If you are sending work faster than you can
-   * consume it, this can easily lead to an out of memory condition.
+   * <p>
+   * This method is thread safe - multiple threads can send values into this channel concurrently.
+   * This channel is unbounded, so it will continue accept new items immediately and store them in
+   * memory until they can be consumed. If you are sending work faster than you can consume it, this
+   * can easily lead to an out of memory condition.
    *
    * @param item the item to be sent into the channel
    * @return true if the item was accepted, false if it was rejected because the channel has already
-   *     been terminated
+   *         been terminated
    */
   boolean send(T item);
 
   /**
    * Terminates the channel, disabling {@link #send}.
    *
-   * <p>After the channel is terminated all subsequent sends will be rejected, returning false.
-   * After the consumer consumes whatever was sent before the terminate, the consumer will receive
-   * an end of iteration notification.
+   * <p>
+   * After the channel is terminated all subsequent sends will be rejected, returning false. After
+   * the consumer consumes whatever was sent before the terminate, the consumer will receive an end
+   * of iteration notification.
    *
-   * <p>This method is thread-safe, and can be called multiple times. An attempt to terminate after
+   * <p>
+   * This method is thread-safe, and can be called multiple times. An attempt to terminate after
    * termination has already occurred is a no-op.
    */
   void terminate();
 
   /**
    * Gets a result from the channel if there is one ready right now.
    *
-   * <p>This method consumes parts of the channel, so like the consumption methods on {@link
-   * AsyncIterator}, this method is not thread-safe should be used in a single threaded fashion.
-   * After {@link #terminate()} is called and all outstanding results are consumed, poll will always
-   * return empty. This method <b> should not </b> be used if there are null values in the channel.
-   * <br>
+   * <p>
+   * This method consumes parts of the channel, so like the consumption methods on
+   * {@link AsyncIterator}, this method is not thread-safe should be used in a single threaded
+   * fashion. After {@link #terminate()} is called and all outstanding results are consumed, poll
+   * will always return empty. This method <b> should not </b> be used if there are null values in
+   * the channel. <br>
    * Notice that the channel being closed is indistinguishable from the channel being transiently
    * empty. To discover that no more results will ever be available, you must use the normal means
    * on {@link AsyncIterator}: either calling {@link #nextFuture()} and seeing if the result
@@ -165,7 +178,7 @@ public interface AsyncChannel<T> extends AsyncIterator<T> {
    *
    * @throws NullPointerException if the polled result is null
    * @return a present T value if there was one immediately available in the channel, empty if the
-   *     channel is currently empty
+   *         channel is currently empty
    */
   Optional<T> poll();
 }

src/main/java/com/ibm/async_util/iteration/AsyncChannels.java

@@ -40,9 +40,10 @@ private AsyncChannels() {}
   /**
    * Creates an unbounded AsyncChannel.
    *
-   * <p>Sends on an unbounded channel always complete synchronously, and throttling must be managed
-   * by the senders to ensure senders don't get too far ahead of the consumer. {@link AsyncChannel}
-   * for details.
+   * <p>
+   * Sends on an unbounded channel always complete synchronously, and throttling must be managed by
+   * the senders to ensure senders don't get too far ahead of the consumer. {@link AsyncChannel} for
+   * details.
    *
    * @return an {@link AsyncChannel}
    */
@@ -53,7 +54,8 @@ public static <T> AsyncChannel<T> unbounded() {
   /**
    * Creates a bounded AsyncChannel.
    *
-   * <p>This channel can only accept one value at a time until it is consumed. It is useful when you
+   * <p>
+   * This channel can only accept one value at a time until it is consumed. It is useful when you
    * want to produce work potentially in parallel, but want to be throttled at the rate at which you
    * can consume this work. See {@link BoundedAsyncChannel} for details.
    *
@@ -70,11 +72,12 @@ public static <T> BoundedAsyncChannel<T> bounded() {
   /**
    * Creates a buffered AsyncChannel.
    *
-   * <p>This channel can accept up to {@code maxBuffer} values before the futures returned by send
+   * <p>
+   * This channel can accept up to {@code maxBuffer} values before the futures returned by send
    * become delayed. See {@link BoundedAsyncChannel} for details
    *
    * @param maxBuffer the maximum number of values that the channel will accept before applying
-   *     backpressure to senders
+   *        backpressure to senders
    * @param <T> the type of elements in the returned channel
    * @return a {@link BoundedAsyncChannel} with a buffer size of {@code maxBuffer} elements
    */
@@ -87,29 +90,33 @@ public static <T> BoundedAsyncChannel<T> buffered(final int maxBuffer) {
    * multi-producer single-consumer model. This implementation is Fair - if there are two
    * non-overlapping calls to send, the consumer will see the first call before the second.
    *
-   * <p>The approach is simple. There is a singly linked list of futures with a head and tail
-   * pointer, with the following invariants
+   * <p>
+   * The approach is simple. There is a singly linked list of futures with a head and tail pointer,
+   * with the following invariants
    * <li>There is always at least one node in the list
    *
-   *     <p>While the Channel is open:
+   * <p>
+   * While the Channel is open:
    * <li>After any call to send or nextFuture completes, there is exactly one uncompleted future in
-   *     the list
+   * the list
    * <li>After any call to send or nextFuture completes, the uncompleted future is pointed at by
-   *     head
+   * head
    *
-   *     <p>The list starts with an initial uncompleted future. When send is called, the sender
-   *     attempts to update the tail with a new uncompleted future via CAS. When it succeeds, it
-   *     completes the former tails future. Readers just return the future pointed at by head, when
-   *     that future completes the head pointer is moved forward.
+   * <p>
+   * The list starts with an initial uncompleted future. When send is called, the sender attempts to
+   * update the tail with a new uncompleted future via CAS. When it succeeds, it completes the
+   * former tails future. Readers just return the future pointed at by head, when that future
+   * completes the head pointer is moved forward.
    *
-   *     <p>Concretely, if there are values to read when the reader comes in, it simply observes an
-   *     already completed future. If there are no values to read, it will wait on the head future
-   *     (which is also the tail in this case), which will be completed by whichever sender updates
-   *     the tail.
+   * <p>
+   * Concretely, if there are values to read when the reader comes in, it simply observes an already
+   * completed future. If there are no values to read, it will wait on the head future (which is
+   * also the tail in this case), which will be completed by whichever sender updates the tail.
    *
-   *     <p>When a sender sends an Optional.empty, the tail is replaced with STOP marker. Future
-   *     senders see the tail and don't try to update it. Once the reader hits the empty tail, it
-   *     stops moving the head
+   * <p>
+   * When a sender sends an Optional.empty, the tail is replaced with STOP marker. Future senders
+   * see the tail and don't try to update it. Once the reader hits the empty tail, it stops moving
+   * the head
    *
    * @param <T>
    */
@@ -155,13 +162,12 @@ public CompletableFuture<Either<End, T>> nextFuture() {
       // whenever we get a value from the head future, we should unlink that node, and move the head
       // pointer forward. We know head.next must exist, because a node's next value is always
       // updated before completion.
-      return this.head.thenApply(
-          res -> {
-            // note: we don't need to check for exceptions here, because there is no way to send an
-            // exceptional result into a channel
-            this.head = this.head.next;
-            return res;
-          });
+      return this.head.thenApply(res -> {
+        // note: we don't need to check for exceptions here, because there is no way to send an
+        // exceptional result into a channel
+        this.head = this.head.next;
+        return res;
+      });
     }
 
     public Optional<T> poll() {
@@ -216,8 +222,9 @@ private static <T> Node<T> stopNode() {
    * Implementation is backed by an {@link AsyncSemaphore} which throttles the number of elements
    * that can be in the linked list in the backing {@link UnboundedChannel}.
    *
-   * <p>This implementation is Fair - if there are two non-overlapping calls to send, the consumer
-   * will see the first call before the second.
+   * <p>
+   * This implementation is Fair - if there are two non-overlapping calls to send, the consumer will
+   * see the first call before the second.
    *
    * @param <T>
    */
@@ -236,13 +243,14 @@ private static class BufferedChannel<T> implements BoundedAsyncChannel<T> {
     public CompletionStage<Either<End, T>> nextFuture() {
       return this.backingChannel
           .nextFuture()
-          .thenApply(
-              res -> {
-                // only need to release if the backing channel is open. after it is closed, senders will
-                // release automatically
-                res.forEach(ig -> {}, t -> this.sendThrottle.release());
-                return res;
-              });
+          .thenApply(res -> {
+            // only need to release if the backing channel is open. after it is closed, senders
+            // will
+            // release automatically
+            res.forEach(ig -> {
+            }, t -> this.sendThrottle.release());
+            return res;
+          });
     }
 
     @Override
@@ -252,31 +260,30 @@ public CompletionStage<Boolean> send(final T item) {
           // acquire a permit, this represents the node we put
           // in the queue in our underlying channel
           .acquire()
-          .thenApply(
-              ig -> {
-                final boolean accepted = this.backingChannel.send(item);
-                if (!accepted) {
-                  // the backing channel was closed, so our item will never be consumed. we should
-                  // release the permit we acquired
-                  this.sendThrottle.release();
-                }
-                return accepted;
-              });
+          .thenApply(ig -> {
+            final boolean accepted = this.backingChannel.send(item);
+            if (!accepted) {
+              // the backing channel was closed, so our item will never be consumed. we should
+              // release the permit we acquired
+              this.sendThrottle.release();
+            }
+            return accepted;
+          });
     }
 
     @Override
     public CompletionStage<Void> terminate() {
       // note we still want to respect the buffer here, fairness of our backing semaphore will
-      // ensure that any sends queued before the terminate will still hit the backingChannel before the
+      // ensure that any sends queued before the terminate will still hit the backingChannel before
+      // the
       // terminate does
       return this.sendThrottle
           .acquire()
-          .thenApply(
-              res -> {
-                this.backingChannel.terminate();
-                this.sendThrottle.release();
-                return res;
-              });
+          .thenApply(res -> {
+            this.backingChannel.terminate();
+            this.sendThrottle.release();
+            return res;
+          });
     }
 
     @Override