n1chre · June 28, 2017 18:31 · Jun 28, 2017 · Jun 28, 2017 · Jun 28, 2017
diff --git a/ProxyConsumer.java b/ProxyConsumer.java
@@ -13,6 +13,7 @@
  * To use this class, simply wrap your Consumer into this class.
  *
  * !!! TODO this class will leave a thread waiting if null isn't consumed after consuming relevant objects.
+ * Or call finishConsuming() when done. 
  * 
  * Created by fhrenic on 28/06/2017.
  */
@@ -57,6 +58,10 @@ public synchronized void accept(T element) {
 		}
 		notify();
 	}
+
+	public void finishConsuming() {
+		noMoreElements.set(true);
+	}
 
 	private synchronized void startProcessing() {
 		while (true) {

diff --git a/ProxyConsumer.java b/ProxyConsumer.java
@@ -11,6 +11,8 @@
  * much slower than produce(). That's why elements are queued and processed when they can be processed.
  * 
  * To use this class, simply wrap your Consumer into this class.
+ *
+ * !!! TODO this class will leave a thread waiting if null isn't consumed after consuming relevant objects.
  * 
  * Created by fhrenic on 28/06/2017.
  */

diff --git a/ProxyConsumer.java b/ProxyConsumer.java
@@ -0,0 +1,90 @@
+package hrucc;
+
+import java.util.LinkedList;
+import java.util.Queue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+
+/**
+ * This class should be used when some object is producing stuff for some consumer and consume() is
+ * much slower than produce(). That's why elements are queued and processed when they can be processed.
+ * 
+ * To use this class, simply wrap your Consumer into this class.
+ * 
+ * Created by fhrenic on 28/06/2017.
+ */
+public class ProxyConsumer<T> implements Consumer<T> {
+
+	public static void main(String[] args) throws InterruptedException {
+		ProxyConsumer<Integer> proxy = new ProxyConsumer<>(number -> {
+			try {
+				TimeUnit.SECONDS.sleep(1);
+			} catch (InterruptedException e) {
+				System.out.println("can't sleep");
+				return;
+			}
+			System.out.println("got number " + number);
+		});
+
+		for (int i = 0; i < 10; i++) {
+			proxy.accept(i);
+		}
+		TimeUnit.SECONDS.sleep(20); // wait here
+		proxy.accept(420);
+		TimeUnit.SECONDS.sleep(5);
+		proxy.accept(null);
+	}
+
+	private final Consumer<T> consumer;
+	private final Queue<T> elements;
+	private AtomicBoolean noMoreElements;
+
+	public ProxyConsumer(Consumer<T> consumer) {
+		this.consumer = consumer;
+		elements = new LinkedList<>();
+		noMoreElements = new AtomicBoolean(false);
+		new Thread(this::startProcessing).start();
+	}
+
+	public synchronized void accept(T element) {
+		if (element != null) {
+			add(element);
+		} else {
+			noMoreElements.set(true);
+		}
+		notify();
+	}
+
+	private synchronized void startProcessing() {
+		while (true) {
+			while (hasElements()) {
+				consumer.accept(take());
+			}
+			if (noMoreElements.get()) {
+				return; // notify but no elements added
+			}
+			try {
+				wait();
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+				return;
+			}
+		}
+	}
+
+	// queue operations
+
+	private synchronized void add(T element) {
+		elements.add(element);
+	}
+
+	private synchronized boolean hasElements() {
+		return !elements.isEmpty();
+	}
+
+	private synchronized T take() {
+		return elements.poll();
+	}
+
+}