package net.pixelcop.sewer.sink.debug;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;

import net.pixelcop.sewer.DrainSink;
import net.pixelcop.sewer.Event;
import net.pixelcop.sewer.Sink;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@DrainSink
public class CountingSink extends Sink {

  private static final Logger LOG = LoggerFactory.getLogger(CountingSink.class);

  private static final AtomicInteger appendCount = new AtomicInteger();
  private static final AtomicInteger openCount = new AtomicInteger();
  private static final AtomicInteger closeCount = new AtomicInteger();

  public CountingSink(String[] args) {
  }

  @Override
  public void close() throws IOException {
    LOG.debug("close");
    synchronized (closeCount) {
      closeCount.incrementAndGet();
    }
    setStatus(CLOSED);
  }

  @Override
  public void append(Event event) throws IOException {
    synchronized (appendCount) {
      appendCount.incrementAndGet();
    }
  }

  @Override
  public void open() throws IOException {
    LOG.debug("open");
    synchronized (openCount) {
      openCount.incrementAndGet();
    }
    setStatus(FLOWING);
  }

  /**
   * Gets the number of events appended to the sink
   *
   * @return int Append count
   */
  public static int getAppendCount() {
    return appendCount.get();
  }

  /**
   * Gets the open count
   * @return
   */
  public static int getOpenCount() {
    return openCount.get();
  }

  /**
   * Get the close count
   * @return
   */
  public static int getCloseCount() {
    return closeCount.get();
  }

  /**
   * Reset counters
   */
  public static void reset() {
    appendCount.set(0);
    openCount.set(0);
    closeCount.set(0);
  }

}

@@ -754,37 +754,83 @@
    s
-ynchronized (closeCount) %7B%0A  
+etStatus(CLOSING);%0A    if (subSink != null) %7B%0A      subSink.close();%0A    %7D%0A
     
@@ -851,38 +851,32 @@
 rementAndGet();%0A
-    %7D%0A
     setStatus(CL
@@ -962,71 +962,92 @@
     
-synchronized (appendCount) %7B%0A      appendCount.incrementAndGet(
+appendCount.incrementAndGet();%0A    if (subSink != null) %7B%0A      subSink.append(event
 );%0A 
@@ -1142,36 +1142,82 @@
    s
-ynchronized (openCount) %7B%0A  
+etStatus(OPENING);%0A    if (createSubSink()) %7B%0A      subSink.open();%0A    %7D%0A
     
@@ -1241,30 +1241,24 @@
 ntAndGet();%0A
-    %7D%0A
     setStatu

support appending to a configured subsink