package com.trendmicro.grid.acl.ds.bclog;
   
   import com.trendmicro.grid.acl.l0.datatypes.ProcessPackageDataSet;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import org.springframework.stereotype.Repository;
   
   import javax.annotation.Resource;
   import java.io.IOException;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.concurrent.ArrayBlockingQueue;
  
  /**
   * Queues datasets and writes them inside a separate writer thread that runs every second.
   *
   * @author juergen_kellerer, 2010-05-28
   * @version 1.0
   */
  @Repository
  public final class BinaryChangeLogQueue extends ArrayBlockingQueue<ProcessPackageDataSet> {
  
  	private static final Logger log = LoggerFactory.getLogger(BinaryChangeLogQueue.class);
  	private static final long serialVersionUID = 4904787490898023306L;
  
  	@Resource
  	BinaryChangeLogWriterService writerService;
  
  	private transient final Runnable writer = new Runnable() {
  		public void run() {
  			ProcessPackageDataSet first = null;
  
  			while ((first = poll()) != null) {
  				// Copy elements in batches to reduce the locking overhead of constant polling.
  				List<ProcessPackageDataSet> elements = new ArrayList<ProcessPackageDataSet>(25);
  				elements.add(first);
  				drainTo(elements, 24);
  
  				for (ProcessPackageDataSet element : elements)
  					try {
  						writerService.writeNext(element);
  					} catch (IOException e) {
  						log.error("TMACL-00650:Failed writing results of job '" +
  							element.getReferringJob() + "' to the binary changelog. " +
  							"Operations that depend on this data (e.g. multisite) will be outdated.", e);
  					}
  
  			}
  
  			synchronized (checkpoint) {
  				checkpoint.notifyAll();
  			}
  		}
  	};
  
  	final Object checkpoint = new Object();
  	final transient Thread executor;
  
  	public BinaryChangeLogQueue() {
  		super(250, false);
  
  		executor = new Thread() {
  			public void run() {
  				while (!isInterrupted()) {
  					try {
  						synchronized (this) {
  							wait(2000);
  						}
  
  						writer.run();
  					} catch (InterruptedException e) {
  						return;
  					} catch (Throwable e) {
  						log.error("TMACL-00700:Cought an unexpected exception on " +
  							"the attempt to write binary logs.", e);
  					}
  				}
  			}
  		};
  		executor.setName("Binary ChangeLog Writer");
  		executor.start();
  	}
  
  	/**
  	 * Ensures that all queued elements are written.
  	 *
  	 * @throws java.io.IOException  if the disk io failed.
  	 * @throws InterruptedException in cause of waiting for the queue to become empty got interrupted.
  	 */
  	public synchronized void checkpoint() throws InterruptedException, IOException {
  		// Notify the worker.
  		synchronized (executor) {
  			executor.notify();
  		}
  
  		// Wait for the signal
  		synchronized (checkpoint) {
  			checkpoint.wait();
  		}
 
 		writerService.fsync();
 	}
 }