The code here will be directly pasted from this project. Quick summary: it is related to a Stack Overflow question.

import javax.annotation.Nonnull;
import java.io.IOException;
import java.util.PriorityQueue;
import java.util.concurrent.CountDownLatch;
/**
 * A waiter on a number of available characters in a {@link TextDecoder}
 *
 * <p>When it is woken up, it will check for the status of the operation; it
 * will throw a {@link RuntimeException} if the decoding operation fails, or it
 * has waited to more characters than what is actually available.</p>
 *
 * <p>It implements {@link Comparable} since instances of this class are used in
 * a {@link PriorityQueue}.</p>
 *
 * <p>Inspired from <a href="http://stackoverflow.com/a/22055231/1093528">this
 * StackOverflow answer</a>.</p>
 *
 * @see DecodingStatus
 * @see TextDecoder#needChars(int)
 */
final class CharWaiter
 implements Comparable<CharWaiter>
{
 private final int required;
 private final CountDownLatch latch = new CountDownLatch(1);
 private int nrChars = 0;
 private IOException exception = null;
 CharWaiter(final int required)
 {
 if (required < 0)
 throw new ArrayIndexOutOfBoundsException(required);
 this.required = required;
 }
 void setNrChars(final int nrChars)
 {
 this.nrChars = nrChars;
 }
 void setException(final IOException exception)
 {
 this.exception = exception;
 }
 int getRequired()
 {
 return required;
 }
 void await()
 throws InterruptedException
 {
 latch.await();
 if (exception != null)
 throw new RuntimeException("decoding error", exception);
 if (nrChars < required)
 throw new ArrayIndexOutOfBoundsException(required);
 }
 void wakeUp()
 {
 latch.countDown();
 }
 @Override
 public int compareTo(@Nonnull final CharWaiter o)
 {
 return Integer.compare(required, o.required);
 }
 @Override
 public String toString()
 {
 return "waiting for " + required + " character(s)";
 }
}

The code here will be directly pasted from this project. Quick summary: it is related to a Stack Overflow question.

import javax.annotation.Nonnull;
import java.io.IOException;
import java.util.PriorityQueue;
import java.util.concurrent.CountDownLatch;
/**
 * A waiter on a number of available characters in a {@link TextDecoder}
 *
 * <p>When it is woken up, it will check for the status of the operation; it
 * will throw a {@link RuntimeException} if the decoding operation fails, or it
 * has waited to more characters than what is actually available.</p>
 *
 * <p>It implements {@link Comparable} since instances of this class are used in
 * a {@link PriorityQueue}.</p>
 *
 * <p>Inspired from <a href="https://stackoverflow.com/a/22055231/1093528">this
 * StackOverflow answer</a>.</p>
 *
 * @see DecodingStatus
 * @see TextDecoder#needChars(int)
 */
final class CharWaiter
 implements Comparable<CharWaiter>
{
 private final int required;
 private final CountDownLatch latch = new CountDownLatch(1);
 private int nrChars = 0;
 private IOException exception = null;
 CharWaiter(final int required)
 {
 if (required < 0)
 throw new ArrayIndexOutOfBoundsException(required);
 this.required = required;
 }
 void setNrChars(final int nrChars)
 {
 this.nrChars = nrChars;
 }
 void setException(final IOException exception)
 {
 this.exception = exception;
 }
 int getRequired()
 {
 return required;
 }
 void await()
 throws InterruptedException
 {
 latch.await();
 if (exception != null)
 throw new RuntimeException("decoding error", exception);
 if (nrChars < required)
 throw new ArrayIndexOutOfBoundsException(required);
 }
 void wakeUp()
 {
 latch.countDown();
 }
 @Override
 public int compareTo(@Nonnull final CharWaiter o)
 {
 return Integer.compare(required, o.required);
 }
 @Override
 public String toString()
 {
 return "waiting for " + required + " character(s)";
 }
}

import javax.annotation.concurrent.ThreadSafe;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
/**
 * The watchdog class for a text decoding operation
 *
 * <p>This class takes care of {@link CharWaiter}s and callers to {@link
 * TextDecoder#getTotalChars()}.</p>
 *
 * <p>The decoding process in {@link TextDecoder} will update the internal
 * status of this object when the decoding operation makes progress; on an
 * update, this class will wake up the relevant waiters.</p>
 *
 * <p>In the event of an error, all waiters are woken up.</p>
 *
 * @see CharWaiter
 */
@ThreadSafe
final class DecodingStatus
{
 private boolean finished = false;
 private int nrChars = -1;
 private IOException exception = null;
 private final Queue<CharWaiter> waiters = new PriorityQueue<>();
 private final CountDownLatch endLatch = new CountDownLatch(1);
 synchronized boolean addWaiter(final CharWaiter waiter)
 {
 if (exception != null)
 throw new RuntimeException("decoding error", exception);
 final int required = waiter.getRequired();
 if (required <= nrChars)
 return false;
 if (!finished) {
 waiters.add(waiter);
 return true;
 }
 if (required > nrChars)
 throw new ArrayIndexOutOfBoundsException(required);
 return false;
 }
 synchronized void setNrChars(final int nrChars)
 {
 this.nrChars = nrChars;
 CharWaiter waiter;
 while (!waiters.isEmpty()) {
 waiter = waiters.peek();
 if (waiter.getRequired() > nrChars)
 break;
 waiter.setNrChars(nrChars);
 waiters.remove().wakeUp();
 }
 }
 synchronized void setFailed(final IOException exception)
 {
 this.exception = exception;
 final List<CharWaiter> list = new ArrayList<>(waiters);
 waiters.clear();
 for (final CharWaiter waiter: list) {
 waiter.setException(exception);
 waiter.wakeUp();
 }
 endLatch.countDown();
 }
 synchronized void setFinished(final int nrChars)
 {
 finished = true;
 this.nrChars = nrChars;
 final List<CharWaiter> list = new ArrayList<>(waiters);
 waiters.clear();
 for (final CharWaiter waiter: list) {
 waiter.setNrChars(nrChars);
 waiter.wakeUp();
 }
 endLatch.countDown();
 }
 int getTotalSize()
 {
 try {
 endLatch.await();
 } catch (InterruptedException e) {
 throw new RuntimeException("interrupted", e);
 }
 if (exception != null)
 throw new RuntimeException("decoding error", exception);
 return nrChars;
 }
 @Override
 public synchronized String toString()
 {
 if (exception != null)
 return "decoding error after reading " + nrChars + " character(s)";
 return "currently decoded: " + nrChars + " character(s); finished: "
 + finished;
 }
}

import javax.annotation.concurrent.ThreadSafe;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
/**
 * The watchdog class for a text decoding operation
 *
 * <p>This class takes care of {@link CharWaiter}s and callers to {@link
 * TextDecoder#getTotalChars()}.</p>
 *
 * <p>The decoding process in {@link TextDecoder} will update the internal
 * status of this object when the decoding operation makes progress; on an
 * update, this class will wake up the relevant waiters.</p>
 *
 * <p>In the event of an error, all waiters are woken up.</p>
 *
 * @see CharWaiter
 */
@ThreadSafe
final class DecodingStatus
{
 private boolean finished = false;
 private int nrChars = -1;
 private IOException exception = null;
 private final Queue<CharWaiter> waiters = new PriorityQueue<>();
 private final CountDownLatch endLatch = new CountDownLatch(1);
 synchronized boolean addWaiter(final CharWaiter waiter)
 {
 if (exception != null)
 throw new RuntimeException("decoding error", exception);
 final int required = waiter.getRequired();
 if (required <= nrChars)
 return false;
 if (!finished) {
 waiters.add(waiter);
 return true;
 }
 if (required > nrChars)
 throw new ArrayIndexOutOfBoundsException(required);
 return false;
 }
 synchronized void setNrChars(final int nrChars)
 {
 this.nrChars = nrChars;
 CharWaiter waiter;
 while (!waiters.isEmpty()) {
 waiter = waiters.peek();
 if (waiter.getRequired() > nrChars)
 break;
 waiter.setNrChars(nrChars);
 waiters.remove().wakeUp();
 }
 }
 synchronized void setFailed(final IOException exception)
 {
 this.exception = exception;
 final List<CharWaiter> list = new ArrayList<>(waiters);
 waiters.clear();
 for (final CharWaiter waiter: list) {
 waiter.setException(exception);
 waiter.wakeUp();
 }
 endLatch.countDown();
 }
 synchronized void setFinished(final int nrChars)
 {
 finished = true;
 this.nrChars = nrChars;
 final List<CharWaiter> list = new ArrayList<>(waiters);
 waiters.clear();
 for (final CharWaiter waiter: list) {
 waiter.setNrChars(nrChars);
 waiter.wakeUp();
 }
 endLatch.countDown();
 }
 int getTotalSize()
 {
 try {
 endLatch.await();
 } catch (InterruptedException e) {
 Thread.currentThread().interrupt();
 throw new RuntimeException("interrupted", e);
 }
 if (exception != null)
 throw new RuntimeException("decoding error", exception);
 return nrChars;
 }
 @Override
 public synchronized String toString()
 {
 if (exception != null)
 return "decoding error after reading " + nrChars + " character(s)";
 return "currently decoded: " + nrChars + " character(s); finished: "
 + finished;
 }
}

The code here will be directly pasted from this project. Quick summary: it is related to a StackOverflow question.

The code here will be directly pasted from this project. Quick summary: it is related to a Stack Overflow question.