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.datastore;
  
  import java.io.File;
  import java.io.IOException;
  import java.nio.charset.Charset;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.json.simple.JSONArray;
  import org.json.simple.JSONObject;
  import org.json.simple.JSONValue;
  import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
  import org.apache.hadoop.chukwa.hicc.HiccWebServer;
  import org.apache.hadoop.chukwa.rest.bean.UserBean;
  import org.apache.hadoop.chukwa.util.ExceptionUtil;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.FSDataInputStream;
  import org.apache.hadoop.fs.FSDataOutputStream;
  import org.apache.hadoop.fs.FileStatus;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  
  public class UserStore {
    private String uid = null;
    private UserBean profile = null;
    private static Log log = LogFactory.getLog(UserStore.class);
    private static Configuration config = new Configuration();
    private static ChukwaConfiguration chukwaConf = new ChukwaConfiguration();
    private static String hiccPath = null;
      
    static {
      config = HiccWebServer.getConfig();
      hiccPath = config.get("fs.defaultFS")+File.separator+chukwaConf.get("chukwa.data.dir")+File.separator+"hicc"+File.separator+"users";
  }
    
    public UserStore() throws IllegalAccessException {
    }
  
    public UserStore(String uid) throws IllegalAccessException {
      this.uid = uid;
      init(uid);
    }
  
    public void init(String uid) throws IllegalAccessException {
      StringBuilder profilePath = new StringBuilder();
      profilePath.append(hiccPath);
      profilePath.append(File.separator);
      profilePath.append(uid);
      profilePath.append(".profile");
      Path profileFile = new Path(profilePath.toString());
      FileSystem fs;
      try {
        fs = FileSystem.get(config);
        if(fs.exists(profileFile)) {
          FileStatus[] fstatus = fs.listStatus(profileFile);
          long size = fstatus[0].getLen();
          FSDataInputStream viewStream = fs.open(profileFile);
          byte[] buffer = new byte[(int)size];
          viewStream.readFully(buffer);
          viewStream.close();
          try {
            JSONObject json = (JSONObject) JSONValue.parse(new String(buffer, Charset.forName("UTF-8")));
            profile = new UserBean(json);
          } catch (Exception e) {
            log.error(ExceptionUtil.getStackTrace(e));
            throw new IllegalAccessException("Unable to access user profile database.");
          }
        } else {
          profile = new UserBean();
          profile.setId(uid);
          JSONArray ja = new JSONArray();
          profile.setViews(ja);
          JSONObject json = new JSONObject();
          profile.setProperties(json.toString());
        }
      } catch (IOException ex) {
        log.error(ExceptionUtil.getStackTrace(ex));
      }
    }
    
    public UserBean get() throws IllegalAccessException {
      if(profile==null) {
        init(uid);
      }
      return profile;
    }
    
    public void set(UserBean profile) throws IllegalAccessException {
      StringBuilder profilePath = new StringBuilder();
      profilePath.append(hiccPath);
      profilePath.append(File.separator);
      profilePath.append(profile.getId());
      profilePath.append(".profile");
      Path profileFile = new Path(profilePath.toString());
      FileSystem fs;
      try {
        fs = FileSystem.get(config);
        FSDataOutputStream out = fs.create(profileFile,true);
        out.write(profile.deserialize().toString().getBytes(Charset.forName("UTF-8")));
        out.close();
      } catch (IOException ex) {
        log.error(ExceptionUtil.getStackTrace(ex));
        throw new IllegalAccessException("Unable to access user profile database.");
      }
      this.profile = profile;
    }
    
    public static JSONArray list() throws IllegalAccessException {
      StringBuilder profilePath = new StringBuilder();
      profilePath.append(hiccPath);
      profilePath.append(File.separator);
      profilePath.append("*.profile");
      Path viewFile = new Path(profilePath.toString());
      FileSystem fs;
      JSONArray list = new JSONArray();
      try {
        fs = FileSystem.get(config);
        FileStatus[] fstatus = fs.listStatus(viewFile);
        if(fstatus!=null) {
          for(int i=0;i<fstatus.length;i++) {
            long size = fstatus[i].getLen();
            FSDataInputStream profileStream = fs.open(fstatus[i].getPath());
            byte[] buffer = new byte[(int)size];
            profileStream.readFully(buffer);
            profileStream.close();
            try {
              UserBean user = new UserBean((JSONObject) JSONValue.parse(new String(buffer, Charset.forName("UTF-8"))));
              list.add(user.getId());
            } catch (Exception e) {
              log.error(ExceptionUtil.getStackTrace(e));
            }
          }
        }
      } catch (IOException ex) {
        log.error(ExceptionUtil.getStackTrace(ex));
        throw new IllegalAccessException("Unable to access user profile database."); 
      }
      return list;    
    }
  }
             
  