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.util;
  
  
  import java.util.Random;
  import java.util.regex.*;
  import org.apache.hadoop.chukwa.*;
  import org.apache.hadoop.chukwa.datacollection.*;
  import org.apache.hadoop.chukwa.datacollection.adaptor.*;
  import org.apache.hadoop.conf.Configuration;
  
  /**
   * Emits chunks at a roughly constant data rate. Chunks are in a very particular
   * format: the output data is verifiable, but sufficiently non-deterministic
   * that two different instances of this adaptor are very likely to have
   * distinct outputs.
   * 
   * 
   * Each chunk is full of random bytes; the randomness comes from 
   * an instance of java.util.Random seeded with the offset xored
   * with the time-of-generation. The time of generation is stored, big-endian,
   * in the first eight bytes of each chunk.
   */
  public class ConstRateAdaptor extends AbstractAdaptor implements Runnable {
  
    private int SLEEP_VARIANCE = 200;
    private int MIN_SLEEP = 300;
  
    private long offset;
    private int bytesPerSec;
  
    Random timeCoin;
    long seed;
    
    private volatile boolean stopping = false;
  
    public String getCurrentStatus() {
      return type.trim() + " " + bytesPerSec + " " + seed;
    }
  
    public void start(long offset) throws AdaptorException {
  
      this.offset = offset;
      Configuration conf = control.getConfiguration();
      MIN_SLEEP = conf.getInt("constAdaptor.minSleep", MIN_SLEEP);
      SLEEP_VARIANCE = conf.getInt("constAdaptor.sleepVariance", SLEEP_VARIANCE);
      
      timeCoin = new Random(seed);
      long o =0;
      while(o < offset)
        o += (int) ((timeCoin.nextInt(SLEEP_VARIANCE) + MIN_SLEEP) *
            (long) bytesPerSec / 1000L) + 8;
      new Thread(this).start(); // this is a Thread.start
    }
  
    public String parseArgs(String bytesPerSecParam) {
      try {
        Matcher m = Pattern.compile("([0-9]+)(?:\\s+([0-9]+))?\\s*").matcher(bytesPerSecParam);
        if(!m.matches())
          return null;
        bytesPerSec = Integer.parseInt(m.group(1));
        String rate = m.group(2);
        if(rate != null)
          seed = Long.parseLong(m.group(2));
        else
          seed = System.currentTimeMillis();
      } catch (NumberFormatException e) {
        //("bad argument to const rate adaptor: ["  + bytesPerSecParam + "]");
        return null;
      }
      return bytesPerSecParam;
    }
  
    public void run() {
      try {
        while (!stopping) {
          int MSToSleep = timeCoin.nextInt(SLEEP_VARIANCE) + MIN_SLEEP; 
          int arraySize = (int) (MSToSleep * (long) bytesPerSec / 1000L) + 8;
          ChunkImpl evt = nextChunk(arraySize );
  
          dest.add(evt);
  
          Thread.sleep(MSToSleep);
        } // end while
      } catch (InterruptedException ie) {
      } // abort silently
    }
  
    public ChunkImpl nextChunk(int arraySize) {
      byte[] data = new byte[arraySize];
      Random dataPattern = new Random(offset ^ seed);
      long s = this.seed;
      offset += data.length;
      dataPattern.nextBytes(data);
      for(int i=0; i < 8; ++i)  {
        data[7-i] = (byte) (s & 0xFF);
        s >>= 8;
      }
      ChunkImpl evt = new ChunkImpl(type, "random ("+ this.seed+")", offset, data,
          this);
      return evt;
    }
  
    public String toString() {
      return "const rate " + type;
    }
  
    @Override
    public long shutdown(AdaptorShutdownPolicy shutdownPolicy) {
      stopping = true;
      return offset;
    }
    
    public static boolean checkChunk(Chunk chunk) {
      byte[] data = chunk.getData();
      byte[] correctData = new byte[data.length];
      
      long seed = 0;
      for(int i=0; i < 8; ++i) 
        seed = (seed << 8) | (0xFF & data[i] );
  
      seed ^= (chunk.getSeqID() - data.length);
      Random dataPattern = new Random(seed);
      dataPattern.nextBytes(correctData);
      for(int i=8; i < data.length ; ++i) 
        if(data [i] != correctData[i])
          return false;
       
      return true;
    }
    
    void test_init(String type) {
      this.type = type;
      seed = System.currentTimeMillis();
    }
  }
  