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 "infinispan/hotrod/ClientListener.h"
#include "infinispan/hotrod/Query.h"
#include "infinispan/hotrod/QueryUtils.h"
#include "infinispan/hotrod/ContinuousQueryListener.h"
#include "infinispan/hotrod/BasicTypesProtoStreamMarshaller.h"
#include "infinispan/hotrod/JBossMarshaller.h"
#include "infinispan/hotrod/RemoteExecution.h"
#include <cmath>
#include <set>
#include <map>
#include <sstream>
#include <stdexcept>
#include <vector>
#include <functional>
#include <future>
#include <iterator>

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);
    }

    template<typename Function>
    inline std::future<void> goAsync(Function&& f, std::function<void()> success,
            std::function<void(std::exception&)> fail)
    {
        auto fq = [=]
        {   try
            {   if (success==0) f();
                else success();}
            catch (std::exception& ex)
            {   if (fail!=0)
                {   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);
    }

private:
    V* get_async(const K& key)
    {
        return (V *) base_get(&key);
    }

public:

    std::map<std::shared_ptr<K>, std::shared_ptr<V> > getAll(const std::set<K>& keySet)
    {
        std::set<std::vector<char> > keySetMarshalled;
        for (auto item = keySet.begin(); item != keySet.end(); item++)
        {
            std::vector<char> v;
            this->keyMarshaller->marshall(*item, v);
            keySetMarshalled.insert(v);
        }
        auto marshalledResult = base_getAll(keySetMarshalled);
        std::map<std::shared_ptr<K>, std::shared_ptr<V> > result;
        for (auto item : marshalledResult)
        {
            std::shared_ptr<K> rk(this->keyMarshaller->unmarshall(item.first));
            std::shared_ptr<V> rv(this->keyMarshaller->unmarshall(item.second));
            result[rk] = rv;
        }
        return result;
    }

    std::future<V*> getAsync(const K& key, std::function<V* (V*)> success = nullptr,
            std::function<V* (std::exception&)> fail = nullptr)
    {
        const K* pKey = &key;
        auto currTx = transactionManager.getCurrentTransaction();
        std::function<V* (void)> f = [=]
        {   return (V*)this->base_get(pKey, currTx);};
        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)
    {
        const K* pKey = &key;
        const V* pVal = &val;
        auto currTx = transactionManager.getCurrentTransaction();
        std::function<V* (void)> f = [=]
        {   return (V*)this->base_put(pKey, pVal, lifespan, maxIdle, currTx);};
        return goAsync(f, success, fail);
    }

    std::future<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)
    {
        const K* pKey = &key;
        const V* pVal = &val;
        auto currTx = transactionManager.getCurrentTransaction();
        std::function<V* (void)> f = [=]
        {   return (V*)this->base_put(pKey, pVal, lifespan, lifespanUnit, currTx);};
        return goAsync(f, success, fail);
    }

    std::future<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)
    {
        const K* pKey = &key;
        const V* pVal = &val;
        auto currTx = transactionManager.getCurrentTransaction();
        std::function<V* (void)> f = [=]
        { return (V*)this->base_put(pKey, pVal, toSeconds(lifespan, lifespanUnit), toSeconds(maxIdle, maxIdleUnit), currTx); };
        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));
    }
    std::future<V*> putIfAbsentAsync(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)
    {
        const K* pKey = &key;
        const V* pVal = &val;
        auto currTx = transactionManager.getCurrentTransaction();
        std::function<V* (void)> f = [=]
        {   return (V*)this->base_putIfAbsent(pKey,pVal, lifespan, maxIdle, currTx);};
        return goAsync(f, success, fail);
    }
    std::future<V*> putIfAbsentAsync(const K& key, const V& val, uint64_t lifespan, TimeUnit lifespanUnit,
            std::function<V* (V*)> success = nullptr, std::function<V* (std::exception&)> fail = nullptr)
    {
        const K* pKey = &key;
        const V* pVal = &val;
        auto currTx = transactionManager.getCurrentTransaction();
        std::function<V* (void)> f = [=]
        { return (V*)this->base_putIfAbsent(pKey, pVal, toSeconds(lifespan, lifespanUnit), toSeconds(0, SECONDS), currTx); };
        return goAsync(f, success, fail);
    }
    std::future<V*> putIfAbsentAsync(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)
    {
        const K* pKey = &key;
        const V* pVal = &val;
        auto currTx = transactionManager.getCurrentTransaction();
        std::function<V* (void)> f = [=]
        { return (V*)this->base_putIfAbsent(pKey,pVal, toSeconds(lifespan, lifespanUnit), toSeconds(maxIdle, maxIdleUnit), currTx);};
        return goAsync(f, success, fail);
    }

    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);
        std::map<const void*, const void*> tmpMap;
        for (auto& it : map)
        {
            tmpMap[(const void *)&it.first] = (const void *)&it.second;
        }
        base_putAll(tmpMap, lifespanMillis, maxIdleMillis);
    }

    std::future<void> putAllAsync(const std::map<K, V>& map, uint64_t lifespan = 0, uint64_t maxIdle = 0,
            std::function<void()> success = nullptr, std::function<void(std::exception&)> fail = nullptr)
    {
        auto pMap = map;
        auto f = [=]
        {   return 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<void(void)> success = nullptr, std::function<void(std::exception&)> fail = nullptr)
    {
        auto pMap = map;
        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<void(void)> success = nullptr,
            std::function<void(std::exception&)> fail = nullptr)
    {
        auto pMap = &map;
        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();
    }

    std::future<V*> replaceAsync(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->replace(*pKey, *pVal, lifespan, SECONDS, maxIdle, SECONDS);};
        return goAsync(f, success, fail);
    }

    std::future<V*> replaceAsync(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->replace(*pKey, *pVal, lifespan, lifespanUnit, 0, SECONDS);};
        return goAsync(f, success, fail);
    }

    std::future<V*> replaceAsync(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 = [=]
        {   return this->replace(*pKey, *pVal, lifespan, lifespanUnit, maxIdle, maxIdleUnit);};
        return goAsync(f, success, fail);
    }

    std::future<V*> replaceAsync(const K& key, const V& oldVal, 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 f = [=]
        {   return this->replace(key, oldVal, val, lifespan, SECONDS, maxIdle, SECONDS);};
        return goAsync(f, success, fail);
    }

    std::future<V*> replaceAsync(const K& key, const V& oldVal, const V& val, uint64_t lifespan, TimeUnit lifespanUnit,
            std::function<V* (V*)> success = nullptr, std::function<V* (std::exception&)> fail = nullptr)
    {
        auto f = [=]
        {   return this->replace(key, oldVal, val, lifespan, lifespanUnit, 0, SECONDS);};
        return goAsync(f, success, fail);
    }

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

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

    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<bool> replaceWithVersionAsync(const K& key, const V& val, uint64_t version, uint64_t lifespan,
            uint64_t maxIdle, std::function<bool(bool)> success = nullptr, std::function<bool(std::exception&)> fail =
                    nullptr)
    {
        auto f = [=]
        {   return 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::future<bool> removeWithVersionAsync(const K& key, uint64_t version,
            std::function<bool(bool)> success = nullptr, std::function<bool(std::exception&)> fail = nullptr)
    {
        auto f = [=]
        {   return this->removeWithVersion(key, version);};
        return goAsync(f, success, fail);
    }

    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)
    {
        std::map<void*, void*> mbuf;
        std::map<std::shared_ptr<K>, std::shared_ptr<V> > result;
        base_getBulk(nrOfEntries, mbuf);
        for (auto it = mbuf.begin(); it != mbuf.end(); ++it)
        {
            result[std::shared_ptr<K>((K*) it->first)] = std::shared_ptr<V>((V*) it->second);
        }
        return result;
    }
    std::vector<unsigned char> execute(const std::string& name, const std::map<std::string, std::string>& args)
    {
        return base_execute(name, args);
    }

    std::vector<unsigned char> execute(const std::string& name, const std::map<std::vector<char>, std::vector<char> >& 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()
    {
        std::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()
    {
        std::map<std::string, std::string> statistics;
        base_stats(statistics);
        return statistics;
    }
    void clear()
    {
        base_clear();
    }

    std::future<void> clearAsync(std::function<void(void)> success = nullptr,
            std::function<void(std::exception&)> fail = nullptr)
    {
        auto f = [=]
        {   this->clear();};
        return goAsync(f, success, fail);
    }

    void ping()
    {
        base_ping();
    }

    void addClientListener(ClientListener& clientListener, std::vector<std::vector<char> > filterFactoryParams,
            std::vector<std::vector<char> > converterFactoryParams, const std::function<void()> &recoveryCallback =
                    nullptr)
    {
        base_addClientListener(clientListener, filterFactoryParams, converterFactoryParams, recoveryCallback);
    }

    void removeClientListener(ClientListener& clientListener)
    {
        base_removeClientListener(clientListener);
    }

#if !defined(SWIG) && !defined(SWIGCSHARP)

    void addContinuousQueryListener(ContinuousQueryListener<K, V>& cql)
    {
        base_addContinuousQueryListener(cql);
    }
    template<typename ... Params>
    void addContinuousQueryListener(ContinuousQueryListener<K, V, Params...>& cql)
    {
        base_addContinuousQueryListener(cql);
    }

    template<typename ... Params>
    void removeContinuousQueryListener(ContinuousQueryListener<Params...>& cql)
    {
        base_removeClientListener(cql.cl);
    }
#endif

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

    template <class M = JBossMarshaller>
    RemoteExecution<M> getRemoteExecution()
    {
        return RemoteExecution<M>(*this);
    }

    RemoteCache(const RemoteCache &other) :
            RemoteCacheBase(other), keyMarshaller(other.keyMarshaller), valueMarshaller(other.valueMarshaller)
    {
        setMarshallers(this, &keyMarshall, &valueMarshall, &keyUnmarshall, &valueUnmarshall);
        valueCopyConstructor = [](const void* src) { V* v = new V(*static_cast<const V*>(src)); return (void*) v; };
        valueDestructor = [](const void* obj) { delete static_cast<const V*>(obj); };
    }

    RemoteCache withDataFormat(DataFormat<K,V>* df) {
        RemoteCache rc(*this);
        rc.keyMarshaller = (df->keyMarshaller != nullptr) ? df->keyMarshaller : this->keyMarshaller;
        rc.valueMarshaller = (df->valueMarshaller != nullptr) ? df->valueMarshaller : this->valueMarshaller;
        rc.cloneImplWithDataFormat(df);
        return rc;
    }

    template <class K1, class V1> RemoteCache<K1,V1> withDataFormat(DataFormat<K1,V1>* df) {
        RemoteCache<K1,V1> rc((RemoteCacheBase&)*this);
        rc.keyMarshaller = df->keyMarshaller;
        rc.valueMarshaller = df->valueMarshaller;
        rc.cloneImplWithDataFormat(df);
    return rc;
    }

protected:

    RemoteCache(const RemoteCacheBase& rcb) : RemoteCacheBase(rcb){
        setRemoteCachePtr(this);
    }

    RemoteCache(TransactionManager& tm, TransactionTable& tt, bool forceReturnValue, bool transactional) :
            RemoteCacheBase(tm, tt, forceReturnValue, transactional)
    {
        valueCopyConstructor = [](const void* src) { V* v = new V(*static_cast<const V*>(src)); return (void*) v; };
        valueDestructor = [](const void* obj) { delete static_cast<const V*>(obj); };
        setMarshallers(this, &keyMarshall, &valueMarshall, &keyUnmarshall, &valueUnmarshall);
    }

private:
    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);
    }

    std::shared_ptr<Marshaller<K> > keyMarshaller;
    std::shared_ptr<Marshaller<V> > valueMarshaller;

    friend class RemoteCacheManager;
    template <class K1, class V1> friend class RemoteCache;
};
}
} // namespace

#endif  /* ISPN_HOTROD_REMOTECACHE_H */