package com.asksunny.appgallaries.cdp;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

public class ConcurrentDataProcessor implements Runnable {

	private final AtomicBoolean endOfDataStream = new AtomicBoolean(false);
	private final AtomicLong numberOfRecords = new AtomicLong(0);
	private long processed = 0;
	private CountDownLatch readyLatch;
	private CyclicBarrier processBarrier;
	private ConcurrentLinkedQueue<String[]> datain = new ConcurrentLinkedQueue<String[]>();

	public ConcurrentDataProcessor(final CountDownLatch readyLatch, final CyclicBarrier processBarrier) {
		this.processBarrier = processBarrier;
		this.readyLatch = readyLatch;
	}

	public void init() {

	}

	public void run() {
		readyLatch.countDown();
		while (endOfDataStream.get() == false
				|| processed < numberOfRecords.get()) {
			if (processed < numberOfRecords.get()) {
				String[] data = datain.isEmpty() ? null : datain.remove();
				if (data != null) {
					// Process data here	
					System.out.println(data[0]);
					processed++;
					try {
						processBarrier.await();
					} catch (Exception e) {
						;
					} 
				}
				
			}

		}
	}

	public void sendData(String[] data) 
	{
		datain.add(data);
		numberOfRecords.incrementAndGet();
	}
	
	public void endOfDataStream()
	{
		this.endOfDataStream.set(true);
	}

}

@@ -37,60 +37,8 @@
 p;%0A%0A
-import java.util.concurrent.BrokenBarrierException;%0A
 impo
@@ -84,16 +84,16 @@
 dQueue;%0A
+
 import j
@@ -468,37 +468,9 @@
 (0);
-%0A%09private long processed = 0;
+%09
 %0A%09pr
@@ -931,27 +931,16 @@
 %0A%09%09%09%09%7C%7C 
-processed %3C
  numberO
@@ -949,24 +949,27 @@
 ecords.get()
+ %3E0
 ) %7B%0A%09%09%09if (p
@@ -971,19 +971,8 @@
 if (
-processed %3C
  num
@@ -989,16 +989,21 @@
 ds.get()
+ %3E 0 
 ) %7B%0A%09%09%09%09
@@ -1155,19 +1155,176 @@
 %09%09%09%09
-processed++
+//Simulate data process here%0A%09%09%09%09%09try %7B%0A%09%09%09%09%09%09Thread.sleep(1*1000);%0A%09%09%09%09%09%7D catch (InterruptedException e1) %7B%0A%09%09%09%09%09%09;%0A%09%09%09%09%09%7D%0A%09%09%09%09%09%0A%09%09%09%09%09numberOfRecords.decrementAndGet()
 ;%0A%09%09

clean up