contrib/src/main/java/org/apache/apex/malhar/contrib/geode/GeodeStore.java

/**
 * 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.apex.malhar.contrib.geode;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.apex.malhar.lib.db.KeyValueStore;

import com.gemstone.gemfire.cache.CacheClosedException;
import com.gemstone.gemfire.cache.CacheWriterException;
import com.gemstone.gemfire.cache.EntryNotFoundException;
import com.gemstone.gemfire.cache.Region;
import com.gemstone.gemfire.cache.TimeoutException;
import com.gemstone.gemfire.cache.client.ClientCache;
import com.gemstone.gemfire.cache.client.ClientCacheFactory;
import com.gemstone.gemfire.cache.client.ClientRegionShortcut;
import com.gemstone.gemfire.cache.query.FunctionDomainException;
import com.gemstone.gemfire.cache.query.NameResolutionException;
import com.gemstone.gemfire.cache.query.QueryInvocationTargetException;
import com.gemstone.gemfire.cache.query.SelectResults;
import com.gemstone.gemfire.cache.query.TypeMismatchException;

/**
 * Provides the implementation of a Geode store.
 * Geode is a distributed in-memory database
 *  that provides reliable asynchronous event notifications and guaranteed message delivery.
 * Geode is a data management platform that provides real-time
 * , consistent access to data-intensive applications.
 *
 *
 * @since 3.4.0
 */
public class GeodeStore implements KeyValueStore, Serializable
{
  /**
   *
   */
  private static final long serialVersionUID = -5076452548893319967L;
  private static final Logger logger = LoggerFactory.getLogger(GeodeStore.class);
  private transient ClientCache clientCache = null;
  private transient Region<Object, Object> region = null;
  private String locatorHost;
  private int locatorPort;
  private String regionName;

  private ClientCache initClient()
  {
    try {
      clientCache = new ClientCacheFactory().addPoolLocator(getLocatorHost(), getLocatorPort()).create();
    } catch (CacheClosedException ex) {
      throw new RuntimeException("Exception while creating cache", ex);

    }

    return clientCache;
  }

  /**
   * @return the regionName
   */
  public String getRegionName()
  {
    return regionName;
  }


  /**
   * @return the clientCache
   */
  public ClientCache getClientCache()
  {
    return clientCache;
  }

  /**
   * @return the locatorPort
   */
  public int getLocatorPort()
  {
    return locatorPort;
  }

  /**
   * @param locatorPort
   *          the locatorPort to set
   */
  public void setLocatorPort(int locatorPort)
  {
    this.locatorPort = locatorPort;
  }

  /**
   * @return the locatorHost
   */
  public String getLocatorHost()
  {
    return locatorHost;
  }

  /**
   * @param locatorHost
   *          the locatorHost to set
   */
  public void setLocatorHost(String locatorHost)
  {
    this.locatorHost = locatorHost;
  }

  /**
   * @return the region
   * @throws IOException
   */
  public Region<Object, Object> getRegion() throws IOException
  {
    // return region;
    if (clientCache == null || clientCache.isClosed()) {
      initClient();
    }

    if (region == null) {
      region = clientCache.getRegion(regionName);
      if (region == null) {
        region = clientCache.<Object, Object>createClientRegionFactory(ClientRegionShortcut.PROXY).create(regionName);
      }
    }

    return region;
  }

  @Override
  public void connect() throws IOException
  {
    try {
      clientCache = new ClientCacheFactory().addPoolLocator(getLocatorHost(), getLocatorPort()).create();
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }

    region = clientCache.getRegion(getRegionName());

    if (region == null) {
      region = clientCache.<Object, Object>createClientRegionFactory(ClientRegionShortcut.PROXY).create(
          getRegionName());
    }

  }

  @Override
  public void disconnect() throws IOException
  {
    clientCache.close();

  }

  @Override
  public boolean isConnected()
  {
    return (clientCache.isClosed());

  }

  /**
   * Gets the value given the key. Note that it does NOT work with hash values
   * or list values
   *
   * @param key
   * @return The value.
   */
  @Override
  public Object get(Object key)
  {

    try {
      return (getRegion().get(key));
    } catch (IOException ex) {
      throw new RuntimeException("Exception while getting the object", ex);

    }

  }

  /**
   * Gets all the values given the keys. Note that it does NOT work with hash
   * values or list values
   *
   * @param keys
   * @return All values for the given keys.
   */
  @SuppressWarnings("unchecked")
  @Override
  public List<Object> getAll(List<Object> keys)
  {

    List<Object> values = new ArrayList<Object>();

    try {
      final Map<Object, Object> entries = getRegion().getAll(keys);
      for (int i = 0; i < keys.size(); i++) {
        values.add(entries.get(keys.get(i)));
      }
    } catch (IOException ex) {
      logger.info("error getting region ", ex);
    }

    return (values);

  }

  /**
   * @param regionName
   *          the regionName to set
   */
  public void setRegionName(String regionName)
  {
    this.regionName = regionName;
  }


  public Map<Object, Object> getAllMap(List<Object> keys)
  {

    try {
      final Map<Object, Object> entries = getRegion().getAll(keys);
      return (entries);
    } catch (IOException ex) {
      logger.info("error getting object ", ex);
      return null;
    }

  }

  @SuppressWarnings("rawtypes")
  public SelectResults query(String predicate)
  {
    try {
      return (getRegion().query(predicate));
    } catch (FunctionDomainException | TypeMismatchException | NameResolutionException | QueryInvocationTargetException
        | IOException e) {
      logger.info("error in querying object ", e);
      return null;
    }

  }

  @Override
  public void put(Object key, Object value)
  {
    try {
      getRegion().put(key, value);
    } catch (IOException e) {
      logger.info("while putting in region", e);
    }
  }

  @Override
  public void putAll(Map<Object, Object> map)
  {
    try {
      getRegion().putAll(map);
    } catch (IOException e) {
      logger.info("while putting all in region", e);
    }
  }

  @Override
  public void remove(Object key)
  {
    try {
      getRegion().destroy(key);
    } catch (TimeoutException | CacheWriterException | EntryNotFoundException | IOException e) {
      logger.info("while deleting", e);
    }
  }

}