JDK 7: added example of TransferQueue

Author: wesleyegberto

java-7/README.md

@@ -0,0 +1,18 @@
+# Java 7
+
+## Features
+
+* Collections API:
+  * `TransferQueue<E>`
+    * extends `BlockingQueue<E>` interface
+    * implementated by `LinkedTransferQueue<E>`
+    * blocking queue in which producer may wait for consumers to receive elements
+    * it can publish an element and wait for consumers or just publish and forget
+    * it can use a timeout to wait for a consumer
+    * methods:
+      * `getWaitingConsumerCount()`: returns an estimate of the number of waiting consumer to receive elements via `BlockingQueue` `take` or timed `poll`.
+      * `hasWaitingConsumer()`: returns true if there is any waiting consumer.
+      * `transfer(E)`: send an element and wait for the consumer.
+      * `tryTransfer(E)`: send an element to a waiting consumer, return false immediately if there isn't any consumer.
+      * `tryTransfer(E, long, TimeUnit)`: like `tryTransfer` but with a timeout to wait for a consumer before returning false.
+

java-7/TransferQueueExample.java

@@ -0,0 +1,158 @@
+import java.util.Scanner;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedTransferQueue;
+import java.util.concurrent.TransferQueue;
+import java.util.concurrent.TimeUnit;
+
+public class TransferQueueExample {
+	public static void main(String[] args) throws Exception {
+		TransferQueue<String> queue = new LinkedTransferQueue<>();
+
+		var producer = new Producer(queue);
+		var consumer = new Consumer(queue);
+
+		new Thread(consumer).start();
+		new Thread(producer).start();
+
+		Thread.currentThread().join();
+	}
+}
+
+class Producer implements Runnable {
+	private TransferQueue<String> queue;
+	private Scanner input;
+
+	Producer(TransferQueue<String> queue) {
+		this.queue = queue;
+		this.input = new Scanner(System.in);
+	}
+
+	public void run() {
+		log("started");
+		while (true) {
+			log("=== Enter the number: 1 - wait; 2 - wait with timeout; 3 - transfer timedout");
+
+			var command = input.nextLine();
+			switch (command) {
+				case "1":
+					send("wait");
+					break;
+				case "2":
+					send("timeout");
+					break;
+				case "3":
+					send("transfer_timeout");
+					waitAndSendWithTimeout("command will not be received");
+					break;
+				default:
+					log("Invalid command");
+			}
+		}
+	}
+
+	private void send(String command) {
+		try {
+			log("Sending command");
+			queue.transfer(command);
+			log("Command sent and received");
+		} catch (Exception ex) {
+			log(ex.getMessage());
+		}
+	}
+
+	private void sendWithTimeout(String command) {
+		try {
+			boolean received = queue.tryTransfer(command, 3, TimeUnit.SECONDS);
+			log("Command sent, received: " + received);
+		} catch (Exception ex) {
+			log(ex.getMessage());
+		}
+	}
+
+	private void waitAndSendWithTimeout(String command) {
+		try {
+			log("Sleeping...");
+			Thread.sleep(1000);
+			log("Sending command waiting for 3s");
+			boolean received = queue.tryTransfer(command, 3, TimeUnit.SECONDS);
+			log("Command sent, received: " + received);
+		} catch (Exception ex) {
+			log(ex.getMessage());
+		}
+	}
+
+	private void log(String message) {
+		System.out.printf("[Producer] %s%n", message);
+	}
+}
+
+class Consumer implements Runnable {
+	private BlockingQueue<String> queue;
+
+	Consumer(BlockingQueue<String> queue) {
+		this.queue = queue;
+	}
+
+	public void run() {
+		log("started");
+		String command = waitIndefinitely();
+
+		while (true) {
+			switch (command) {
+				case "wait":
+					command = waitIndefinitely();
+					break;
+				case "timeout":
+					command = waitWithTimeout(5);
+					break;
+				case "transfer_timeout":
+					sleep();
+					command = waitWithTimeout(5);
+					break;
+				default:
+					log("Invalid command");
+					command = waitIndefinitely();
+			}
+		}
+	}
+
+	private String waitIndefinitely() {
+		log("Waiting...");
+		try {
+			String command = queue.take();
+			log("Command received: " + command);
+			return command;
+		} catch (Exception ex) {
+			log(ex.getMessage());
+			return null;
+		}
+	}
+
+	private String waitWithTimeout(long seconds) {
+		log("Waiting " + seconds + "s");
+		try {
+			String command = queue.poll(seconds, TimeUnit.SECONDS);
+
+			if (command == null) {
+				log("Timedout");
+				return "";
+			}
+
+			log("Command received: " + command);
+			return command;
+		} catch (Exception ex) {
+			log(ex.getMessage());
+			return null;
+		}
+	}
+
+	private void sleep() {
+		try {
+			Thread.sleep(5000);
+		} catch (InterruptedException ex) {}
+	}
+
+	private void log(String message) {
+		System.out.printf("[Consumer] %s%n", message);
+	}
+}