/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.chukwa.datacollection.connector;

/**
 * This class is responsible for setting up connections with configured
 * storage writers base on configuration of chukwa_agent.xml.
 * 
 * On error, tries the list of available storage writers, pauses for a minute, 
 * and then repeats.
 *
 */

import java.util.ArrayList;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.hadoop.chukwa.Chunk;
import org.apache.hadoop.chukwa.datacollection.ChunkQueue;
import org.apache.hadoop.chukwa.datacollection.DataFactory;
import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
import org.apache.hadoop.chukwa.datacollection.connector.Connector;
import org.apache.hadoop.chukwa.datacollection.writer.ChukwaWriter;
import org.apache.hadoop.chukwa.datacollection.writer.ChukwaWriter.CommitStatus;
import org.apache.hadoop.chukwa.datacollection.writer.PipelineStageWriter;
import org.apache.hadoop.chukwa.datacollection.writer.WriterException;
import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;

public class PipelineConnector implements Connector, Runnable {

  static Logger log = Logger.getLogger(PipelineConnector.class);

  Timer statTimer = null;
  volatile int chunkCount = 0;
  
  int MAX_SIZE_PER_POST = 2 * 1024 * 1024;
  int MIN_POST_INTERVAL = 5 * 1000;
  public static final String MIN_POST_INTERVAL_OPT = "pipelineConnector.minPostInterval";
  public static final String MAX_SIZE_PER_POST_OPT = "pipelineConnector.maxPostSize";
  public static final String ASYNC_ACKS_OPT = "pipelineConnector.asyncAcks";

  ChunkQueue chunkQueue;

  private ChukwaAgent agent = null;

  private volatile boolean stopMe = false;
  protected ChukwaWriter writers = null;

  public PipelineConnector() {
    //instance initializer block
    statTimer = new Timer();
    statTimer.schedule(new TimerTask() {
      public void run() {
        int count = chunkCount;
        chunkCount = 0;
        log.info("# Data chunks sent since last report: " + count);
      }
    }, 100, 60 * 1000);
  }
  
  public void start() {
    chunkQueue = DataFactory.getInstance().getEventQueue();
    agent = ChukwaAgent.getAgent();
    Configuration conf = agent.getConfiguration();
    MAX_SIZE_PER_POST = conf.getInt(MAX_SIZE_PER_POST_OPT, MAX_SIZE_PER_POST);
    MIN_POST_INTERVAL = conf.getInt(MIN_POST_INTERVAL_OPT, MIN_POST_INTERVAL);
    try {
      writers = new PipelineStageWriter(conf);
      (new Thread(this, "Pipeline connector thread")).start();
    } catch(Exception e) {
      log.error("Pipeline initialization error: ", e);
    }
  }

  public void shutdown() {
    stopMe = true;
    try {
      writers.close();
    } catch (WriterException e) {
      log.warn("Shutdown error: ",e);
    }
  }

  public void run() {
    log.info("PipelineConnector started at time:" + System.currentTimeMillis());

    try {
      long lastPost = System.currentTimeMillis();
      while (!stopMe) {
        List<Chunk> newQueue = new ArrayList<Chunk>();
        try {
          // get all ready chunks from the chunkQueue to be sent
          chunkQueue.collect(newQueue, MAX_SIZE_PER_POST);
          CommitStatus result = writers.add(newQueue);
          if(result.equals(ChukwaWriter.COMMIT_OK)) {
            chunkCount = newQueue.size();
            for (Chunk c : newQueue) {
              agent.reportCommit(c.getInitiator(), c.getSeqID());
            }          
          }
        } catch (WriterException e) {
          log.warn("PipelineStageWriter Exception: ", e);
        } catch (InterruptedException e) {
          log.warn("thread interrupted during addChunks(ChunkQueue)");
          Thread.currentThread().interrupt();
          break;
        }
        long now = System.currentTimeMillis();
        long delta = MIN_POST_INTERVAL - now + lastPost;
        if(delta > 0) {
          Thread.sleep(delta); // wait for stuff to accumulate
        }
        lastPost = now;
      } // end of try forever loop
      log.info("received stop() command so exiting run() loop to shutdown connector");
    } catch (OutOfMemoryError e) {
      log.warn("Bailing out", e);
      throw new RuntimeException("Shutdown pipeline connector.");
    } catch (InterruptedException e) {
      // do nothing, let thread die.
      log.warn("Bailing out", e);
      throw new RuntimeException("Shutdown pipeline connector.");
    } catch (Throwable e) {
      log.error("connector failed; shutting down agent: ", e);
      throw new RuntimeException("Shutdown pipeline connector.");
    }
  }

  @Override
  public void reloadConfiguration() {
  }
  
}