// Path: src/main/java/net/pixelcop/sewer/Sink.java
// public abstract class Sink extends Plumbing {
// 
//   private static final Logger LOG = LoggerFactory.getLogger(Sink.class);
// 
//   protected Sink subSink = null;
// 
//   /**
//    * Append an Event to the Sink
//    * @param event
//    * @throws IOException
//    */
//   public abstract void append(Event event) throws IOException;
// 
//   public void setSubSink(Sink subSink) {
//     this.subSink = subSink;
//   }
// 
//   public Sink getSubSink() {
//     return subSink;
//   }
// 
//   /**
//    * Create the SubSink, if possible
//    *
//    * @return True if a SubSink was created, false otherwise.
//    */
//   public boolean createSubSink() {
//     if (getSinkFactory() == null) {
//       return false;
//     }
// 
//     subSink = getSinkFactory().build();
//     return true;
//   }
// 
//   @Override
//   protected void finalize() throws Throwable {
//     if (getStatus() != CLOSED) {
//       try {
//         close();
//       } catch (Throwable t) {
//         LOG.warn("Caught during finalizer close(): " + t.getMessage(), t);
//         // rethrow? this should be good enough since the object was disposed of anyway
//       }
//     }
//   }
// 
// }
// 
// Path: src/main/java/net/pixelcop/sewer/util/BackoffHelper.java
// public class BackoffHelper {
// 
//   private int failures;
//   private Stopwatch stopwatch;
// 
//   public BackoffHelper() {
//     failures = 0;
//     stopwatch = new Stopwatch();
//     stopwatch.start();
//   }
// 
//   /**
//    * Increments failure count, prints a diagnostic message and sleeps for some amount of time
//    * based on the number of failures that have occurred. Max sleep time is 30 seconds.
//    *
//    * @param t
//    * @param log
//    * @param msg
//    * @param retryCanceled
//    * @throws InterruptedException
//    */
//   public void handleFailure(Throwable t, Logger log, String msg, boolean retryCanceled) throws InterruptedException {
// 
//     failures++;
//     if (log.isWarnEnabled()) {
//       log.warn(msg + ", failures = " + failures + " (" + message(t) + ")", t);
//     }
// 
//     if (retryCanceled) {
//       return;
//     }
// 
//     int backoff = 5000;
//     if (failures > 30) {
//       backoff = 10000;
// 
//     } else if (failures > 100) {
//       backoff = 30000;
//     }
// 
//     log.info("Sleeping for " + backoff + "ms before retry");
//     Thread.sleep(backoff);
//   }
// 
//   private String message(Throwable t) {
//     if (t.getMessage() != null) {
//       return t.getMessage();
//     }
//     return t.getClass().getCanonicalName();
//   }
// 
//   /**
//    * After successfully completing the action, print a diagnostic message if necessary
//    *
//    * @param log
//    * @param msg
//    */
//   public void resolve(Logger log, String msg) {
// 
//     if (!log.isDebugEnabled() && (failures == 0 || !log.isInfoEnabled())) {
//       return;
//     }
// 
//     // will always be shown if debug is enabled
//     log.debug(msg + " after " + failures + " failures (" + getElapsedTime() + ")");
//   }
// 
//   public int getNumFailures() {
//     return failures;
//   }
// 
//   public String getElapsedTime() {
//     return stopwatch.toString();
//   }
// 
// }


import java.io.IOException;
import net.pixelcop.sewer.Sink;
import net.pixelcop.sewer.util.BackoffHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
package net.pixelcop.sewer.sink.durable;




/**
 * Asynchronously open the given Sink, retrying until it actually opens
 *
 * @author chetan
 *
 */
public class SinkOpenerThread extends Thread {

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

  private static final int RUNNING  = 1;
  private static final int CANCELED = -1;

  private int status = RUNNING;
  private Sink sink;
  private SinkOpenerEvents callback;

  public SinkOpenerThread(long id, Sink sink, SinkOpenerEvents callback) {
// Path: src/main/java/net/pixelcop/sewer/Sink.java
// public abstract class Sink extends Plumbing {
// 
//   private static final Logger LOG = LoggerFactory.getLogger(Sink.class);
// 
//   protected Sink subSink = null;
// 
//   /**
//    * Append an Event to the Sink
//    * @param event
//    * @throws IOException
//    */
//   public abstract void append(Event event) throws IOException;
// 
//   public void setSubSink(Sink subSink) {
//     this.subSink = subSink;
//   }
// 
//   public Sink getSubSink() {
//     return subSink;
//   }
// 
//   /**
//    * Create the SubSink, if possible
//    *
//    * @return True if a SubSink was created, false otherwise.
//    */
//   public boolean createSubSink() {
//     if (getSinkFactory() == null) {
//       return false;
//     }
// 
//     subSink = getSinkFactory().build();
//     return true;
//   }
// 
//   @Override
//   protected void finalize() throws Throwable {
//     if (getStatus() != CLOSED) {
//       try {
//         close();
//       } catch (Throwable t) {
//         LOG.warn("Caught during finalizer close(): " + t.getMessage(), t);
//         // rethrow? this should be good enough since the object was disposed of anyway
//       }
//     }
//   }
// 
// }
// 
// Path: src/main/java/net/pixelcop/sewer/util/BackoffHelper.java
// public class BackoffHelper {
// 
//   private int failures;
//   private Stopwatch stopwatch;
// 
//   public BackoffHelper() {
//     failures = 0;
//     stopwatch = new Stopwatch();
//     stopwatch.start();
//   }
// 
//   /**
//    * Increments failure count, prints a diagnostic message and sleeps for some amount of time
//    * based on the number of failures that have occurred. Max sleep time is 30 seconds.
//    *
//    * @param t
//    * @param log
//    * @param msg
//    * @param retryCanceled
//    * @throws InterruptedException
//    */
//   public void handleFailure(Throwable t, Logger log, String msg, boolean retryCanceled) throws InterruptedException {
// 
//     failures++;
//     if (log.isWarnEnabled()) {
//       log.warn(msg + ", failures = " + failures + " (" + message(t) + ")", t);
//     }
// 
//     if (retryCanceled) {
//       return;
//     }
// 
//     int backoff = 5000;
//     if (failures > 30) {
//       backoff = 10000;
// 
//     } else if (failures > 100) {
//       backoff = 30000;
//     }
// 
//     log.info("Sleeping for " + backoff + "ms before retry");
//     Thread.sleep(backoff);
//   }
// 
//   private String message(Throwable t) {
//     if (t.getMessage() != null) {
//       return t.getMessage();
//     }
//     return t.getClass().getCanonicalName();
//   }
// 
//   /**
//    * After successfully completing the action, print a diagnostic message if necessary
//    *
//    * @param log
//    * @param msg
//    */
//   public void resolve(Logger log, String msg) {
// 
//     if (!log.isDebugEnabled() && (failures == 0 || !log.isInfoEnabled())) {
//       return;
//     }
// 
//     // will always be shown if debug is enabled
//     log.debug(msg + " after " + failures + " failures (" + getElapsedTime() + ")");
//   }
// 
//   public int getNumFailures() {
//     return failures;
//   }
// 
//   public String getElapsedTime() {
//     return stopwatch.toString();
//   }
// 
// }


// Path: src/main/java/net/pixelcop/sewer/sink/durable/SinkOpenerThread.java
import java.io.IOException;
import net.pixelcop.sewer.Sink;
import net.pixelcop.sewer.util.BackoffHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

package net.pixelcop.sewer.sink.durable;




/**
 * Asynchronously open the given Sink, retrying until it actually opens
 *
 * @author chetan
 *
 */
public class SinkOpenerThread extends Thread {

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

  private static final int RUNNING  = 1;
  private static final int CANCELED = -1;

  private int status = RUNNING;
  private Sink sink;
  private SinkOpenerEvents callback;

  public SinkOpenerThread(long id, Sink sink, SinkOpenerEvents callback) {
    setName("SinkOpener " + id);