/*
 * 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.
 */

#include <boost/test/unit_test.hpp>
#include <boost/chrono.hpp>
#include <boost/thread.hpp>

#include <ignite/ignition.h>
#include <ignite/test_utils.h>

using namespace ignite;
using namespace ignite::cache;
using namespace ignite::cluster;
using namespace ignite::compute;
using namespace ignite::common::concurrent;
using namespace ignite::impl;
using namespace ignite_test;

using namespace boost::unit_test;

enum { RETRIES_FOR_STABLE_TOPOLOGY = 5};

/*
 * Test setup fixture for cache affinity.
 */
struct ComputeTestSuiteFixtureAffinity
{
    static const char* cacheName;

    Ignite node0;
    Ignite node1;
    Ignite node2;

    Ignite MakeNode(const char* name)
    {
#ifdef IGNITE_TESTS_32
        const char* config = "cache-test-32.xml";
#else
        const char* config = "affinity-test.xml";
#endif
        return StartNode(config, name);
    }

    /*
     * Constructor.
     */
    ComputeTestSuiteFixtureAffinity()
    {
        node0 = MakeNode("ComputeAffinityNode0");

        BOOST_REQUIRE(WaitForRebalance0());

        node1 = MakeNode("ComputeAffinityNode1");

        BOOST_REQUIRE(WaitForRebalance1());

        node2 = MakeNode("ComputeAffinityNode2");

        BOOST_REQUIRE(WaitForRebalance2());
    }

    /*
     * Destructor.
     */
    ~ComputeTestSuiteFixtureAffinity()
    {
        Ignition::StopAll(false);
    }

    /**
     * Check whether rebalance is complete for the cluster.
     * @return true if complete.
     */
    bool IsRebalanceComplete0()
    {
        return
            node0.GetAffinity<int32_t>(cacheName).MapKeyToNode(0).IsLocal() &&
            node0.GetAffinity<int32_t>(cacheName).MapKeyToNode(1).IsLocal() &&
            node0.GetAffinity<int32_t>(cacheName).MapKeyToNode(6).IsLocal();
    }

    /**
     * Check whether rebalance is complete for the cluster.
     * @return true if complete.
     */
    bool IsRebalanceComplete1()
    {
        return
            node0.GetAffinity<int32_t>(cacheName).MapKeyToNode(0).IsLocal() &&
            node1.GetAffinity<int32_t>(cacheName).MapKeyToNode(1).IsLocal() &&
            node1.GetAffinity<int32_t>(cacheName).MapKeyToNode(6).IsLocal();
    }

    /**
     * Check whether rebalance is complete for the cluster.
     * @return true if complete.
     */
    bool IsRebalanceComplete2()
    {
        return
            node0.GetAffinity<int32_t>(cacheName).MapKeyToNode(0).IsLocal() &&
            node1.GetAffinity<int32_t>(cacheName).MapKeyToNode(1).IsLocal() &&
            node2.GetAffinity<int32_t>(cacheName).MapKeyToNode(6).IsLocal();
    }

    /**
     * Wait for rebalance.
     * @param timeout Timeout to wait.
     * @return True if condition was met, false if timeout has been reached.
     */
    bool WaitForRebalance0(int32_t timeout = 5000)
    {
        return WaitForCondition(boost::bind(&ComputeTestSuiteFixtureAffinity::IsRebalanceComplete0, this), timeout);
    }

    /**
     * Wait for rebalance.
     * @param timeout Timeout to wait.
     * @return True if condition was met, false if timeout has been reached.
     */
    bool WaitForRebalance1(int32_t timeout = 5000)
    {
        return WaitForCondition(boost::bind(&ComputeTestSuiteFixtureAffinity::IsRebalanceComplete1, this), timeout);
    }

    /**
     * Wait for rebalance.
     * @param timeout Timeout to wait.
     * @return True if condition was met, false if timeout has been reached.
     */
    bool WaitForRebalance2(int32_t timeout = 5000)
    {
        return WaitForCondition(boost::bind(&ComputeTestSuiteFixtureAffinity::IsRebalanceComplete2, this), timeout);
    }
};

const char* ComputeTestSuiteFixtureAffinity::cacheName = "test_backups_0";

/*
 * Test setup fixture.
 */
struct ComputeTestSuiteFixture
{
    Ignite node;

    Ignite MakeNode(const char* name)
    {
#ifdef IGNITE_TESTS_32
        const char* config = "cache-test-32.xml";
#else
        const char* config = "cache-test.xml";
#endif
        return StartNode(config, name);
    }

    /*
     * Constructor.
     */
    ComputeTestSuiteFixture() :
        node(MakeNode("ComputeNode1"))
    {
        // No-op.
    }

    /*
     * Destructor.
     */
    ~ComputeTestSuiteFixture()
    {
        Ignition::StopAll(true);
    }
};

/*
 * Test setup fixture for cluster group.
 */
struct ComputeTestSuiteFixtureClusterGroup
{
    enum NodeType {
        SERVER_NODE_ATTRIBUTE_VALUE0,
        SERVER_NODE_ATTRIBUTE_VALUE1,
        CLIENT_NODE,
    };

    Ignite server0;
    Ignite server1;
    Ignite server2;
    Ignite client;

    Ignite MakeNode(const char* name, NodeType type)
    {
        std::string config;

        switch (type) {
        case SERVER_NODE_ATTRIBUTE_VALUE0:
            config = "compute-server0.xml";
            break;

        case SERVER_NODE_ATTRIBUTE_VALUE1:
            config = "compute-server1.xml";
            break;

        case CLIENT_NODE:
            config = "compute-client.xml";
            break;
        }

#ifdef IGNITE_TESTS_32
        config.replace(config.begin() + config.find(".xml"), config.end(), "-32.xml");
#endif

        return StartNode(config.c_str(), name);
    }

    /*
     * Constructor.
     */
    ComputeTestSuiteFixtureClusterGroup() :
        server0(MakeNode("ServerNode0", SERVER_NODE_ATTRIBUTE_VALUE0)),
        server1(MakeNode("ServerNode1", SERVER_NODE_ATTRIBUTE_VALUE1)),
        server2(MakeNode("ServerNode2", SERVER_NODE_ATTRIBUTE_VALUE1)),
        client(MakeNode("ClientNode", CLIENT_NODE))
    {
        // No-op.
    }

    /*
     * Destructor.
     */
    ~ComputeTestSuiteFixtureClusterGroup()
    {
        Ignition::StopAll(true);
    }
};

struct Func1 : ComputeFunc<std::string>
{
    Func1() :
        a(), b(), err()
    {
        // No-op.
    }

    Func1(int32_t a, int32_t b) :
        a(a), b(b), err()
    {
        // No-op.
    }

    Func1(IgniteError err) :
        a(), b(), err(err)
    {
        // No-op.
    }

    virtual std::string Call()
    {
        if (err.GetCode() != IgniteError::IGNITE_SUCCESS)
            throw err;

        std::stringstream tmp;

        tmp << a << '.' << b;

        return tmp.str();
    }

    int32_t a;
    int32_t b;
    IgniteError err;
};

struct Func2 : ComputeFunc<std::string>
{
    Func2() :
        a(), b(), err()
    {
        // No-op.
    }

    Func2(int32_t a, int32_t b) :
        a(a), b(b), err()
    {
        // No-op.
    }

    Func2(IgniteError err) :
        a(), b(), err(err)
    {
        // No-op.
    }

    virtual std::string Call()
    {
        boost::this_thread::sleep_for(boost::chrono::milliseconds(200));

        if (err.GetCode() != IgniteError::IGNITE_SUCCESS)
            throw err;

        std::stringstream tmp;

        tmp << a << '.' << b;

        return tmp.str();
    }

    int32_t a;
    int32_t b;
    IgniteError err;
};

struct Func3 : ComputeFunc<void>
{
    Func3() :
        a(), b(), err()
    {
        // No-op.
    }

    Func3(int32_t a, int32_t b) :
        a(a), b(b), err()
    {
        // No-op.
    }

    Func3(IgniteError err) :
        a(), b(), err(err)
    {
        // No-op.
    }

    virtual void Call()
    {
        boost::this_thread::sleep_for(boost::chrono::milliseconds(200));

        if (err.GetCode() != IgniteError::IGNITE_SUCCESS)
            throw err;

        std::stringstream tmp;

        tmp << a << '.' << b;

        res = tmp.str();
    }

    int32_t a;
    int32_t b;
    IgniteError err;

    static std::string res;
};

std::string Func3::res;

void EmptyDeleter(IgniteEnvironment*)
{
    // No-op.
}

struct FuncAffinityCall : ComputeFunc<int32_t>
{
    FuncAffinityCall() :
        cacheName(), cacheKey(), err()
    {
        // No-op.
    }

    FuncAffinityCall(std::string cacheName, int32_t cacheKey) :
        cacheName(cacheName), cacheKey(cacheKey), err()
    {
        // No-op.
    }

    FuncAffinityCall(IgniteError err) :
        cacheName(), cacheKey(), err(err)
    {
        // No-op.
    }

    virtual int32_t Call()
    {
        Ignite& node = GetIgnite();

        Cache<int32_t, int32_t> cache = node.GetCache<int32_t, int32_t>(cacheName.c_str());

        return cache.LocalPeek(cacheKey, CachePeekMode::PRIMARY);
    }

    std::string cacheName;
    int32_t cacheKey;
    IgniteError err;
};

struct FuncAffinityRun : ComputeFunc<void>
{
    FuncAffinityRun() :
        cacheName(), cacheKey(), err()
    {
        // No-op.
    }

    FuncAffinityRun(std::string cacheName, int32_t cacheKey, int32_t checkKey) :
        cacheName(cacheName), cacheKey(cacheKey), checkKey(checkKey), err()
    {
        // No-op.
    }

    FuncAffinityRun(IgniteError err) :
        cacheName(), cacheKey(), err(err)
    {
        // No-op.
    }

    virtual void Call()
    {
        Ignite& node = GetIgnite();
        Cache<int32_t, int32_t> cache = node.GetCache<int32_t, int32_t>(cacheName.c_str());

        int32_t res = cache.LocalPeek(cacheKey, CachePeekMode::PRIMARY);
        cache.Put(checkKey, res);
        res = cache.Get(checkKey);
    }

    std::string cacheName;
    int32_t cacheKey;
    int32_t checkKey;
    IgniteError err;
};

namespace ignite
{
    namespace binary
    {
        template<>
        struct BinaryType<Func1> : BinaryTypeDefaultAll<Func1>
        {
            static void GetTypeName(std::string& dst)
            {
                dst = "Func1";
            }

            static void Write(BinaryWriter& writer, const Func1& obj)
            {
                writer.WriteInt32("a", obj.a);
                writer.WriteInt32("b", obj.b);
                writer.WriteObject<IgniteError>("err", obj.err);
            }

            static void Read(BinaryReader& reader, Func1& dst)
            {
                dst.a = reader.ReadInt32("a");
                dst.b = reader.ReadInt32("b");
                dst.err = reader.ReadObject<IgniteError>("err");
            }
        };

        template<>
        struct BinaryType<Func2> : BinaryTypeDefaultAll<Func2>
        {
            static void GetTypeName(std::string& dst)
            {
                dst = "Func2";
            }

            static void Write(BinaryWriter& writer, const Func2& obj)
            {
                writer.WriteInt32("a", obj.a);
                writer.WriteInt32("b", obj.b);
                writer.WriteObject<IgniteError>("err", obj.err);
            }

            static void Read(BinaryReader& reader, Func2& dst)
            {
                dst.a = reader.ReadInt32("a");
                dst.b = reader.ReadInt32("b");
                dst.err = reader.ReadObject<IgniteError>("err");
            }
        };

        template<>
        struct BinaryType<Func3> : BinaryTypeDefaultAll<Func3>
        {
            static void GetTypeName(std::string& dst)
            {
                dst = "Func3";
            }

            static void Write(BinaryWriter& writer, const Func3& obj)
            {
                writer.WriteInt32("a", obj.a);
                writer.WriteInt32("b", obj.b);
                writer.WriteObject<IgniteError>("err", obj.err);
            }

            static void Read(BinaryReader& reader, Func3& dst)
            {
                dst.a = reader.ReadInt32("a");
                dst.b = reader.ReadInt32("b");
                dst.err = reader.ReadObject<IgniteError>("err");
            }
        };

        template<>
        struct BinaryType<FuncAffinityCall> : BinaryTypeDefaultAll<FuncAffinityCall>
        {
            static void GetTypeName(std::string& dst)
            {
                dst = "FuncAffinityCall";
            }

            static void Write(BinaryWriter& writer, const FuncAffinityCall& obj)
            {
                writer.WriteString("cacheName", obj.cacheName);
                writer.WriteInt32("cacheKey", obj.cacheKey);
                writer.WriteObject<IgniteError>("err", obj.err);
            }

            static void Read(BinaryReader& reader, FuncAffinityCall& dst)
            {
                dst.cacheName = reader.ReadString("cacheName");
                dst.cacheKey = reader.ReadInt32("cacheKey");
                dst.err = reader.ReadObject<IgniteError>("err");
            }
        };

        template<>
        struct BinaryType<FuncAffinityRun> : BinaryTypeDefaultAll<FuncAffinityRun>
        {
            static void GetTypeName(std::string& dst)
            {
                dst = "FuncAffinityRun";
            }

            static void Write(BinaryWriter& writer, const FuncAffinityRun& obj)
            {
                writer.WriteString("cacheName", obj.cacheName);
                writer.WriteInt32("cacheKey", obj.cacheKey);
                writer.WriteInt32("checkKey", obj.checkKey);
                writer.WriteObject<IgniteError>("err", obj.err);
            }

            static void Read(BinaryReader& reader, FuncAffinityRun& dst)
            {
                dst.cacheName = reader.ReadString("cacheName");
                dst.cacheKey = reader.ReadInt32("cacheKey");
                dst.checkKey = reader.ReadInt32("checkKey");
                dst.err = reader.ReadObject<IgniteError>("err");
            }
        };
    }
}

IGNITE_EXPORTED_CALL void IgniteModuleInit1(IgniteBindingContext& context)
{
    IgniteBinding binding = context.GetBinding();

    binding.RegisterComputeFunc<Func1>();
    binding.RegisterComputeFunc<Func2>();
    binding.RegisterComputeFunc<Func3>();
    binding.RegisterComputeFunc<FuncAffinityCall>();
    binding.RegisterComputeFunc<FuncAffinityRun>();
}

template<typename TK>
std::vector<int32_t> GetPrimaryKeys(size_t num, ClusterNode& node, CacheAffinity<TK>& affinity)
{
    std::vector<int32_t> ret;

    if (!num)
        return ret;

    for (int32_t i = 0; i < INT_MAX; i++)
    {
        if (affinity.IsPrimary(node, i)) {
            ret.push_back(i);

            if (ret.size() >= num)
                return ret;
        }
    }

    BOOST_CHECK(false);

    return ret;
}

BOOST_FIXTURE_TEST_SUITE(ComputeTestSuiteAffinity, ComputeTestSuiteFixtureAffinity)

BOOST_AUTO_TEST_CASE(IgniteAffinityCall)
{
    const int32_t key = 100;
    const int32_t value = 500;

    Cache<int32_t, int32_t> cache = node0.GetCache<int32_t, int32_t>(cacheName);

    cache.Put(key, value);

    CacheAffinity<int> affinity = node0.GetAffinity<int32_t>(cache.GetName());
    Compute compute = node0.GetCompute();

    ClusterNode clusterNode0 = affinity.MapKeyToNode(100);

    BOOST_TEST_CHECKPOINT("Starting local calls loop");

    std::vector<int32_t> aKeys = GetPrimaryKeys(10, clusterNode0, affinity);
    for (size_t i = 0; i < aKeys.size(); i++)
    {
        int32_t res = compute.AffinityCall<int32_t>(cache.GetName(), aKeys[i],
            FuncAffinityCall(cache.GetName(), key));

        BOOST_CHECK_EQUAL(res, value);
    }

    ClusterNode clusterNode1 = node0.GetCluster().GetLocalNode();

    if (clusterNode0.GetId() == clusterNode1.GetId())
        clusterNode1 = node1.GetCluster().GetLocalNode();

    BOOST_REQUIRE_NE(clusterNode0.GetId(), clusterNode1.GetId());
    BOOST_REQUIRE(!affinity.IsPrimary(clusterNode1, key));

    BOOST_TEST_CHECKPOINT("Starting remote calls loop");

    aKeys = GetPrimaryKeys(10, clusterNode1, affinity);
    for (size_t i = 0; i < aKeys.size(); i++)
    {
        int32_t res = compute.AffinityCall<int32_t>(cache.GetName(), aKeys[i],
            FuncAffinityCall(cache.GetName(), key));

        BOOST_CHECK_EQUAL(res, 0);
    }
}

BOOST_AUTO_TEST_CASE(IgniteAffinityCallAsync)
{
    const int32_t key = 100;
    const int32_t value = 500;

    Cache<int32_t, int32_t> cache = node0.GetCache<int32_t, int32_t>(cacheName);

    cache.Put(key, value);

    CacheAffinity<int> affinity = node0.GetAffinity<int32_t>(cache.GetName());
    Compute compute = node0.GetCompute();

    ClusterNode clusterNode0 = affinity.MapKeyToNode(100);

    BOOST_TEST_CHECKPOINT("Starting calls loop");

    std::vector<int32_t> aKeys = GetPrimaryKeys(10, clusterNode0, affinity);
    for (size_t i = 0; i < aKeys.size(); i++)
    {
        Future<int32_t> res = compute.AffinityCallAsync<int32_t>(cache.GetName(), aKeys[i],
            FuncAffinityCall(cache.GetName(), key));

        int32_t resVal = res.GetValue();
        BOOST_CHECK_EQUAL(value, resVal);
    }

    ClusterNode clusterNode1 = node0.GetCluster().GetLocalNode();

    if (clusterNode0.GetId() == clusterNode1.GetId())
        clusterNode1 = node1.GetCluster().GetLocalNode();

    BOOST_REQUIRE_NE(clusterNode0.GetId(), clusterNode1.GetId());
    BOOST_REQUIRE(!affinity.IsPrimary(clusterNode1, key));

    BOOST_TEST_CHECKPOINT("Starting remote calls loop");

    aKeys = GetPrimaryKeys(10, clusterNode1, affinity);
    for (size_t i = 0; i < aKeys.size(); i++)
    {
        Future<int32_t> res = compute.AffinityCallAsync<int32_t>(cache.GetName(), aKeys[i],
            FuncAffinityCall(cache.GetName(), key));

        int32_t resVal = res.GetValue();
        BOOST_CHECK_EQUAL(0, resVal);
    }
}

BOOST_AUTO_TEST_CASE(IgniteAffinityRun)
{
    const int32_t key = 100;
    const int32_t checkKey = -1;
    const int32_t value = 500;

    Cache<int32_t, int32_t> cache = node0.GetCache<int32_t, int32_t>(cacheName);

    cache.Put(key, value);

    CacheAffinity<int> affinity = node0.GetAffinity<int32_t>(cache.GetName());
    Compute compute = node0.GetCompute();

    ClusterNode clusterNode0 = affinity.MapKeyToNode(100);

    BOOST_TEST_CHECKPOINT("Starting calls loop");

    std::vector<int32_t> aKeys = GetPrimaryKeys(10, clusterNode0, affinity);
    for (size_t i = 0; i < aKeys.size(); i++)
    {
        compute.AffinityRun(cache.GetName(), aKeys[i],
            FuncAffinityRun(cache.GetName(), key, checkKey));

        int32_t resVal = cache.Get(checkKey);
        BOOST_CHECK_EQUAL(500, resVal);
    }

    ClusterNode clusterNode1 = node0.GetCluster().GetLocalNode();

    if (clusterNode0.GetId() == clusterNode1.GetId())
        clusterNode1 = node1.GetCluster().GetLocalNode();

    BOOST_REQUIRE_NE(clusterNode0.GetId(), clusterNode1.GetId());
    BOOST_REQUIRE(!affinity.IsPrimary(clusterNode1, key));

    BOOST_TEST_CHECKPOINT("Starting remote calls loop");

    aKeys = GetPrimaryKeys(10, clusterNode1, affinity);
    for (size_t i = 0; i < aKeys.size(); i++)
    {
        compute.AffinityRun(cache.GetName(), aKeys[i],
            FuncAffinityRun(cache.GetName(), key, checkKey));

        int32_t resVal = cache.Get(checkKey);
        BOOST_CHECK_EQUAL(0, resVal);
    }
}

BOOST_AUTO_TEST_CASE(IgniteAffinityRunAsync)
{
    const int32_t key = 100;
    const int32_t checkKey = -1;
    const int32_t value = 500;

    Cache<int32_t, int32_t> cache = node0.GetCache<int32_t, int32_t>(cacheName);

    cache.Put(key, value);

    CacheAffinity<int> affinity = node0.GetAffinity<int32_t>(cache.GetName());
    Compute compute = node0.GetCompute();

    ClusterNode clusterNode0 = affinity.MapKeyToNode(100);

    BOOST_TEST_CHECKPOINT("Starting calls loop");

    std::vector<int32_t> aKeys = GetPrimaryKeys(10, clusterNode0, affinity);
    for (size_t i = 0; i < aKeys.size(); i++)
    {
        Future<void> res = compute.AffinityRunAsync(cache.GetName(), aKeys[i],
            FuncAffinityRun(cache.GetName(), key, checkKey));

        res.GetValue();

        int32_t resVal = cache.Get(checkKey);
        BOOST_CHECK_EQUAL(500, resVal);
    }

    ClusterNode clusterNode1 = node0.GetCluster().GetLocalNode();

    if (clusterNode0.GetId() == clusterNode1.GetId())
        clusterNode1 = node1.GetCluster().GetLocalNode();

    BOOST_REQUIRE_NE(clusterNode0.GetId(), clusterNode1.GetId());
    BOOST_REQUIRE(!affinity.IsPrimary(clusterNode1, key));

    BOOST_TEST_CHECKPOINT("Starting remote calls loop");

    aKeys = GetPrimaryKeys(10, clusterNode1, affinity);
    for (size_t i = 0; i < aKeys.size(); i++)
    {
        Future<void> res = compute.AffinityRunAsync<int32_t>(cache.GetName(), aKeys[i],
            FuncAffinityRun(cache.GetName(), key, checkKey));

        res.GetValue();

        int32_t resVal = cache.Get(checkKey);
        BOOST_CHECK_EQUAL(0, resVal);
    }
}

BOOST_AUTO_TEST_SUITE_END()

BOOST_FIXTURE_TEST_SUITE(ComputeTestSuite, ComputeTestSuiteFixture)

BOOST_AUTO_TEST_CASE(IgniteCallSyncLocal)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Making Call");
    std::string res = compute.Call<std::string>(Func1(8, 5));

    BOOST_CHECK_EQUAL(res, "8.5");
}

BOOST_AUTO_TEST_CASE(IgniteCallAsyncLocal)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Making Call");
    Future<std::string> res = compute.CallAsync<std::string>(Func2(312, 245));

    BOOST_CHECK(!res.IsReady());

    BOOST_TEST_CHECKPOINT("Waiting with timeout");
    res.WaitFor(100);

    BOOST_CHECK(!res.IsReady());

    BOOST_CHECK_EQUAL(res.GetValue(), "312.245");
}

BOOST_AUTO_TEST_CASE(IgniteCallSyncLocalError)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Making Call");

    BOOST_CHECK_EXCEPTION(compute.Call<std::string>(Func1(MakeTestError())), IgniteError, IsTestError);
}

BOOST_AUTO_TEST_CASE(IgniteCallAsyncLocalError)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Making Call");
    Future<std::string> res = compute.CallAsync<std::string>(Func2(MakeTestError()));

    BOOST_CHECK(!res.IsReady());

    BOOST_TEST_CHECKPOINT("Waiting with timeout");
    res.WaitFor(100);

    BOOST_CHECK(!res.IsReady());

    BOOST_CHECK_EXCEPTION(res.GetValue(), IgniteError, IsTestError);
}

BOOST_AUTO_TEST_CASE(IgniteCallTestRemote)
{
    Ignite node2 = MakeNode("ComputeNode2");
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Making Call");
    compute.CallAsync<std::string>(Func2(8, 5));

    std::string res = compute.Call<std::string>(Func1(42, 24));

    BOOST_CHECK_EQUAL(res, "42.24");
}

BOOST_AUTO_TEST_CASE(IgniteCallTestRemoteError)
{
    Ignite node2 = MakeNode("ComputeNode2");
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Making Call");
    compute.CallAsync<std::string>(Func2(8, 5));

    Future<std::string> res = compute.CallAsync<std::string>(Func2(MakeTestError()));

    BOOST_CHECK(!res.IsReady());

    BOOST_TEST_CHECKPOINT("Waiting with timeout");
    res.WaitFor(100);

    BOOST_CHECK(!res.IsReady());

    BOOST_CHECK_EXCEPTION(res.GetValue(), IgniteError, IsTestError);
}

BOOST_AUTO_TEST_CASE(IgniteRunSyncLocal)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Running");
    compute.Run(Func3(8, 5));

    BOOST_CHECK_EQUAL(Func3::res, "8.5");
}

BOOST_AUTO_TEST_CASE(IgniteRunAsyncLocal)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Running");
    Future<void> res = compute.RunAsync(Func3(312, 245));

    BOOST_CHECK(!res.IsReady());

    BOOST_TEST_CHECKPOINT("Waiting with timeout");
    res.WaitFor(100);

    BOOST_CHECK(!res.IsReady());

    res.GetValue();

    BOOST_CHECK_EQUAL(Func3::res, "312.245");
}

BOOST_AUTO_TEST_CASE(IgniteRunSyncLocalError)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Running");

    BOOST_CHECK_EXCEPTION(compute.Run(Func3(MakeTestError())), IgniteError, IsTestError);
}

BOOST_AUTO_TEST_CASE(IgniteRunAsyncLocalError)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Running");
    Future<void> res = compute.RunAsync(Func3(MakeTestError()));

    BOOST_CHECK(!res.IsReady());

    BOOST_TEST_CHECKPOINT("Waiting with timeout");
    res.WaitFor(100);

    BOOST_CHECK(!res.IsReady());

    BOOST_CHECK_EXCEPTION(res.GetValue(), IgniteError, IsTestError);
}

BOOST_AUTO_TEST_CASE(IgniteRunRemote)
{
    Ignite node2 = MakeNode("ComputeNode2");
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Running");
    compute.CallAsync<std::string>(Func2(8, 5));

    compute.Run(Func3(42, 24));

    BOOST_CHECK_EQUAL(Func3::res, "42.24");
}

BOOST_AUTO_TEST_CASE(IgniteRunRemoteError)
{
    Ignite node2 = MakeNode("ComputeNode2");
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Running");
    compute.CallAsync<std::string>(Func2(8, 5));

    Future<void> res = compute.RunAsync(Func3(MakeTestError()));

    BOOST_CHECK(!res.IsReady());

    BOOST_TEST_CHECKPOINT("Waiting with timeout");
    res.WaitFor(100);

    BOOST_CHECK(!res.IsReady());

    BOOST_CHECK_EXCEPTION(res.GetValue(), IgniteError, IsTestError);
}

BOOST_AUTO_TEST_CASE(IgniteBroadcastLocalSync)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Broadcasting");
    std::vector<std::string> res = compute.Broadcast<std::string>(Func2(8, 5));

    BOOST_CHECK_EQUAL(res.size(), 1);
    BOOST_CHECK_EQUAL(res[0], "8.5");
}

BOOST_AUTO_TEST_CASE(IgniteBroadcastLocalAsync)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Broadcasting");
    Future< std::vector<std::string> > res = compute.BroadcastAsync<std::string>(Func2(312, 245));

    BOOST_CHECK(!res.IsReady());

    BOOST_TEST_CHECKPOINT("Waiting with timeout");
    res.WaitFor(100);

    BOOST_CHECK(!res.IsReady());

    std::vector<std::string> value = res.GetValue();

    BOOST_CHECK_EQUAL(value.size(), 1);
    BOOST_CHECK_EQUAL(value[0], "312.245");
}

BOOST_AUTO_TEST_CASE(IgniteBroadcastSyncLocalError)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Broadcasting");

    BOOST_CHECK_EXCEPTION(compute.Broadcast(Func2(MakeTestError())), IgniteError, IsTestError);
}

BOOST_AUTO_TEST_CASE(IgniteBroadcastAsyncLocalError)
{
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Broadcasting");
    Future<void> res = compute.BroadcastAsync(Func2(MakeTestError()));

    BOOST_CHECK(!res.IsReady());

    BOOST_TEST_CHECKPOINT("Waiting with timeout");
    res.WaitFor(100);

    BOOST_CHECK(!res.IsReady());

    BOOST_CHECK_EXCEPTION(res.GetValue(), IgniteError, IsTestError);
}

BOOST_AUTO_TEST_CASE(IgniteBroadcastRemote)
{
    Ignite node2 = MakeNode("ComputeNode2");
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Broadcasting");
    std::vector<std::string> res = compute.Broadcast<std::string>(Func2(8, 5));

    BOOST_CHECK_EQUAL(res.size(), 2);
    BOOST_CHECK_EQUAL(res[0], "8.5");
    BOOST_CHECK_EQUAL(res[1], "8.5");
}

BOOST_AUTO_TEST_CASE(IgniteBroadcastRemoteError)
{
    Ignite node2 = MakeNode("ComputeNode2");
    Compute compute = node.GetCompute();

    BOOST_TEST_CHECKPOINT("Broadcasting");
    Future< std::vector<std::string> > res = compute.BroadcastAsync<std::string>(Func2(MakeTestError()));

    BOOST_CHECK(!res.IsReady());

    BOOST_TEST_CHECKPOINT("Waiting with timeout");
    res.WaitFor(100);

    BOOST_CHECK(!res.IsReady());

    BOOST_CHECK_EXCEPTION(res.GetValue(), IgniteError, IsTestError);
}

BOOST_AUTO_TEST_SUITE_END()

BOOST_FIXTURE_TEST_SUITE(ComputeTestSuiteClusterGroup, ComputeTestSuiteFixtureClusterGroup)

BOOST_AUTO_TEST_CASE(IgniteGetClusterGroupForServers)
{
    ClusterGroup localGroup = client.GetCluster().AsClusterGroup();
    ClusterGroup group = localGroup.ForServers();

    Compute compute = client.GetCompute(group);

    BOOST_TEST_CHECKPOINT("Broadcasting");
    std::vector<std::string> res = compute.Broadcast<std::string>(Func2(8, 5));

    BOOST_CHECK_EQUAL(res.size(), 3);
    BOOST_CHECK_EQUAL(res[0], "8.5");
    BOOST_CHECK_EQUAL(res[1], "8.5");
    BOOST_CHECK_EQUAL(res[2], "8.5");
}

BOOST_AUTO_TEST_CASE(IgniteGetClusterGroupForAttribute)
{
    ClusterGroup localGroup = client.GetCluster().AsClusterGroup();
    ClusterGroup group1 = localGroup.ForAttribute("TestAttribute", "Value0");
    ClusterGroup group2 = localGroup.ForAttribute("TestAttribute", "Value1");

    Compute compute1 = client.GetCompute(group1);
    Compute compute2 = client.GetCompute(group2);

    BOOST_TEST_CHECKPOINT("Broadcasting1");
    std::vector<std::string> res1 = compute1.Broadcast<std::string>(Func2(8, 5));

    BOOST_CHECK_EQUAL(res1.size(), 1);
    BOOST_CHECK_EQUAL(res1[0], "8.5");

    BOOST_TEST_CHECKPOINT("Broadcasting2");
    std::vector<std::string> res2 = compute2.Broadcast<std::string>(Func2(8, 5));

    BOOST_CHECK_EQUAL(res2.size(), 2);
    BOOST_CHECK_EQUAL(res2[0], "8.5");
    BOOST_CHECK_EQUAL(res2[1], "8.5");
}

BOOST_AUTO_TEST_SUITE_END()