content: /*
   * 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.sender;
  
  
  import java.io.*;
  import java.nio.charset.Charset;
  import java.util.*;
  
  import org.apache.hadoop.conf.Configuration;
  
  /***
   * An iterator returning a list of Collectors to try. This class is
   * nondeterministic, since it puts collectors back on the list after some
   * period.
   * 
   * No node will be polled more than once per maxRetryRateMs milliseconds.
   * hasNext() will continue return true if you have not called it recently.
   * 
   * 
   */
  public class RetryListOfCollectors implements Iterator<String>, Cloneable {
  
    int maxRetryRateMs;
    List<String> collectors;
    long lastLookAtFirstNode;
    int nextCollector = 0;
    private String portNo;
    public static final String RETRY_RATE_OPT = "chukwaAgent.connector.retryRate";
  
    public RetryListOfCollectors(File collectorFile, Configuration conf)
        throws IOException {
      this(conf);
      try {
        FileInputStream fis = new FileInputStream(collectorFile);
        BufferedReader br = new BufferedReader(new InputStreamReader(fis, Charset.forName("UTF-8")));
        String line, parsedline;
        while ((line = br.readLine()) != null) {
          parsedline = canonicalizeLine(line);
          collectors.add(parsedline);
        }
        
        br.close();
      } catch (FileNotFoundException e) {
        System.err.println("Error in RetryListOfCollectors() opening file"
              + collectorFile.getCanonicalPath() + ", double check that you have"
              + "set the CHUKWA_CONF_DIR environment variable. Also, ensure file"
              + " exists and is in classpath");
        throw e;
      } catch (IOException e) {
        System.err
            .println("I/O error in RetryListOfcollectors instantiation in readLine() from specified collectors file");
        throw e;
      }
      shuffleList();
    }
  
    private String canonicalizeLine(String line) {
      String parsedline;
      if (!line.contains("://")) {
        // no protocol, assume http
        if (line.matches(".*:\\d+.*")) {
          parsedline = "http://" + line+"/";
        } else {
          parsedline = "http://" + line + ":" + portNo;
        }
      } else {
        if (line.matches(".*:\\d+.*")) {
          parsedline = line;
        } else {
          parsedline = line + ":" + portNo;
        }
      }
      if(!parsedline.matches(".*\\w/.*")) //no resource name
        parsedline = parsedline+"/";
      return parsedline;
    }
  
    /**
     * This is only used for debugging. Possibly it should sanitize urls the same way the other
     * constructor does.
     * @param collectors is list of collector hostname
     * @param conf is Chukwa configuration
     */
    public RetryListOfCollectors(final List<String> collectors, Configuration conf) {
      this(conf);
      this.collectors.addAll(collectors);
      //we don't shuffle the list here -- this constructor is only used for test purposes
    }
    
    public RetryListOfCollectors(Configuration conf) {
      collectors = new ArrayList<String>();
      portNo = conf.get("chukwaCollector.http.port", "8080");
      maxRetryRateMs = conf.getInt(RETRY_RATE_OPT, 15 * 1000);
      lastLookAtFirstNode = 0;
    }
  
    // for now, use a simple O(n^2) algorithm.
    // safe, because we only do this once, and on smallish lists
    public void shuffleList() {
      ArrayList<String> newList = new ArrayList<String>();
      Random r = new java.util.Random();
      while (!collectors.isEmpty()) {
        int toRemove = r.nextInt(collectors.size());
        String next = collectors.remove(toRemove);
        newList.add(next);
      }
      collectors = newList;
    }
  
    public boolean hasNext() {
      return collectors.size() > 0
          && ((nextCollector != 0) || (System.currentTimeMillis()
              - lastLookAtFirstNode > maxRetryRateMs));
    }
  
    public String next() {
      if (hasNext()) {
        int currCollector = nextCollector;
        nextCollector = (nextCollector + 1) % collectors.size();
        if (currCollector == 0)
          lastLookAtFirstNode = System.currentTimeMillis();
        return collectors.get(currCollector);
      } else
        return null;
    }
  
    public void add(String collector) {
      collectors.add(collector);
    }
  
    public void remove() {
      throw new UnsupportedOperationException();
      // FIXME: maybe just remove a collector from our list and then
      // FIXME: make sure next doesn't break (i.e. reset nextCollector if
      // necessary)
    }
  
    /**
     * 
     * @return total number of collectors in list
     */
    int total() {
      return collectors.size();
    }
    
    public RetryListOfCollectors clone() {
      try {
        RetryListOfCollectors clone = (RetryListOfCollectors) super.clone();
        return clone;
      } catch(CloneNotSupportedException e) {
        return null;
      }
    }
  
  }
  