package com.twitter.mesos.scheduler.log.mesos;

import java.util.Iterator;
import java.util.List;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
import com.google.inject.Inject;

import org.apache.mesos.Log;

import com.twitter.common.base.Function;
import com.twitter.common.stats.Stats;

/**
 * A {@code Log} implementation backed by a true distributed log in mesos core.
 *
 * @author John Sirois
 */
public class MesosLog implements com.twitter.mesos.scheduler.log.Log {
  private final Log log;

  @Inject
  public MesosLog(Log log) {
    this.log = Preconditions.checkNotNull(log);
  }

  @Override
  public Stream open() {
    return new LogStream(log);
  }

  private static class LogStream implements com.twitter.mesos.scheduler.log.Log.Stream {
    private static class Vars {
      final AtomicLong readTimeouts = Stats.exportLong("scheduler_log_native_read_timeouts");
      final AtomicLong readFailures = Stats.exportLong("scheduler_log_native_read_failures");
      final AtomicLong appendTimeouts = Stats.exportLong("scheduler_log_native_append_timeouts");
      final AtomicLong appendFailures = Stats.exportLong("scheduler_log_native_append_failures");
      final AtomicLong truncateTimeouts =
          Stats.exportLong("scheduler_log_native_truncate_timeouts");
      final AtomicLong truncateFailures =
          Stats.exportLong("scheduler_log_native_truncate_failures");
      final AtomicLong invalidatedWriters =
          Stats.exportLong("scheduler_log_native_invalidated_writers");
    }
    private final Vars vars = new Vars();

    private final Log log;
    private final Log.Reader reader;
    private Log.Writer writer;

    LogStream(Log log) {
      this.log = log;
      this.reader = new Log.Reader(log);
    }

    private static final Function<Log.Entry, LogEntry> MESOS_ENTRY_TO_ENTRY =
        new Function<Log.Entry, LogEntry>() {
          @Override public LogEntry apply(Log.Entry entry) {
            return new LogEntry(entry);
          }
        };

    @Override
    public Iterator<Entry> readFrom(com.twitter.mesos.scheduler.log.Log.Position position)
        throws StreamAccessException {

      Preconditions.checkArgument(position instanceof LogPosition);

      Log.Position from = ((LogPosition) position).unwrap();
      Log.Position to = end().unwrap();
      try {
        List<Log.Entry> entries = reader.read(from, to);
        return Iterables.<Log.Entry, Entry>transform(entries, MESOS_ENTRY_TO_ENTRY).iterator();
      } catch (TimeoutException e) {
        vars.readTimeouts.getAndIncrement();
        throw new StreamAccessException("Timeout reading from log.", e);
      } catch (Log.OperationFailedException e) {
        vars.readFailures.getAndIncrement();
        throw new StreamAccessException("Problem reading from log", e);
      }
    }

    @Override
    public LogPosition append(byte[] contents)
        throws StreamAccessException {

      try {
        Log.Position position = writer().append(contents);
        return LogPosition.wrap(position);
      } catch (TimeoutException e) {
        vars.appendTimeouts.getAndIncrement();
        throw new StreamAccessException("Timeout appending entry to the log.", e);
      } catch (Log.WriterFailedException e) {
        vars.appendFailures.getAndIncrement();

        // We must throw away a writer on a write failure - this could be because of a coordinator
        // election in which case we must trigger a new election.
        invalidateWriter();

        throw new StreamAccessException("Problem appending entry to the log.", e);
      }
    }

    @Override
    public void truncateBefore(com.twitter.mesos.scheduler.log.Log.Position position)
        throws StreamAccessException {

      Preconditions.checkArgument(position instanceof LogPosition);

      Log.Position before = ((LogPosition) position).unwrap();
      try {
        writer().truncate(before);
      } catch (TimeoutException e) {
        vars.truncateTimeouts.getAndIncrement();
        throw new StreamAccessException("Timeout truncating log.", e);
      } catch (Log.WriterFailedException e) {
        vars.truncateFailures.getAndIncrement();

        // We must throw away a writer on a write failure - this could be because of a coordinator
        // election in which case we must trigger a new election.
        invalidateWriter();

        throw new StreamAccessException("Problem truncating log.", e);
      }
    }

    @Override
    public LogPosition beginning() {
      return LogPosition.wrap(reader.beginning());
    }

    @Override
    public LogPosition end() {
      return LogPosition.wrap(reader.ending());
    }

    @Override
    public LogPosition position(byte[] identity) {
      return LogPosition.wrap(log.position(identity));
    }

    @Override
    public void close() {
      // noop
    }

    private synchronized Log.Writer writer() {
      if (writer == null) {
        writer = new Log.Writer(log);
      }
      return writer;
    }

    private synchronized void invalidateWriter() {
      vars.invalidatedWriters.getAndIncrement();
      writer = null;
    }

    private static class LogPosition implements com.twitter.mesos.scheduler.log.Log.Position {
      static LogPosition wrap(Log.Position position) {
        return new LogPosition(position);
      }

      private final Log.Position underlying;

      LogPosition(Log.Position underlying) {
        this.underlying = underlying;
      }

      Log.Position unwrap() {
        return underlying;
      }

      @Override public byte[] identity() {
        return underlying.identity();
      }

      @Override public int compareTo(Position o) {
        Preconditions.checkArgument(o instanceof LogPosition);
        return underlying.compareTo(((LogPosition) o).underlying);
      }
    }

    private static class LogEntry implements com.twitter.mesos.scheduler.log.Log.Entry {
      private final Log.Entry underlying;

      public LogEntry(Log.Entry entry) {
        this.underlying = entry;
      }

      @Override
      public LogPosition position() {
        return LogPosition.wrap(underlying.position);
      }

      @Override
      public byte[] contents() {
        return underlying.data;
      }
    }
  }
}

@@ -2154,16 +2154,29 @@
  public 
+synchronized 
 Iterator
@@ -2192,16 +2192,25 @@
 eadFrom(
+%0A        
 com.twit
@@ -2251,32 +2251,24 @@
 on position)
-%0A       
  throws Stre
@@ -2976,32 +2976,45 @@
 ride%0A    public 
+synchronized 
 LogPosition appe
@@ -3761,24 +3761,37 @@
 %0A    public 
+synchronized 
 void truncat
@@ -4622,32 +4622,45 @@
 ride%0A    public 
+synchronized 
 LogPosition begi
@@ -4744,32 +4744,45 @@
 ride%0A    public 
+synchronized 
 LogPosition end(
@@ -4857,32 +4857,45 @@
 ride%0A    public 
+synchronized 
 LogPosition posi
@@ -5053,29 +5053,16 @@
 private 
-synchronized 
 Log.Writ
@@ -5169,32 +5169,32 @@
  writer;%0A    %7D%0A%0A
+
     private sync
@@ -5189,29 +5189,16 @@
 private 
-synchronized 
 void inv

Synchronize MesosLog operations - Coordinator is not re-entrant and we use one from 2 threads (timeds snapshots, scheduler)