// Path: blog-code/src/main/java/pl/ciruk/blog/producerconsumer/concurrent/Consumer.java
// public class Consumer<T extends Queueable> implements Runnable {
// 	private BlockingQueue<T> queue;
// 	
// 	private Processable<T> processor;
// 	
// 	public Consumer(BlockingQueue<T> queue) {
// 		this(queue, Processable.EMPTY);
// 	}
// 	
// 	public Consumer(BlockingQueue<T> queue, Processable<T> processor) {
// 		this.queue = queue;
// 		this.processor = processor;
// 	}
// 
// 	@Override
// 	public void run() {
// 		
// 		try {
// 			consumeMessages();
// 		} catch (InterruptedException e) {
// 			throw new RuntimeException(e);
// 		}
// 	}
// 	
// 	private void consumeMessages() throws InterruptedException {
// 		while (true) {
// 			T message = queue.poll(1, TimeUnit.SECONDS);
// 			if (Queueables.isLastMessage(message)) {
// 				return;
// 			}
// 			
// 			processor.process(message);
// 		}
// 	}
// }
// 
// Path: blog-code/src/main/java/pl/ciruk/blog/producerconsumer/concurrent/Processable.java
// public interface Processable<T> {
// 	Processable EMPTY = new Processable<Object>() {
// 		@Override
// 		public void process(Object element) {
// 			// Nothing to do, just visiting
// 		}
// 		
// 	};
// 	
// 	void process(T element);
// }
// 
// Path: blog-code/src/main/java/pl/ciruk/blog/producerconsumer/concurrent/Producer.java
// public class Producer<T extends Queueable> {
// 	private BlockingQueue<T> queue;
// 	
// 	public Producer(BlockingQueue<T> queue) {
// 		this.queue = queue;
// 	}
// 	
// 	public void send(T message) {
// 		try {
// 			queue.put(message);
// 		} catch (InterruptedException e) {
// 			throw new RuntimeException(e);
// 		}
// 	}
// }


import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import pl.ciruk.blog.producerconsumer.concurrent.Consumer;
import pl.ciruk.blog.producerconsumer.concurrent.Processable;
import pl.ciruk.blog.producerconsumer.concurrent.Producer;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
package pl.ciruk.blog.producerconsumer.web;




public class PagesProcessor {
	Map<String, Producer<Page>> producers = new HashMap<>();
	
	ExecutorService executor = Executors.newFixedThreadPool(4);
	
	public void createProducer(final String producerId) {
		BlockingQueue<Page> queue = new LinkedBlockingQueue<Page>();
		
		Producer<Page> producer = new Producer<>(queue);
		producers.put(producerId, producer);
		
// Path: blog-code/src/main/java/pl/ciruk/blog/producerconsumer/concurrent/Consumer.java
// public class Consumer<T extends Queueable> implements Runnable {
// 	private BlockingQueue<T> queue;
// 	
// 	private Processable<T> processor;
// 	
// 	public Consumer(BlockingQueue<T> queue) {
// 		this(queue, Processable.EMPTY);
// 	}
// 	
// 	public Consumer(BlockingQueue<T> queue, Processable<T> processor) {
// 		this.queue = queue;
// 		this.processor = processor;
// 	}
// 
// 	@Override
// 	public void run() {
// 		
// 		try {
// 			consumeMessages();
// 		} catch (InterruptedException e) {
// 			throw new RuntimeException(e);
// 		}
// 	}
// 	
// 	private void consumeMessages() throws InterruptedException {
// 		while (true) {
// 			T message = queue.poll(1, TimeUnit.SECONDS);
// 			if (Queueables.isLastMessage(message)) {
// 				return;
// 			}
// 			
// 			processor.process(message);
// 		}
// 	}
// }
// 
// Path: blog-code/src/main/java/pl/ciruk/blog/producerconsumer/concurrent/Processable.java
// public interface Processable<T> {
// 	Processable EMPTY = new Processable<Object>() {
// 		@Override
// 		public void process(Object element) {
// 			// Nothing to do, just visiting
// 		}
// 		
// 	};
// 	
// 	void process(T element);
// }
// 
// Path: blog-code/src/main/java/pl/ciruk/blog/producerconsumer/concurrent/Producer.java
// public class Producer<T extends Queueable> {
// 	private BlockingQueue<T> queue;
// 	
// 	public Producer(BlockingQueue<T> queue) {
// 		this.queue = queue;
// 	}
// 	
// 	public void send(T message) {
// 		try {
// 			queue.put(message);
// 		} catch (InterruptedException e) {
// 			throw new RuntimeException(e);
// 		}
// 	}
// }


// Path: blog-code/src/main/java/pl/ciruk/blog/producerconsumer/web/PagesProcessor.java
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import pl.ciruk.blog.producerconsumer.concurrent.Consumer;
import pl.ciruk.blog.producerconsumer.concurrent.Processable;
import pl.ciruk.blog.producerconsumer.concurrent.Producer;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;

package pl.ciruk.blog.producerconsumer.web;




public class PagesProcessor {
	Map<String, Producer<Page>> producers = new HashMap<>();
	
	ExecutorService executor = Executors.newFixedThreadPool(4);
	
	public void createProducer(final String producerId) {
		BlockingQueue<Page> queue = new LinkedBlockingQueue<Page>();
		
		Producer<Page> producer = new Producer<>(queue);
		producers.put(producerId, producer);
		
		Consumer<Page> consumer = new Consumer<>(queue, new Processable<Page>() {