RemoteCache.h

Go to the documentation of this file.

#ifndef ISPN_HOTROD_REMOTECACHE_H
#define ISPN_HOTROD_REMOTECACHE_H



#include "infinispan/hotrod/RemoteCacheBase.h"
#include "infinispan/hotrod/Marshaller.h"
#include "infinispan/hotrod/Flag.h"
#include "infinispan/hotrod/MetadataValue.h"
#include "infinispan/hotrod/TimeUnit.h"
#include "infinispan/hotrod/VersionedValue.h"
#include "infinispan/hotrod/Version.h"
#include "infinispan/hotrod/exceptions.h"
#include "query.pb.h"

#include <cmath>
#include <set>
#include <map>
#include <sstream>
#include <stdexcept>
#include <vector>
#include <functional>
#include <future>

namespace infinispan {
namespace hotrod {

template <class K, class V> class RemoteCache : private RemoteCacheBase
{
  private:
#ifndef SWIG      // Let SWIG ignore this method
        template<typename Function>
                inline std::future<typename std::result_of<Function()>::type> goAsync(Function&& f,
                        std::function<typename std::result_of<Function()>::type (typename std::result_of<Function()>::type)> success, std::function<typename std::result_of<Function()>::type (std::exception&)> fail)
                                {
                                auto fq= [=]{ try{ return success==0 ? f() : success(f());}
                        catch (std::exception& ex)
                                      {if (fail!=0){return fail(ex);}
                                      else {throw ex;}}
                          };
                                return std::async(fq);
                                }
#endif

  public:
    std::string getName() {
        return std::string(base_getName());
    }

    std::string getVersion() {
        return Version::getVersionCString();
    }

    std::string getProtocolVersion() {
        return Version::getProtocolVersionCString();
    }

    V* get(const K& key) {
        return (V *) base_get(&key);
    }

     std::future<V*> getAsync(const K& key, std::function<V* (V*)> success=nullptr, std::function<V* (std::exception&)> fail=nullptr) {
         auto pKey=&key;
         auto f= [=] { return this->get(*pKey); };
         return goAsync(f,success,fail);
     }

    V* put(const K& key, const V& val, uint64_t lifespan = 0, uint64_t maxIdle = 0) {
        return put(key, val, lifespan, SECONDS, maxIdle, SECONDS);
    }
    V* put(const K& key, const V& val, uint64_t lifespan, TimeUnit lifespanUnit) {
        return put(key, val, lifespan, lifespanUnit, 0, SECONDS);
    }

    V* put(const K& key, const V& val, uint64_t lifespan, TimeUnit lifespanUnit, uint64_t maxIdle, TimeUnit maxIdleUnit) {
        return (V *) base_put(&key, &val, toSeconds(lifespan, lifespanUnit), toSeconds(maxIdle, maxIdleUnit));
    }

     std::future<V*> putAsync(const K& key, const V& val, uint64_t lifespan = 0, uint64_t maxIdle = 0, std::function<V* (V*)> success=nullptr, std::function<V* (std::exception&)> fail=nullptr) {
                auto pKey=&key, pVal=&val;
                auto f=[=] { return this->put(*pKey,*pVal,lifespan,maxIdle); };
                return goAsync(f,success,fail);
     }

    V* putAsync(const K& key, const V& val, uint64_t lifespan, TimeUnit lifespanUnit, std::function<V* (V*)> success=nullptr, std::function<V* (std::exception&)> fail=nullptr) {
        auto pKey=&key, pVal=&val;
        auto f= [=] { return this->put(*pKey,*pVal,lifespan,lifespanUnit); };
            return goAsync(f,success,fail);
    }

    V* putAsync(const K& key, const V& val, uint64_t lifespan, TimeUnit lifespanUnit, uint64_t maxIdle, TimeUnit maxIdleUnit, std::function<V* (V*)> success=nullptr, std::function<V* (std::exception&)> fail=nullptr) {
                auto pKey=&key, pVal=&val;
        auto f= [=] { this->put(*pKey,*pVal,lifespan, lifespanUnit,maxIdle,maxIdleUnit); };
        return goAsync(f,success,fail);
    }

    V* putIfAbsent(const K& key, const V& val, uint64_t lifespan = 0, uint64_t maxIdle = 0) {
        return putIfAbsent(key, val, lifespan, SECONDS, maxIdle, SECONDS);
    }
    V* putIfAbsent(const K& key, const V& val, uint64_t lifespan, TimeUnit lifespanUnit) {
        return putIfAbsent(key, val, lifespan, lifespanUnit, 0, SECONDS);
    }
    V* putIfAbsent(const K& key, const V& val, uint64_t lifespan, TimeUnit lifespanUnit, uint64_t maxIdle, TimeUnit maxIdleUnit) {
        return (V *)base_putIfAbsent(&key, &val, toSeconds(lifespan, lifespanUnit), toSeconds(maxIdle, maxIdleUnit));
    }
    void putAll(const std::map<K, V>& map, uint64_t lifespan = 0, uint64_t maxIdle = 0) {
        putAll(map, lifespan, SECONDS, maxIdle, SECONDS);
    }

    void putAll(const std::map<K, V>& map, uint64_t lifespan, TimeUnit lifespanUnit) {
        putAll(map, lifespan, lifespanUnit, 0, SECONDS);
    }

    void putAll(const std::map<K, V>& map, uint64_t lifespan, TimeUnit lifespanUnit, uint64_t maxIdle, TimeUnit maxIdleUnit) {
        uint64_t lifespanMillis = toSeconds(lifespan, lifespanUnit);
        uint64_t maxIdleMillis = toSeconds(maxIdle, maxIdleUnit);

        for (typename std::map<K, V>::const_iterator it = map.begin(); it != map.end(); ++it) {
            put(it->first, it->second, lifespanMillis, maxIdleMillis);
        }
    }

    std::future<void> putAllAsync(const std::map<K, V>& map, uint64_t lifespan = 0, uint64_t maxIdle = 0, std::function<V* (V*)> success=nullptr, std::function<void (std::exception&)> fail=nullptr) {
        auto pMap = &pMap;
        auto f = [=] { this->putAll(*pMap,lifespan, maxIdle); };
        return goAsync(f,success,fail);
    }

    std::future<void> putAllAsync(const std::map<K, V>& map, uint64_t lifespan, TimeUnit lifespanUnit, std::function<V* (V*)> success=nullptr, std::function<void (std::exception&)> fail=nullptr) {
        auto pMap = &pMap;
        auto f = [=] { this->putAll(*pMap,lifespan, lifespanUnit); };
        return goAsync(f,success,fail);
    }

    std::future<void> putAllAsync(const std::map<K, V>& map, uint64_t lifespan, TimeUnit lifespanUnit, uint64_t maxIdle, TimeUnit maxIdleUnit, std::function<V* (V*)> success=nullptr, std::function<void (std::exception&)> fail=nullptr) {
        auto pMap = &pMap;
        auto f = [=] { this->putAll(*pMap,lifespan, lifespanUnit, maxIdle, maxIdleUnit); };
        return goAsync(f,success,fail);
    }

    V* replace(const K& key, const V& val, uint64_t lifespan = 0, uint64_t maxIdle = 0) {
        return replace(key, val, lifespan, SECONDS, maxIdle, SECONDS);
    }
    V* replace(const K& key, const V& val, uint64_t lifespan, TimeUnit lifespanUnit) {
        return replace(key, val, lifespan, lifespanUnit, 0, SECONDS);
    }
    V* replace(const K& key, const V& val, uint64_t lifespan, TimeUnit lifespanUnit, uint64_t maxIdle, TimeUnit maxIdleUnit) {
        return (V *) base_replace(&key, &val, toSeconds(lifespan, lifespanUnit), toSeconds(maxIdle, maxIdleUnit));
    }
    V* replace(const K& key, const V& oldVal, const V& val, uint64_t lifespan = 0, uint64_t maxIdle = 0) {
        return replace(key, oldVal, val, lifespan, SECONDS, maxIdle, SECONDS);
    }
    V* replace(const K& key, const V& oldVal, const V& val, uint64_t lifespan, TimeUnit lifespanUnit) {
        return replace(key, oldVal, val, lifespan, lifespanUnit, 0, SECONDS);
    }
    V* replace(const K& key, const V& oldVal, const V& val, uint64_t lifespan, TimeUnit lifespanUnit, uint64_t maxIdle, TimeUnit maxIdleUnit) {
        throw UnsupportedOperationException();
    }

    V* remove(const K& key) {
        return (V *) base_remove(&key);
    }

    bool containsKey(const K& key) {
        return base_containsKey(&key);
    }

    bool containsValue(const V& val) {
        throw UnsupportedOperationException();
    }
    bool replaceWithVersion(
        const K& key, const V& val,
        uint64_t version, uint64_t lifespan = 0, uint64_t maxIdle = 0) {
        return base_replaceWithVersion(&key, &val, version, lifespan, maxIdle);
    }

    std::future<void> replaceWithVersionAsync(const K& key, const V& val,
            uint64_t version, uint64_t lifespan, uint64_t maxIdle, std::function<V* (V*)> success=nullptr, std::function<void (std::exception&)> fail=nullptr) {
        auto pMap = &pMap;
        auto f = [=] { this->replaceWithVersion(&key, &val, version, lifespan, maxIdle); };
        return goAsync(f,success,fail);
    }

    bool removeWithVersion(const K& key, uint64_t version) {
        return base_removeWithVersion(&key, version);
    }
    std::pair<std::shared_ptr<V>, VersionedValue> getWithVersion(const K& key) {
        VersionedValue version;
        void *value = base_getWithVersion(&key, &version);
        return std::make_pair(std::shared_ptr<V>((V *) value), version);
    }
    std::pair<std::shared_ptr<V>, MetadataValue> getWithMetadata(const K& key) {
        MetadataValue metadata;
        void *value = base_getWithMetadata(&key, &metadata);
        return std::make_pair(std::shared_ptr<V>((V *) value), metadata);
    }
    std::map<std::shared_ptr<K>, std::shared_ptr<V> > getBulk() {
        return getBulk(0);
    }
    std::map<std::shared_ptr<K>, std::shared_ptr<V> > getBulk(int nrOfEntries) {
        portable::map<void*, void*> mbuf;
        base_getBulk(nrOfEntries, mbuf);

        std::map<std::shared_ptr<K>, std::shared_ptr<V> > result;
        const portable::pair<void*, void*> *data = mbuf.data();
        for (size_t i = 0; i < mbuf.size(); i++) {
            result.insert(std::make_pair(
                std::shared_ptr<K>((K*)data[i].key), std::shared_ptr<V>((V*)data[i].value)));
        }
        return result;
    }
    std::vector<unsigned char> execute(const std::string& name, const std::map<std::string,std::string>& args)
    {
        return base_execute(name,args);
    }

    QueryResponse query(const QueryRequest &qr)
    {
        return base_query(qr);
    }

    std::vector<unsigned char> query(std::vector<unsigned char> qr, size_t size)
    {
        return base_query_char(qr,size);
    }

    std::set<std::pair<K, V> > entrySet() {
        throw UnsupportedOperationException();
    }
    std::set<std::shared_ptr<K> > keySet() {
        portable::vector<void*> p;
        base_keySet(0, p);

        std::set<std::shared_ptr<K> > result;
        for (size_t i = 0; i < p.size(); ++i) {
            result.insert(std::shared_ptr<K>((K*)p.data()[i]));
        }
        return result;
    }
    uint64_t size() {
        return base_size();
    }
    bool isEmpty() {
        return 0 == size();
    }
    std::vector<V> values() {
        throw UnsupportedOperationException();
    }
    std::map<std::string, std::string> stats() {
        portable::map<portable::string,portable::string> statistics;
        base_stats(statistics);
        return statistics.std_map<std::string, portable::string::convert, std::string, portable::string::convert>
            (portable::string::convert(), portable::string::convert());
    }
    void clear() {
        base_clear();
    }
    void ping() {
        base_ping();
    }

    CacheTopologyInfo getCacheTopologyInfo(){
        return base_getCacheTopologyInfo();
    }
    RemoteCache<K,V>& withFlags(Flag flags) {
        base_withFlags(flags);
        return *this;
    }

    RemoteCache(const RemoteCache &other) :
      RemoteCacheBase(other), keyMarshaller(other.keyMarshaller), valueMarshaller(other.valueMarshaller)
    {
        setMarshallers(this, &keyMarshall, &valueMarshall, &keyUnmarshall, &valueUnmarshall);
    }

    RemoteCache<K,V>& operator=(const RemoteCache& other) {
        RemoteCacheBase::operator=(other);
        keyMarshaller = other.keyMarshaller;
        valueMarshaller = other.valueMarshaller;
        setMarshallers(this, &keyMarshall, &valueMarshall, &keyUnmarshall, &valueUnmarshall);
        return *this;
    }

  private:
    RemoteCache() : RemoteCacheBase() {
        setMarshallers(this, &keyMarshall, &valueMarshall, &keyUnmarshall, &valueUnmarshall);
    }

    uint64_t toSeconds(uint64_t time, TimeUnit unit) {
        uint64_t result;
        switch (unit) {
        case NANOSECONDS:
            result = (uint64_t) ceil(time / 1000000000.0);
            break;
        case MICROSECONDS:
            result = (uint64_t) ceil(time / 1000000.0);
            break;
        case MILLISECONDS:
            result = (uint64_t) ceil(time / 1000.0);
            break;
        case SECONDS:
            result = time;
            break;
        case MINUTES:
            result = time * 60;
            break;
        case HOURS:
            result = time * 3600;
            break;
        case DAYS:
            result = time * 86400;
            break;
        default:
            std::stringstream ss;
            ss << "Unhandled TimeUnit specified: " << unit << ".";
            throw std::invalid_argument(ss.str());
        }
        return result;
    }

    // type-hiding and resurrecting support
    static void keyMarshall(void *thisp, const void* key, std::vector<char> &buf) {
        ((RemoteCache<K, V> *) thisp)->keyMarshaller->marshall(*(const K *) key, buf);
    }
    static void valueMarshall(void* thisp, const void* val, std::vector<char> &buf) {
        ((RemoteCache<K, V> *)thisp)->valueMarshaller->marshall(*(const V *) val, buf);
    }
    static void* keyUnmarshall(void *thisp, const std::vector<char> &buf) {
        return ((RemoteCache<K, V> *) thisp)->keyMarshaller->unmarshall(buf);
    }
    static void* valueUnmarshall(void* thisp, const std::vector<char> &buf) {
        return ((RemoteCache<K, V> *)thisp)->valueMarshaller->unmarshall(buf);
    }

    portable::counting_ptr<Marshaller<K> > keyMarshaller;
    portable::counting_ptr<Marshaller<V> > valueMarshaller;

  friend class RemoteCacheManager;
};

}} // namespace

#endif  /* ISPN_HOTROD_REMOTECACHE_H */