code: /**
   * Copyright (c) 2016 YCSB contributors. All rights reserved.
   *
   * Licensed 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. See accompanying
   * LICENSE file.
   */
  
  package com.yahoo.ycsb.db;
  
  import com.ceph.rados.Rados;
  import com.ceph.rados.IoCTX;
  import com.ceph.rados.jna.RadosObjectInfo;
  import com.ceph.rados.ReadOp;
  import com.ceph.rados.ReadOp.ReadResult;
  import com.ceph.rados.exceptions.RadosException;
  
  import com.yahoo.ycsb.ByteIterator;
  import com.yahoo.ycsb.DB;
  import com.yahoo.ycsb.DBException;
  import com.yahoo.ycsb.Status;
  import com.yahoo.ycsb.StringByteIterator;
  
  import java.io.File;
  import java.util.HashMap;
  import java.util.Map.Entry;
  import java.util.Properties;
  import java.util.Set;
  import java.util.Vector;
  
  import org.json.JSONObject;
  
  /**
   * YCSB binding for <a href="http://ceph.org/">RADOS of Ceph</a>.
   *
   * See {@code rados/README.md} for details.
   */
  public class RadosClient extends DB {
  
    private Rados rados;
    private IoCTX ioctx;
  
    public static final String CONFIG_FILE_PROPERTY = "rados.configfile";
    public static final String CONFIG_FILE_DEFAULT = "/etc/ceph/ceph.conf";
    public static final String ID_PROPERTY = "rados.id";
    public static final String ID_DEFAULT = "admin";
    public static final String POOL_PROPERTY = "rados.pool";
    public static final String POOL_DEFAULT = "data";
  
    private boolean isInited = false;
  
    public void init() throws DBException {
      Properties props = getProperties();
  
      String configfile = props.getProperty(CONFIG_FILE_PROPERTY);
      if (configfile == null) {
        configfile = CONFIG_FILE_DEFAULT;
      }
  
      String id = props.getProperty(ID_PROPERTY);
      if (id == null) {
        id = ID_DEFAULT;
      }
  
      String pool = props.getProperty(POOL_PROPERTY);
      if (pool == null) {
        pool = POOL_DEFAULT;
      }
  
      // try {
      // } catch (UnsatisfiedLinkError e) {
      //   throw new DBException("RADOS library is not loaded.");
      // }
  
      rados = new Rados(id);
      try {
        rados.confReadFile(new File(configfile));
        rados.connect();
        ioctx = rados.ioCtxCreate(pool);
      } catch (RadosException e) {
        throw new DBException(e.getMessage() + ": " + e.getReturnValue());
      }
  
      isInited = true;
    }
  
    public void cleanup() throws DBException {
      if (isInited) {
        rados.shutDown();
        rados.ioCtxDestroy(ioctx);
        isInited = false;
      }
    }
  
    @Override
    public Status read(String table, String key, Set<String> fields, HashMap<String, ByteIterator> result) {
      byte[] buffer;
  
      try {
        RadosObjectInfo info = ioctx.stat(key);
        buffer = new byte[(int)info.getSize()];
  
        ReadOp rop = ioctx.readOpCreate();
        ReadResult readResult = rop.queueRead(0, info.getSize());
        // TODO: more size than byte length possible;
        // rop.operate(key, Rados.OPERATION_NOFLAG); // for rados-java 0.3.0
        rop.operate(key, 0);
        // readResult.raiseExceptionOnError("Error ReadOP(%d)", readResult.getRVal()); // for rados-java 0.3.0
        if (readResult.getRVal() < 0) {
          throw new RadosException("Error ReadOP", readResult.getRVal());
        }
        if (info.getSize() != readResult.getBytesRead()) {
          return new Status("ERROR", "Error the object size read");
        }
        readResult.getBuffer().get(buffer);
      } catch (RadosException e) {
        return new Status("ERROR-" + e.getReturnValue(), e.getMessage());
      }
  
      JSONObject json = new JSONObject(new String(buffer, java.nio.charset.StandardCharsets.UTF_8));
      Set<String> fieldsToReturn = (fields == null ? json.keySet() : fields);
  
      for (String name : fieldsToReturn) {
        result.put(name, new StringByteIterator(json.getString(name)));
      }
  
      return result.isEmpty() ? Status.ERROR : Status.OK;
    }
  
    @Override
    public Status insert(String table, String key, HashMap<String, ByteIterator> values) {
      JSONObject json = new JSONObject();
      for (final Entry<String, ByteIterator> e : values.entrySet()) {
        json.put(e.getKey(), e.getValue().toString());
      }
  
      try {
        ioctx.write(key, json.toString());
      } catch (RadosException e) {
        return new Status("ERROR-" + e.getReturnValue(), e.getMessage());
      }
      return Status.OK;
    }
  
    @Override
    public Status delete(String table, String key) {
      try {
        ioctx.remove(key);
      } catch (RadosException e) {
        return new Status("ERROR-" + e.getReturnValue(), e.getMessage());
      }
      return Status.OK;
    }
  
    @Override
    public Status update(String table, String key, HashMap<String, ByteIterator> values) {
      Status rtn = delete(table, key);
      if (rtn.equals(Status.OK)) {
        return insert(table, key, values);
      }
      return rtn;
    }
  
    @Override
    public Status scan(String table, String startkey, int recordcount, Set<String> fields,
        Vector<HashMap<String, ByteIterator>> result) {
      return Status.NOT_IMPLEMENTED;
    }
  }
  