package org.shirdrn.storm.api.common;

import java.util.concurrent.BlockingQueue;

import backtype.storm.task.OutputCollector;
import backtype.storm.tuple.Tuple;

public class BoltQueuedDistributor<OUT> extends QueuedDistributor<Tuple, OutputCollector, OUT> {

	private static final long serialVersionUID = 1L;

	public BoltQueuedDistributor(OutputCollector collector) {
		super(collector);
	}
	
	public BoltQueuedDistributor(OutputCollector collector,  BlockingQueue<Tuple> queue) {
		super(collector);
	}

	@Override
	protected Thread newProcessorRunner() {
		return new ProcessorRunner();
	}

	private final class ProcessorRunner extends Thread {
		
		@Override
		public void run() {
			while(true) {
				Tuple input = null;
				try {
					input = queue.take();
					OUT output = processor.process(input);
					collector.emit(input, processor.writeOut(output));
					collector.ack(input);
				} catch (Exception e) {
					collector.fail(input);
				}
			}
		}
		
	}
}

@@ -758,16 +758,42 @@
 take();%0A
+%09%09%09%09%09if(input != null) %7B%0A%09
 %09%09%09%09%09OUT
@@ -824,24 +824,53 @@
 ess(input);%0A
+%09%09%09%09%09%09if(output != null) %7B%0A%09%09
 %09%09%09%09%09collect
@@ -913,16 +913,25 @@
 tput));%0A
+%09%09%09%09%09%09%7D%0A%09
 %09%09%09%09%09col
@@ -945,24 +945,31 @@
 ack(input);%0A
+%09%09%09%09%09%7D%0A
 %09%09%09%09%7D catch 

Optimize codes.