package com.github.triceo.splitlog;
   
   import java.io.File;
   import java.lang.ref.WeakReference;
   import java.util.ArrayList;
   import java.util.List;
   import java.util.concurrent.TimeUnit;
   import java.util.concurrent.atomic.AtomicLong;
   
  import org.apache.commons.collections4.BidiMap;
  import org.apache.commons.collections4.bidimap.DualHashBidiMap;
  import org.apache.commons.lang3.tuple.ImmutablePair;
  import org.apache.commons.lang3.tuple.Pair;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.github.triceo.splitlog.api.Follower;
  import com.github.triceo.splitlog.api.LogWatch;
  import com.github.triceo.splitlog.api.Message;
  import com.github.triceo.splitlog.api.MessageConsumer;
  import com.github.triceo.splitlog.api.MessageDeliveryStatus;
  import com.github.triceo.splitlog.api.MessageListener;
  import com.github.triceo.splitlog.api.MessageMeasure;
  import com.github.triceo.splitlog.api.MessageMetric;
  import com.github.triceo.splitlog.api.MidDeliveryMessageCondition;
  import com.github.triceo.splitlog.api.SimpleMessageCondition;
  import com.github.triceo.splitlog.api.TailSplitter;
  
  /**
   * Default log watch implementation which provides all the bells and whistles so
   * that the rest of the tool can work together.
   *
   * The tailer thread will only be started after {@link #startFollowing()} or
   * {@link #startConsuming(MessageListener)} is called. Subsequently, it will
   * only be stopped after there no more running {@link Follower}s or
   * {@link MessageConsumer}s.
   */
  final class DefaultLogWatch implements LogWatch {
  
      private static final AtomicLong ID_GENERATOR = new AtomicLong(0);
      private static final Logger LOGGER = LoggerFactory.getLogger(DefaultLogWatch.class);
  
      private final MeasuringConsumerManager<LogWatch> consumers = new MeasuringConsumerManager<LogWatch>(this);
      private MessageBuilder currentlyProcessedMessage;
      private final BidiMap<String, MessageMeasure<? extends Number, Follower>> handingDown = new DualHashBidiMap<String, MessageMeasure<? extends Number, Follower>>();
      private boolean isTerminated = false;
      private WeakReference<Message> previousAcceptedMessage;
      private final TailSplitter splitter;
      private final LogWatchStorageManager storage;
      private final LogWatchSweepingManager sweeping;
      private final LogWatchTailingManager tailing;
      private final long uniqueId = DefaultLogWatch.ID_GENERATOR.getAndIncrement();
      private final File watchedFile;
  
      protected DefaultLogWatch(final File watchedFile, final TailSplitter splitter, final int capacity,
          final SimpleMessageCondition acceptanceCondition, final long delayBetweenReads,
          final long delayBetweenSweeps, final boolean ignoreExistingContent, final boolean reopenBetweenReads,
          final int bufferSize, final long delayForTailerStart) {
          this.splitter = splitter;
          this.storage = new LogWatchStorageManager(this, capacity, acceptanceCondition);
          this.tailing = new LogWatchTailingManager(this, delayBetweenReads, delayForTailerStart, ignoreExistingContent,
                  reopenBetweenReads, bufferSize);
          this.sweeping = new LogWatchSweepingManager(this.storage, delayBetweenSweeps);
          this.watchedFile = watchedFile;
      }
  
      synchronized void addLine(final String line) {
          final boolean isMessageBeingProcessed = this.currentlyProcessedMessage != null;
          if (this.splitter.isStartingLine(line)) {
              // new message begins
              if (isMessageBeingProcessed) { // finish old message
                  final Message completeMessage = this.handleCompleteMessage(this.currentlyProcessedMessage);
                  if (completeMessage != null) {
                      this.previousAcceptedMessage = new WeakReference<Message>(completeMessage);
                  }
              }
              // prepare for new message
              this.currentlyProcessedMessage = new MessageBuilder(line);
              if (this.previousAcceptedMessage != null) {
                  this.currentlyProcessedMessage.setPreviousMessage(this.previousAcceptedMessage.get());
              }
          } else {
              // continue present message
              if (!isMessageBeingProcessed) {
                  // most likely just a garbage immediately after start
                  return;
              }
              this.currentlyProcessedMessage.add(line);
          }
          this.handleIncomingMessage(this.currentlyProcessedMessage);
      }
  
      @Override
      public int countConsumers() {
          return this.consumers.countConsumers();
      }
  
      /**
       * <strong>This is not part of the public API.</strong> Purely for purposes
      * of testing the automated message sweep.
      *
      * @return How many messages there currently are in the internal message
      *         store.
      */
     int countMessagesInStorage() {
         return this.storage.getMessageStore().size();
     }
 
     @Override
     public int countMetrics() {
         return this.consumers.countMetrics();
     }
 
     /**
      * Return all messages that have been sent to a given {@link Follower}, from
      * its {@link #startFollowing()} until either its
      * {@link #stopFollowing(Follower)} or to this moment, whichever is
      * relevant.
      *
      * @param follower
      *            The follower in question.
      * @return Unmodifiable list of all the received messages, in the order
      *         received.
      */
     protected List<Message> getAllMessages(final Follower follower) {
         return this.storage.getAllMessages(follower);
     }
 
     @Override
     public MessageMetric<? extends Number, LogWatch> getMetric(final String id) {
         return this.consumers.getMetric(id);
     }
 
     @Override
     public String getMetricId(final MessageMetric<? extends Number, LogWatch> measure) {
         return this.consumers.getMetricId(measure);
     }
 
     public long getUniqueId() {
         return this.uniqueId;
     }
 
     @Override
     public File getWatchedFile() {
         return this.watchedFile;
     }
 
     /**
      * Notify {@link MessageConsumer}s of a message that is either
      * {@link MessageDeliveryStatus#ACCEPTED} or
      * {@link MessageDeliveryStatus#REJECTED}.
      *
      * @param messageBuilder
      *            Builder to use to construct the message.
      * @return The message that was the subject of notifications; null if
      *         rejected.
      */
     private synchronized Message handleCompleteMessage(final MessageBuilder messageBuilder) {
         final Message message = messageBuilder.buildFinal(this.splitter);
         final boolean messageAccepted = this.storage.registerMessage(message, this);
         final MessageDeliveryStatus status = messageAccepted ? MessageDeliveryStatus.ACCEPTED
                 : MessageDeliveryStatus.REJECTED;
         this.consumers.messageReceived(message, status, this);
         return messageAccepted ? message : null;
     }
 
     /**
      * Notify {@link MessageConsumer}s of a message that is
      * {@link MessageDeliveryStatus#INCOMING}.
      *
      * @param messageBuilder
      *            Builder to use to construct the message.
      * @return The message that was the subject of notifications.
      */
     private synchronized Message handleIncomingMessage(final MessageBuilder messageBuilder) {
         final Message message = messageBuilder.buildIntermediate(this.splitter);
         this.consumers.messageReceived(message, MessageDeliveryStatus.INCOMING, this);
         return message;
     }
 
     /**
      * Notify {@link Follower} of a message that could not be delivered fully as
      * the Follower terminated. Will not notify local consumers.
      *
      * @param follower
      *            The follower that was terminated.
      * @param messageBuilder
      *            Builder to use to construct the message.
      * @return The message that was the subject of notifications.
      */
     private synchronized Message handleUndeliveredMessage(final AbstractLogWatchFollower follower,
         final MessageBuilder messageBuilder) {
         final Message message = messageBuilder.buildIntermediate(this.splitter);
         follower.messageReceived(message, MessageDeliveryStatus.INCOMPLETE, this);
         return message;
     }
 
     @Override
     public boolean isConsuming(final MessageConsumer<LogWatch> consumer) {
         return this.consumers.isConsuming(consumer);
     }
 
     @Override
     public synchronized boolean isFollowedBy(final Follower follower) {
         return this.isConsuming(follower);
     }
 
     @Override
     public boolean isHandingDown(final MessageMeasure<? extends Number, Follower> measure) {
         return this.handingDown.containsValue(measure);
     }
 
     @Override
     public boolean isHandingDown(final String id) {
         return this.handingDown.containsKey(id);
     }
 
     @Override
     public boolean isMeasuring(final MessageMetric<? extends Number, LogWatch> metric) {
         return this.consumers.isMeasuring(metric);
     }
 
     @Override
     public boolean isMeasuring(final String id) {
         return this.consumers.isMeasuring(id);
     }
 
     @Override
     public synchronized boolean isTerminated() {
         return this.isTerminated;
     }
 
     @Override
     public synchronized MessageConsumer<LogWatch> startConsuming(final MessageListener<LogWatch> consumer) {
         final MessageConsumer<LogWatch> result = this.consumers.startConsuming(consumer);
         this.startTailingIfNecessary(false);
         return result;
     }
 
     @Override
     public synchronized Follower startFollowing() {
         final Follower f = this.startFollowingActually(false);
         this.tailing.waitUntilStarted();
         return f;
     }
 
     @Override
     public synchronized Pair<Follower, Message> startFollowing(final MidDeliveryMessageCondition<LogWatch> waitFor) {
         final Follower f = this.startFollowingActually(true);
         return ImmutablePair.of(f, f.waitFor(waitFor));
     }
 
     @Override
     public synchronized Pair<Follower, Message> startFollowing(final MidDeliveryMessageCondition<LogWatch> waitFor,
             final long howLong, final TimeUnit unit) {
         final Follower f = this.startFollowingActually(true);
         return ImmutablePair.of(f, f.waitFor(waitFor, howLong, unit));
     }
 
     /**
      * @param boolean If the tailer needs a delayed start because of
      *        {@link #startFollowing(MidDeliveryMessageCondition)}, as explained
      *        in {@link LogWatchBuilder#getDelayBeforeTailingStarts()}.
      * @return The follower that follows this log watch from now on.
      */
     private synchronized Follower startFollowingActually(final boolean needsToWait) {
         if (this.isTerminated()) {
             throw new IllegalStateException("Cannot start tailing on an already terminated LogWatch.");
         }
         this.sweeping.start();
         // assemble list of consumers to be handing down and then the follower
         final List<Pair<String, MessageMeasure<? extends Number, Follower>>> pairs = new ArrayList<Pair<String, MessageMeasure<? extends Number, Follower>>>();
         for (final BidiMap.Entry<String, MessageMeasure<? extends Number, Follower>> entry : this.handingDown
                 .entrySet()) {
             pairs.add(ImmutablePair.<String, MessageMeasure<? extends Number, Follower>> of(entry.getKey(),
                     entry.getValue()));
         }
         // register the follower
         final AbstractLogWatchFollower follower = new NonStoringFollower(this, pairs);
         this.consumers.registerConsumer(follower);
         this.storage.followerStarted(follower);
         DefaultLogWatch.LOGGER.info("Registered {} for {}.", follower, this);
         this.startTailingIfNecessary(needsToWait);
         return follower;
     }
 
     @Override
     public synchronized boolean startHandingDown(final MessageMeasure<? extends Number, Follower> measure,
         final String id) {
         if (this.isTerminated()) {
             throw new IllegalStateException("Log watch already terminated.");
         } else if (measure == null) {
             throw new IllegalArgumentException("Measure may not be null.");
         } else if (id == null) {
             throw new IllegalArgumentException("ID may not be null.");
         } else if (this.handingDown.containsKey(id) || this.handingDown.containsValue(measure)) {
             return false;
         }
         this.handingDown.put(id, measure);
         return true;
     }
 
     @Override
     public <T extends Number> MessageMetric<T, LogWatch> startMeasuring(final MessageMeasure<T, LogWatch> measure,
             final String id) {
         return this.consumers.startMeasuring(measure, id);
     }
 
     private synchronized void startTailingIfNecessary(final boolean needsToWait) {
         if (this.tailing.isRunning()) {
             return;
         }
         if (this.consumers.countConsumers() > 0) {
             // we have listeners, let's start tailing
             this.tailing.start(needsToWait);
         }
     }
 
     @Override
     public synchronized boolean stopConsuming(final MessageConsumer<LogWatch> consumer) {
         final boolean result = this.consumers.stopConsuming(consumer);
         this.stopTailingIfNecessary();
         return result;
     }
 
     @Override
     public synchronized boolean stopFollowing(final Follower follower) {
         if (!this.isFollowedBy(follower)) {
             return false;
         }
         if (this.currentlyProcessedMessage != null) {
             this.handleUndeliveredMessage((AbstractLogWatchFollower) follower, this.currentlyProcessedMessage);
         }
         this.consumers.stopConsuming(follower);
         this.storage.followerTerminated(follower);
         DefaultLogWatch.LOGGER.info("Unregistered {} for {}.", follower, this);
         this.stopTailingIfNecessary();
         return true;
     }
 
     @Override
     public synchronized boolean stopHandingDown(final MessageMeasure<? extends Number, Follower> measure) {
         return (this.handingDown.removeValue(measure) != null);
     }
 
     @Override
     public synchronized boolean stopHandingDown(final String id) {
         return (this.handingDown.remove(id) != null);
     }
 
     @Override
     public boolean stopMeasuring(final MessageMetric<? extends Number, LogWatch> metric) {
         return this.consumers.stopMeasuring(metric);
     }
 
     @Override
     public boolean stopMeasuring(final String id) {
         return this.consumers.stopMeasuring(id);
     }
 
     private synchronized void stopTailingIfNecessary() {
         if (!this.tailing.isRunning()) {
             return;
         }
         if (this.consumers.countConsumers() == 0) {
             this.tailing.stop();
             this.currentlyProcessedMessage = null;
         }
     }
 
     /**
      * Invoking this method will cause the running
      * {@link LogWatchStorageSweeper} to be de-scheduled. Any currently present
      * {@link Message}s will only be removed from memory when this watch
      * instance is removed from memory.
      */
     @Override
     public synchronized boolean terminate() {
         if (this.isTerminated()) {
             return false;
         }
         DefaultLogWatch.LOGGER.info("Terminating {}.", this);
         this.isTerminated = true;
         this.consumers.stop();
         this.handingDown.clear();
         this.sweeping.stop();
         this.previousAcceptedMessage = null;
         DefaultLogWatch.LOGGER.info("Terminated {}.", this);
         return true;
     }
 
     @Override
     public String toString() {
         final StringBuilder builder = new StringBuilder();
         builder.append("DefaultLogWatch [getUniqueId()=").append(this.getUniqueId()).append(", ");
         if (this.getWatchedFile() != null) {
             builder.append("getWatchedFile()=").append(this.getWatchedFile()).append(", ");
         }
         builder.append("isTerminated()=").append(this.isTerminated()).append("]");
         return builder.toString();
     }
 
 }