/**
 * Copyright (c) 2015-2016 YCSB contributors. All rights reserved.
 * <p>
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.yahoo.ycsb.*;
import org.infinispan.client.hotrod.RemoteCache;
import org.infinispan.client.hotrod.RemoteCacheManager;
import org.infinispan.util.logging.Log;
import org.infinispan.util.logging.LogFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.Vector;

/**
 * This is a client implementation for Infinispan 5.x in client-server mode.
 */
public class InfinispanRemoteClient extends DB {

  private static final Log LOGGER = LogFactory.getLog(InfinispanRemoteClient.class);

  private RemoteCacheManager remoteIspnManager;
  private String cacheName = null;

  @Override
  public void init() throws DBException {
    remoteIspnManager = RemoteCacheManagerHolder.getInstance(getProperties());
    cacheName = getProperties().getProperty("cache");
  }

  @Override
  public void cleanup() {
    remoteIspnManager.stop();
    remoteIspnManager = null;
  }

  @Override
  public Status insert(String table, String recordKey, HashMap<String, ByteIterator> values) {
    String compositKey = createKey(table, recordKey);
    Map<String, String> stringValues = new HashMap<>();
    StringByteIterator.putAllAsStrings(stringValues, values);
    try {
      cache().put(compositKey, stringValues);
      return Status.OK;
    } catch (Exception e) {
      LOGGER.error(e);
      return Status.ERROR;
    }
  }

  @Override
  public Status read(String table, String recordKey, Set<String> fields, HashMap<String, ByteIterator> result) {
    String compositKey = createKey(table, recordKey);
    try {
      Map<String, String> values = cache().get(compositKey);

      if (values == null || values.isEmpty()) {
        return Status.NOT_FOUND;
      }

      if (fields == null) { //get all field/value pairs
        StringByteIterator.putAllAsByteIterators(result, values);
      } else {
        for (String field : fields) {
          String value = values.get(field);
          if (value != null) {
            result.put(field, new StringByteIterator(value));
          }
        }
      }

      return Status.OK;
    } catch (Exception e) {
      LOGGER.error(e);
      return Status.ERROR;
    }
  }

  @Override
  public Status scan(String table, String startkey, int recordcount, Set<String> fields,
                     Vector<HashMap<String, ByteIterator>> result) {
    LOGGER.warn("Infinispan does not support scan semantics");
    return Status.NOT_IMPLEMENTED;
  }

  @Override
  public Status update(String table, String recordKey, HashMap<String, ByteIterator> values) {
    String compositKey = createKey(table, recordKey);
    try {
      Map<String, String> stringValues = new HashMap<>();
      StringByteIterator.putAllAsStrings(stringValues, values);
      cache().put(compositKey, stringValues);
      return Status.OK;
    } catch (Exception e) {
      LOGGER.error(e);
      return Status.ERROR;
    }
  }

  @Override
  public Status delete(String table, String recordKey) {
    String compositKey = createKey(table, recordKey);
    try {
      cache().remove(compositKey);
      return Status.OK;
    } catch (Exception e) {
      LOGGER.error(e);
      return Status.ERROR;
    }
  }

  private RemoteCache<String, Map<String, String>> cache() {
    if (this.cacheName != null) {
      return remoteIspnManager.getCache(cacheName);
    } else {
      return remoteIspnManager.getCache();
    }
  }

  private String createKey(String table, String recordKey) {
    return table + "-" + recordKey;
  }
}
