/*
 * 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.rest.resource;

import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.chukwa.Chunk;
import org.apache.hadoop.chukwa.ChunkImpl;
import org.apache.hadoop.chukwa.datacollection.writer.PipelineStageWriter;
import org.apache.hadoop.chukwa.datacollection.writer.SocketTeeWriter;
import org.apache.hadoop.chukwa.datacollection.writer.WriterException;
import org.apache.hadoop.chukwa.rest.bean.ClientTraceBean;
import org.apache.hadoop.chukwa.rest.bean.UserBean;
import org.apache.hadoop.chukwa.rest.bean.WidgetBean;
import org.apache.hadoop.chukwa.util.ExceptionUtil;
import org.apache.hadoop.conf.Configuration;

import com.sun.jersey.api.client.Client;
import com.sun.jersey.api.client.GenericType;

public class TestClientTrace extends SetupTestEnv {  
  public void testClientTrace() {
    // Setup Collector
    Configuration conf = new Configuration();  
    conf.set("chukwaCollector.pipeline",
        SocketTeeWriter.class.getCanonicalName());
    conf.set("chukwaCollector.writerClass", 
        PipelineStageWriter.class.getCanonicalName());    
    try {
      PipelineStageWriter psw = new PipelineStageWriter();
      psw.init(conf);
      // Send a client trace chunk
      ArrayList<Chunk> l = new ArrayList<Chunk>();
      String line = "2009-12-29 22:32:27,047 INFO org.apache.hadoop.hdfs.server.datanode.DataNode.clienttrace: src: /10.10.100.60:43707, dest: /10.10.100.60:50010, bytes: 7003141, op: HDFS_WRITE, cliID: DFSClient_-8389654, offset: 0, srvID: DS-2032680158-98.137.100.60-50010-1259976007324, blockid: blk_-2723720761101769540_705411, duration: 289013780000";      
      l.add(new ChunkImpl("ClientTrace", "name", 1, line.getBytes(), null));
      assertTrue(l.size()==1);
      psw.add(l);
      assertTrue(true);
    } catch (RuntimeException er) {
	fail(ExceptionUtil.getStackTrace(er));
    } catch (WriterException e) {
      fail(ExceptionUtil.getStackTrace(e));
    }
    
    try {
      // Locate the client trace object
      client = Client.create();
      resource = client.resource("http://localhost:"+restPort);
      List<ClientTraceBean> list = resource.path("/hicc/v1/clienttrace").header("Authorization", authorization).get(new GenericType<List<ClientTraceBean>>(){});
      for(ClientTraceBean ctb : list) {
        assertEquals("HDFS_WRITE", ctb.getAction());
      }
    } catch (Exception e) {
      fail(ExceptionUtil.getStackTrace(e));
    }    
  }
}