/*
* All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
* its licensors.
*
* For complete copyright and license terms please see the LICENSE at the root of this
* distribution (the "License"). All use of this software is governed by the License,
* or, if provided, by the license below or the license accompanying this file. Do not
* remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
*
*/
#include "Tests.h"

#include <AzCore/std/parallel/thread.h>
#include <AzCore/std/smart_ptr/scoped_ptr.h>

#include <GridMate/Carrier/DefaultSimulator.h>

#include <GridMate/Containers/unordered_map.h>
#include <GridMate/Containers/unordered_set.h>
#include <GridMate/Containers/vector.h>

#include <GridMate/Replica/Interpolators.h>
#include <GridMate/Replica/Replica.h>
#include <GridMate/Replica/ReplicaFunctions.h>
#include <GridMate/Replica/ReplicaMgr.h>
#include <GridMate/Replica/ReplicaDrillerEvents.h>
#include <GridMate/Replica/BasicHostChunkDescriptor.h>

#include <GridMate/Serialize/DataMarshal.h>
#include <GridMate/Serialize/CompressionMarshal.h>
#include <AzCore/std/smart_ptr/unique_ptr.h>

using namespace GridMate;

namespace UnitTest {

#define GM_REPLICA_TEST_SESSION_CHANNEL 1

// Example: DataSet<int> Data1 = { "Data1", 1 };
#define MAKE_DATASET(N) DataSet<int> Data ## N = { "Data" #N, N };

// Example: DataSet<bool> Data1 = { "Data1", false };
#define MAKE_BOOLDATASET(N) DataSet<bool> Data ## N = { "Data" #N, false }

// Example: DataSet<bool> Data1 = { "Data1", false };
#define MAKE_U8DATASET(N) DataSet<AZ::u8> Data ## N = { "Data" #N, 0 }

template<typename T1>
class IntrospectableRPC
    : public GridMate::Rpc<T1>
{
public:
    typedef Internal::VariadicStorage<T1> TypeTuple;

    template<class C, bool (C::* FuncPtr)(typename T1::Type, const RpcContext&), class Traits = RpcDefaultTraits>
    struct BindInterface
        : public Internal::RpcBindArgsWrapper<T1>::template BindInterface<Traits, TypeTuple, Internal::InterfaceResolver, C, FuncPtr>
    {
        typedef typename Internal::RpcBindArgsWrapper<T1>::template BindInterface<Traits, TypeTuple, Internal::InterfaceResolver, C, FuncPtr> ParentType;
        BindInterface(const char* debugName)
            : ParentType(debugName) { }
        typename ParentType::MarshalerSetType& GetMarshalers() { return ParentType::m_marshalers; }
    };
};

class RPCChunk
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(RPCChunk);
    typedef AZStd::intrusive_ptr<RPCChunk> Ptr;
    static const char* GetChunkName() { return "RPCChunk"; }

    RPCChunk()
        : m_fromMasterBroadcast(0)
        , m_fromMasterNotBroadcast(0)
        , m_fromProxyBroadcast(0)
        , m_fromProxyNotBroadcast(0)
        , FromMasterBroadcast("FromMasterBroadcast")
        , FromMasterNotBroadcast("FromMasterNotBroadcast")
        , FromProxyBroadcast("FromProxyBroadcast")
        , FromProxyNotBroadcast("FromProxyNotBroadcast")
        , BroadcastInt("BroadcastInt")
    { }

    bool IsReplicaMigratable() override { return false; }

    bool FromMasterBroadcastFn(const RpcContext&)
    {
        AZ_TracePrintf("GridMate", "Executed FromMasterBroadcast %d %s\n", GetReplicaId(), GetReplica()->IsMaster() ? "master" : "proxy");
        m_fromMasterBroadcast++;
        return true;
    }

    bool FromMasterNotBroadcastFn(const RpcContext&)
    {
        AZ_TracePrintf("GridMate", "Executed FromMasterNotBroadcast %d %s\n", GetReplicaId(), GetReplica()->IsMaster() ? "master" : "proxy");
        m_fromMasterNotBroadcast++;
        return false;
    }

    bool FromProxyBroadcastFn(const RpcContext&)
    {
        AZ_TracePrintf("GridMate", "Executed FromProxyBroadcast %d %s\n", GetReplicaId(), GetReplica()->IsMaster() ? "master" : "proxy");
        m_fromProxyBroadcast++;
        return true;
    }

    bool FromProxyNotBroadcastFn(const RpcContext&)
    {
        AZ_TracePrintf("GridMate", "Executed FromProxyNotBroadcast %d %s\n", GetReplicaId(), GetReplica()->IsMaster() ? "master" : "proxy");
        m_fromProxyNotBroadcast++;
        return false;
    }

    bool BroadcastIntFn(int val, const RpcContext&)
    {
        m_sentData.push_back(val);
        return true;
    }

    int m_fromMasterBroadcast;
    int m_fromMasterNotBroadcast;
    int m_fromProxyBroadcast;
    int m_fromProxyNotBroadcast;
    AZStd::vector<int> m_sentData;

    Rpc<>::BindInterface<RPCChunk, & RPCChunk::FromMasterBroadcastFn> FromMasterBroadcast;
    Rpc<>::BindInterface<RPCChunk, & RPCChunk::FromMasterNotBroadcastFn> FromMasterNotBroadcast;
    Rpc<>::BindInterface<RPCChunk, & RPCChunk::FromProxyBroadcastFn> FromProxyBroadcast;
    Rpc<>::BindInterface<RPCChunk, & RPCChunk::FromProxyNotBroadcastFn> FromProxyNotBroadcast;
    Rpc<RpcArg<int> >::BindInterface<RPCChunk, & RPCChunk::BroadcastIntFn> BroadcastInt;
};

class FullRPCChunk
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(FullRPCChunk);
    typedef AZStd::intrusive_ptr<FullRPCChunk> Ptr;
    static const char* GetChunkName() { return "FullRPCChunk"; }

    FullRPCChunk()
        : ZeroRPC("ZeroRPC")
        , OneRPC("OneRPC")
        , TwoRPC("TwoRPC")
        , ThreeRPC("ThreeRPC")
        , FourRPC("FourRPC")
        , FiveRPC("FiveRPC")
        , SixRPC("SixRPC")
        , SevenRPC("SevenRPC")
        , EightRPC("EightRPC")
        , NineRPC("NineRPC")
    { }

    bool IsReplicaMigratable() override { return false; }

    bool Zero(const RpcContext&)
    {
        auto& list = (IsMaster() ? m_sentData : m_receivedData)[0];
        (void) list;
        return true;
    }

    bool One(AZ::u32 t1, const RpcContext&)
    {
        auto& list = (IsMaster() ? m_sentData : m_receivedData)[1];
        list.push_back(t1);
        return true;
    }

    bool Two(AZ::u32 t1, AZ::u32 t2, const RpcContext&)
    {
        auto& list = (IsMaster() ? m_sentData : m_receivedData)[2];
        list.push_back(t1);
        list.push_back(t2);
        return true;
    }

    bool Three(AZ::u32 t1, AZ::u32 t2, AZ::u32 t3, const RpcContext&)
    {
        auto& list = (IsMaster() ? m_sentData : m_receivedData)[3];
        list.push_back(t1);
        list.push_back(t2);
        list.push_back(t3);
        return true;
    }

    bool Four(AZ::u32 t1, AZ::u32 t2, AZ::u32 t3, AZ::u32 t4, const RpcContext&)
    {
        auto& list = (IsMaster() ? m_sentData : m_receivedData)[4];
        list.push_back(t1);
        list.push_back(t2);
        list.push_back(t3);
        list.push_back(t4);
        return true;
    }

    bool Five(AZ::u32 t1, AZ::u32 t2, AZ::u32 t3, AZ::u32 t4, AZ::u32 t5, const RpcContext&)
    {
        auto& list = (IsMaster() ? m_sentData : m_receivedData)[5];
        list.push_back(t1);
        list.push_back(t2);
        list.push_back(t3);
        list.push_back(t4);
        list.push_back(t5);
        return true;
    }

    bool Six(AZ::u32 t1, AZ::u32 t2, AZ::u32 t3, AZ::u32 t4, AZ::u32 t5, AZ::u32 t6, const RpcContext&)
    {
        auto& list = (IsMaster() ? m_sentData : m_receivedData)[6];
        list.push_back(t1);
        list.push_back(t2);
        list.push_back(t3);
        list.push_back(t4);
        list.push_back(t5);
        list.push_back(t6);
        return true;
    }

    bool Seven(AZ::u32 t1, AZ::u32 t2, AZ::u32 t3, AZ::u32 t4, AZ::u32 t5, AZ::u32 t6, AZ::u32 t7, const RpcContext&)
    {
        auto& list = (IsMaster() ? m_sentData : m_receivedData)[7];
        list.push_back(t1);
        list.push_back(t2);
        list.push_back(t3);
        list.push_back(t4);
        list.push_back(t5);
        list.push_back(t6);
        list.push_back(t7);
        return true;
    }

    bool Eight(AZ::u32 t1, AZ::u32 t2, AZ::u32 t3, AZ::u32 t4, AZ::u32 t5, AZ::u32 t6, AZ::u32 t7, AZ::u32 t8, const RpcContext&)
    {
        auto& list = (IsMaster() ? m_sentData : m_receivedData)[8];
        list.push_back(t1);
        list.push_back(t2);
        list.push_back(t3);
        list.push_back(t4);
        list.push_back(t5);
        list.push_back(t6);
        list.push_back(t7);
        list.push_back(t8);
        return true;
    }

    bool Nine(AZ::u32 t1, AZ::u32 t2, AZ::u32 t3, AZ::u32 t4, AZ::u32 t5, AZ::u32 t6, AZ::u32 t7, AZ::u32 t8, AZ::u32 t9, const RpcContext&)
    {
        auto& list = (IsMaster() ? m_sentData : m_receivedData)[9];
        list.push_back(t1);
        list.push_back(t2);
        list.push_back(t3);
        list.push_back(t4);
        list.push_back(t5);
        list.push_back(t6);
        list.push_back(t7);
        list.push_back(t8);
        list.push_back(t9);
        return true;
    }

    unordered_map<AZ::u32, vector<AZ::u32> > m_sentData;
    unordered_map<AZ::u32, vector<AZ::u32> > m_receivedData;

    Rpc<>::BindInterface<FullRPCChunk, & FullRPCChunk::Zero> ZeroRPC;
    Rpc<RpcArg<AZ::u32> >::BindInterface<FullRPCChunk, & FullRPCChunk::One> OneRPC;
    Rpc<RpcArg<AZ::u32>, RpcArg<AZ::u32> >::BindInterface<FullRPCChunk, & FullRPCChunk::Two> TwoRPC;
    Rpc<RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32> >::BindInterface<FullRPCChunk, & FullRPCChunk::Three> ThreeRPC;
    Rpc<RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32> >::BindInterface<FullRPCChunk, & FullRPCChunk::Four> FourRPC;
    Rpc<RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32> >::BindInterface<FullRPCChunk, & FullRPCChunk::Five> FiveRPC;
    Rpc<RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32> >::BindInterface<FullRPCChunk, & FullRPCChunk::Six> SixRPC;
    Rpc<RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32> >::BindInterface<FullRPCChunk, & FullRPCChunk::Seven> SevenRPC;
    Rpc<RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32> >::BindInterface<FullRPCChunk, & FullRPCChunk::Eight> EightRPC;
    Rpc<RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32>, RpcArg<AZ::u32> >::BindInterface<FullRPCChunk, & FullRPCChunk::Nine> NineRPC;
};

class DataSetChunk
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(DataSetChunk);

    DataSetChunk()
        : Data1("Data1", 0)
    {
    }

    typedef AZStd::intrusive_ptr<DataSetChunk> Ptr;
    static const char* GetChunkName() { return "DataSetChunk"; }

    bool IsReplicaMigratable() override { return false; }

    void IntHandler(const int& val, const TimeContext&)
    {
        m_changedData.push_back(val);
    }

    AZStd::vector<int> m_changedData;

    DataSet<int>::BindInterface<DataSetChunk, & DataSetChunk::IntHandler> Data1;
};

class MixedTestChunk
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(MixedTestChunk);

    MixedTestChunk()
    {
    }

    typedef AZStd::intrusive_ptr<MixedTestChunk> Ptr;
    static const char* GetChunkName() { return "MixedTestChunk"; }

    bool IsReplicaMigratable() override { return false; }

    DataSet<AZ::u64> Data1 = { "Data1", 42 };
    DataSet<AZ::u64> Data2 = { "Data2", 0 };
};

class LargeChunkWithDefaultsMedium
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(LargeChunkWithDefaultsMedium);

    LargeChunkWithDefaultsMedium()
    {
    }

    typedef AZStd::intrusive_ptr<LargeChunkWithDefaultsMedium> Ptr;
    static const char* GetChunkName() { return "LargeChunkWithDefaultsMedium"; }
    bool IsReplicaMigratable() override { return false; }

    MAKE_DATASET(1)
        MAKE_DATASET(2)
        MAKE_DATASET(3)
        MAKE_DATASET(4)
        MAKE_DATASET(5)
        MAKE_DATASET(6)
        MAKE_DATASET(7)
        MAKE_DATASET(8)
        MAKE_DATASET(9)
        MAKE_DATASET(10)
};

class ChunkWithBools
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(ChunkWithBools);

    ChunkWithBools()
    {
    }

    typedef AZStd::intrusive_ptr<ChunkWithBools> Ptr;
    static const char* GetChunkName() { return "ChunkWithBools"; }
    bool IsReplicaMigratable() override { return false; }

    MAKE_BOOLDATASET(1);
    MAKE_BOOLDATASET(2);
    MAKE_BOOLDATASET(3);
    MAKE_BOOLDATASET(4);
    MAKE_BOOLDATASET(5);
    MAKE_BOOLDATASET(6);
    MAKE_BOOLDATASET(7);
    MAKE_BOOLDATASET(8);
    MAKE_BOOLDATASET(9);
    MAKE_BOOLDATASET(10);
};

class ChunkWithShortInts
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(ChunkWithShortInts);

    ChunkWithShortInts()
    {
    }

    typedef AZStd::intrusive_ptr<ChunkWithShortInts> Ptr;
    static const char* GetChunkName() { return "ChunkWithShortInts"; }
    bool IsReplicaMigratable() override { return false; }

    MAKE_U8DATASET(1);
    MAKE_U8DATASET(2);
    MAKE_U8DATASET(3);
    MAKE_U8DATASET(4);
    MAKE_U8DATASET(5);
    MAKE_U8DATASET(6);
    MAKE_U8DATASET(7);
    MAKE_U8DATASET(8);
    MAKE_U8DATASET(9);
    MAKE_U8DATASET(10);
};

class SourcePeerChunk
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(SourcePeerChunk);

    SourcePeerChunk()
        : PeerRPC("PeerRPC")
        , ForwardPeerRPC("ForwardPeerRPC")
    { }

    typedef AZStd::intrusive_ptr<SourcePeerChunk> Ptr;
    static const char* GetChunkName() { return "SourcePeerChunk"; }

    bool IsReplicaMigratable() override { return false; }

    bool Peer(const RpcContext& context)
    {
        m_peers.push_back(context.m_sourcePeer);
        return true;
    }

    bool ForwardPeer(const RpcContext& context)
    {
        m_forwardPeers.push_back(context.m_sourcePeer);
        return true;
    }

    AZStd::vector<PeerId> m_peers;
    AZStd::vector<PeerId> m_forwardPeers;

    struct Traits
        : public RpcDefaultTraits
    {
        static const bool s_alwaysForwardSourcePeer = true;
    };

    Rpc<>::BindInterface<SourcePeerChunk, & SourcePeerChunk::Peer> PeerRPC;
    Rpc<>::BindInterface<SourcePeerChunk, & SourcePeerChunk::ForwardPeer, Traits> ForwardPeerRPC;
};

class CustomHandler
    : public ReplicaChunkInterface
{
public:
    GM_CLASS_ALLOCATOR(CustomHandler);

    void DataSetHandler(const int& val, const TimeContext&)
    {
        m_dataset.push_back(val);
    }

    bool RpcHandler(AZ::u32 t1, const RpcContext&)
    {
        m_rpc.push_back(t1);
        return true;
    }


    AZStd::vector<int> m_dataset;
    AZStd::vector<int> m_rpc;
};

class CustomHandlerChunk
    : public ReplicaChunkBase
{
public:
    GM_CLASS_ALLOCATOR(CustomHandlerChunk);

    CustomHandlerChunk()
        : Data("Data", 0)
        , RPC("RPC")
    {
    }

    typedef AZStd::intrusive_ptr<CustomHandlerChunk> Ptr;
    static const char* GetChunkName() { return "CustomHandlerChunk"; }

    bool IsReplicaMigratable() override { return false; }

    DataSet<int>::BindInterface<CustomHandler, & CustomHandler::DataSetHandler> Data;
    Rpc<RpcArg<AZ::u32> >::BindInterface<CustomHandler, & CustomHandler::RpcHandler> RPC;
};

class AllEventChunk
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(AllEventChunk);

    AllEventChunk()
        : m_attaches(0)
        , m_detaches(0)
        , m_activates(0)
        , m_deactivates(0)
    {
    }

    typedef AZStd::intrusive_ptr<AllEventChunk> Ptr;
    static const char* GetChunkName() { return "AllEventChunk"; }

    bool IsReplicaMigratable() override { return false; }

    void OnAttachedToReplica(Replica*) override
    {
        m_attaches++;
    }

    void OnDetachedFromReplica(Replica*) override
    {
        m_detaches++;
    }

    void OnReplicaActivate(const ReplicaContext&) override
    {
        m_activates++;
    }

    void OnReplicaDeactivate(const ReplicaContext&) override
    {
        m_deactivates++;
    }

    int m_attaches;
    int m_detaches;
    int m_activates;
    int m_deactivates;
};

class DrillerTestChunk
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(DrillerTestChunk);
    typedef AZStd::intrusive_ptr<DrillerTestChunk> Ptr;
    static const char* GetChunkName() { return "DrillerTestChunk"; }

    DrillerTestChunk() { }

    bool IsReplicaMigratable() override
    {
        return true;
    }
};

class NonConstMarshaler
{
public:
    typedef AZ::u32 DataType;
    typedef AZ::u32 SerializedType;

    static const AZStd::size_t MarshalSize = sizeof(SerializedType);

    NonConstMarshaler()
        : m_valueRead(0)
        , m_valueWritten(0) { }

    void Marshal(WriteBuffer& wb, const DataType& value)
    {
        wb.Write(value);
        m_valueWritten += value;
    }
    void Unmarshal(DataType& value, ReadBuffer& rb)
    {
        if (rb.Read(value))
        {
            m_valueRead += value;
        }
    }

    AZ::u32 m_valueRead;
    AZ::u32 m_valueWritten;
};

class NonConstMarshalerChunk
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(NonConstMarshalerChunk);
    typedef AZStd::intrusive_ptr<NonConstMarshalerChunk> Ptr;
    static const char* GetChunkName() { return "NonConstMarshalerChunk"; }

    NonConstMarshalerChunk()
        : m_data("data", 0)
        , RPCTestRPC("RPC")
    { }

    bool IsReplicaMigratable() override { return false; }

    bool RPCTest(AZ::u32, const RpcContext&)
    {
        return true;
    }

    DataSet<AZ::u32, NonConstMarshaler> m_data;
    IntrospectableRPC<RpcArg<AZ::u32, NonConstMarshaler> >::BindInterface<NonConstMarshalerChunk, & NonConstMarshalerChunk::RPCTest> RPCTestRPC;
};

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
class MPSession
    : public CarrierEventBus::Handler
{
public:

    ~MPSession()
    {
        CarrierEventBus::Handler::BusDisconnect();
    }

    ReplicaManager& GetReplicaMgr() { return m_rm; }
    void            SetTransport(Carrier* transport) { m_pTransport = transport; CarrierEventBus::Handler::BusConnect(transport->GetGridMate()); }
    Carrier*        GetTransport() { return m_pTransport; }
    void            SetClient(bool isClient) { m_client = isClient; }
    void            AcceptConn(bool accept) { m_acceptConn = accept; }

    void Update()
    {
        char buf[1500];
        for (ConnectionSet::iterator iConn = m_connections.begin(); iConn != m_connections.end(); ++iConn)
        {
            ConnectionID conn = *iConn;
            Carrier::ReceiveResult result = m_pTransport->Receive(buf, 1500, conn, GM_REPLICA_TEST_SESSION_CHANNEL);
            if (result.m_state == Carrier::ReceiveResult::RECEIVED)
            {
                if (strcmp(buf, "IM_A_CLIENT") == 0)
                {
                    m_rm.AddPeer(conn, Mode_Client);
                }
                else if (strcmp(buf, "IM_A_PEER") == 0)
                {
                    m_rm.AddPeer(conn, Mode_Peer);
                }
            }
        }
    }

    template<typename T>
    typename T::Ptr GetChunkFromReplica(ReplicaId id)
    {
        ReplicaPtr replica = GetReplicaMgr().FindReplica(id);
        if (!replica)
        {
            return nullptr;
        }
        return replica->FindReplicaChunk<T>();
    }

    //////////////////////////////////////////////////////////////////////////
    // CarrierEventBus
    void OnConnectionEstablished(Carrier* carrier, ConnectionID id) override
    {
        if (carrier != m_pTransport)
        {
            return; // not for us
        }
        m_connections.insert(id);
        if (m_client)
        {
            m_pTransport->Send("IM_A_CLIENT", 12, id, Carrier::SEND_RELIABLE, Carrier::PRIORITY_NORMAL, GM_REPLICA_TEST_SESSION_CHANNEL);
        }
        else
        {
            m_pTransport->Send("IM_A_PEER", 10, id, Carrier::SEND_RELIABLE, Carrier::PRIORITY_NORMAL, GM_REPLICA_TEST_SESSION_CHANNEL);
        }
    }

    void OnDisconnect(Carrier* carrier, ConnectionID id, CarrierDisconnectReason /*reason*/) override
    {
        if (carrier != m_pTransport)
        {
            return; // not for us
        }
        m_rm.RemovePeer(id);
        m_connections.erase(id);
    }

    void OnDriverError(Carrier* carrier, ConnectionID id, const DriverError& error) override
    {
        (void)error;
        if (carrier != m_pTransport)
        {
            return; // not for us
        }
        m_pTransport->Disconnect(id);
    }

    void OnSecurityError(Carrier* carrier, ConnectionID id, const SecurityError& error) override
    {
        (void)carrier;
        (void)id;
        (void)error;
        //Ignore security warnings in unit tests
    }
    //////////////////////////////////////////////////////////////////////////

    ReplicaManager              m_rm;
    Carrier*                    m_pTransport;
    typedef unordered_set<ConnectionID> ConnectionSet;
    ConnectionSet               m_connections;
    bool                        m_client;
    bool                        m_acceptConn;
};

const int k_delay = 100;

enum class TestStatus
{
    Running,
    Completed,
};

class Integ_SimpleTest
    : public UnitTest::GridMateMPTestFixture
    , public ::testing::Test
{
public:
    //GM_CLASS_ALLOCATOR(Integ_SimpleTest);

    Integ_SimpleTest()
        : m_sessionCount(0) { }

    virtual int GetNumSessions() { return 0; }
    virtual int GetHostSession() { return 0; }
    virtual void PreInit() { }
    virtual void PreConnect() { }
    virtual void PostInit() { }

    void SetUp() override
    {
        AZ_TracePrintf("GridMate", "\n");

        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(RPCChunk::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<RPCChunk>();
        }
        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(FullRPCChunk::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<FullRPCChunk>();
        }
        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(DataSetChunk::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<DataSetChunk>();
        }
        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(AllEventChunk::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<AllEventChunk>();
        }
        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(DrillerTestChunk::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<DrillerTestChunk>();
        }
        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(NonConstMarshalerChunk::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<NonConstMarshalerChunk>();
        }
        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(CustomHandlerChunk::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<CustomHandlerChunk>();
        }
        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(SourcePeerChunk::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<SourcePeerChunk>();
        }
        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(MixedTestChunk::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<MixedTestChunk>();
        }
        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(LargeChunkWithDefaultsMedium::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<LargeChunkWithDefaultsMedium>();
        }
        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(ChunkWithBools::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<ChunkWithBools>();
        }
        if (!ReplicaChunkDescriptorTable::Get().FindReplicaChunkDescriptor(ReplicaChunkClassId(ChunkWithShortInts::GetChunkName())))
        {
            ReplicaChunkDescriptorTable::Get().RegisterChunkType<ChunkWithShortInts>();
        }
    }

    void TearDown() override
    {
        for (int i = 0; i < m_sessionCount; ++i)
        {
            m_sessions[i].GetReplicaMgr().Shutdown();
            DefaultCarrier::Destroy(m_sessions[i].GetTransport());
        }
    }

    void RunTickLoop(AZStd::function<TestStatus(int)> tickBody)
    {
        // Setting up simulator with 50% outgoing packet loss
        m_defaultSimulator = AZStd::make_unique<DefaultSimulator>();
        m_defaultSimulator->SetOutgoingPacketLoss(0, 0);

        m_sessionCount = GetNumSessions();

        PreInit();

        // initialize transport
        int basePort = 4427;
        for (int i = 0; i < m_sessionCount; ++i)
        {
            TestCarrierDesc desc;
            desc.m_port = basePort + i;
            desc.m_enableDisconnectDetection = false;
            desc.m_simulator = m_defaultSimulator.get();

            // initialize replica managers
            m_sessions[i].SetTransport(DefaultCarrier::Create(desc, m_gridMate));
            m_sessions[i].AcceptConn(true);
            m_sessions[i].SetClient(false);
            m_sessions[i].GetReplicaMgr().Init(ReplicaMgrDesc(i + 1, m_sessions[i].GetTransport(), 0, i == GetHostSession() ? ReplicaMgrDesc::Role_SyncHost : 0));
            m_sessions[i].GetReplicaMgr().RegisterUserContext(12345, reinterpret_cast<void*>(static_cast<size_t>(i + 1)));
        }
        m_sessions[GetHostSession()].GetReplicaMgr().SetLocalLagAmt(1);

        PreConnect();

        for (int i = 1; i < m_sessionCount; ++i)
        {
            m_sessions[i].GetTransport()->Connect("127.0.0.1", basePort);
        }

        PostInit();

        // main test loop
        int count = 0;
        for (;; )
        {
            if (tickBody(count) == TestStatus::Completed)
            {
                break;
            }

            ++count;
            // tick everything
            for (int i = 0; i < m_sessionCount; ++i)
            {
                m_sessions[i].Update();
                m_sessions[i].GetReplicaMgr().Unmarshal();
            }
            for (int i = 0; i < m_sessionCount; ++i)
            {
                m_sessions[i].GetReplicaMgr().UpdateReplicas();
            }
            for (int i = 0; i < m_sessionCount; ++i)
            {
                m_sessions[i].GetReplicaMgr().UpdateFromReplicas();
                m_sessions[i].GetReplicaMgr().Marshal();
            }
            for (int i = 0; i < m_sessionCount; ++i)
            {
                m_sessions[i].GetTransport()->Update();
            }
            AZStd::this_thread::sleep_for(AZStd::chrono::milliseconds(k_delay));
        }
    }

    int m_sessionCount;
    AZStd::array<MPSession, 10> m_sessions;
    AZStd::unique_ptr<DefaultSimulator> m_defaultSimulator;
};

class Integ_ReplicaChunkRPCExec
    : public Integ_SimpleTest
{
public:
    Integ_ReplicaChunkRPCExec()
        : m_chunk(nullptr)
        , m_replicaId(0)
    { }

    enum
    {
        sHost,
        s2,
        s3,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        // put something on s1 to get it going
        auto replica = Replica::CreateReplica(nullptr);
        m_chunk = CreateAndAttachReplicaChunk<RPCChunk>(replica);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(replica);
    }

    RPCChunk::Ptr m_chunk;
    ReplicaId m_replicaId;
};

TEST_F(Integ_ReplicaChunkRPCExec, ReplicaChunkRPCExec)
{
    RunTickLoop([this](int tick) -> TestStatus
    {
        // perform some random actions on a timeline
        switch (tick)
        {
        case 10:
            m_chunk->FromMasterBroadcast();
            break;

        case 20:
            m_chunk->FromMasterNotBroadcast();
            break;

        case 30:
        {
            auto r = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            r->FindReplicaChunk<RPCChunk>()->FromProxyBroadcast();
            break;
        }

        case 40:
        {
            auto r = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            r->FindReplicaChunk<RPCChunk>()->FromProxyNotBroadcast();
            break;
        }

        case 50:
        {
            auto s1host = m_sessions[sHost].GetReplicaMgr().FindReplica(m_replicaId)->FindReplicaChunk<RPCChunk>();
            auto s2proxy = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId)->FindReplicaChunk<RPCChunk>();
            auto s3proxy = m_sessions[s3].GetReplicaMgr().FindReplica(m_replicaId)->FindReplicaChunk<RPCChunk>();

            AZ_TEST_ASSERT(s1host->m_fromMasterBroadcast == 1);
            AZ_TEST_ASSERT(s2proxy->m_fromMasterBroadcast == 1);
            AZ_TEST_ASSERT(s3proxy->m_fromMasterBroadcast == 1);

            AZ_TEST_ASSERT(s1host->m_fromMasterNotBroadcast == 1);
            AZ_TEST_ASSERT(s2proxy->m_fromMasterNotBroadcast == 0);
            AZ_TEST_ASSERT(s3proxy->m_fromMasterNotBroadcast == 0);

            AZ_TEST_ASSERT(s1host->m_fromProxyBroadcast == 1);
            AZ_TEST_ASSERT(s2proxy->m_fromProxyBroadcast == 1);
            AZ_TEST_ASSERT(s3proxy->m_fromProxyBroadcast == 1);

            AZ_TEST_ASSERT(s1host->m_fromProxyNotBroadcast == 1);
            AZ_TEST_ASSERT(s2proxy->m_fromProxyNotBroadcast == 0);
            AZ_TEST_ASSERT(s3proxy->m_fromProxyNotBroadcast == 0);

            return TestStatus::Completed;
        }
        default:
            return TestStatus::Running;
        }
        return TestStatus::Running;
    });
}

//-----------------------------------------------------------------------------
class DestroyRPCChunk
    : public ReplicaChunk
{
public:
    GM_CLASS_ALLOCATOR(DestroyRPCChunk);

    DestroyRPCChunk()
        : DestroyFromMaster("DestroyFromMaster")
        , DestroyFromProxy("DestroyFromProxy")
        , BeforeDestroyFromProxy("BeforeDestroyFromProxy")
        , AfterDestroyFromProxy("AfterDestroyFromProxy")
        , BeforeDestroyFromMaster("BeforeDestroyFromMaster")
        , AfterDestroyFromMaster("AfterDestroyFromMaster")
    {
    }

    typedef AZStd::intrusive_ptr<DestroyRPCChunk> Ptr;
    static const char* GetChunkName() { return "DestroyRPCChunk"; }

    bool IsReplicaMigratable() override { return false; }

    bool DestroyFromMasterFn(const RpcContext&)
    {
        AZ_TracePrintf("GridMate", "Executed DestroyFromMaster %d %s\n", GetReplicaId(), GetReplica()->IsMaster() ? "master" : "proxy");
        ++s_destroyFromMasterCalls;
        if (GetReplica()->IsMaster())
        {
            GetReplica()->Destroy();
        }
        return true;
    }

    bool DestroyFromProxyFn(const RpcContext&)
    {
        AZ_TracePrintf("GridMate", "Executed DestroyFromProxy %d %s\n", GetReplicaId(), GetReplica()->IsMaster() ? "master" : "proxy");
        ++s_destroyFromProxyCalls;
        if (GetReplica()->IsMaster())
        {
            GetReplica()->Destroy();
        }
        return true;
    }

    bool BeforeDestroyFromProxyFn(const RpcContext&)
    {
        AZ_TracePrintf("GridMate", "Executed BeforeDestroyFromProxy %d %s\n", GetReplicaId(), GetReplica()->IsMaster() ? "master" : "proxy");
        ++s_beforeDestroyFromProxyCalls;
        return true;
    }

    bool AfterDestroyFromProxyFn(const RpcContext&)
    {
        AZ_TracePrintf("GridMate", "Executed AfterDestroyFromProxy %d %s\n", GetReplicaId(), GetReplica()->IsMaster() ? "master" : "proxy");
        ++s_afterDestroyFromProxyCalls;
        return true;
    }

    bool BeforeDestroyFromMasterFn(const RpcContext&)
    {
        AZ_TracePrintf("GridMate", "Executed BeforeDestroyFromMaster %d %s\n", GetReplicaId(), GetReplica()->IsMaster() ? "master" : "proxy");
        ++s_beforeDestroyFromMasterCalls;
        return true;
    }

    bool AfterDestroyFromMasterFn(const RpcContext&)
    {
        AZ_TracePrintf("GridMate", "Executed AfterDestroyFromMaster %d %s\n", GetReplicaId(), GetReplica()->IsMaster() ? "master" : "proxy");
        ++s_afterDestroyFromMasterCalls;
        return true;
    }

    Rpc<>::BindInterface<DestroyRPCChunk, & DestroyRPCChunk::DestroyFromMasterFn> DestroyFromMaster;
    Rpc<>::BindInterface<DestroyRPCChunk, & DestroyRPCChunk::DestroyFromProxyFn> DestroyFromProxy;
    Rpc<>::BindInterface<DestroyRPCChunk, & DestroyRPCChunk::BeforeDestroyFromProxyFn> BeforeDestroyFromProxy;
    Rpc<>::BindInterface<DestroyRPCChunk, & DestroyRPCChunk::AfterDestroyFromProxyFn> AfterDestroyFromProxy;
    Rpc<>::BindInterface<DestroyRPCChunk, & DestroyRPCChunk::BeforeDestroyFromMasterFn> BeforeDestroyFromMaster;
    Rpc<>::BindInterface<DestroyRPCChunk, & DestroyRPCChunk::AfterDestroyFromMasterFn> AfterDestroyFromMaster;

    static int s_destroyFromMasterCalls;
    static int s_beforeDestroyFromMasterCalls;
    static int s_afterDestroyFromMasterCalls;
    static int s_destroyFromProxyCalls;
    static int s_beforeDestroyFromProxyCalls;
    static int s_afterDestroyFromProxyCalls;
};

int DestroyRPCChunk::s_destroyFromProxyCalls = 0;
int DestroyRPCChunk::s_destroyFromMasterCalls = 0;
int DestroyRPCChunk::s_beforeDestroyFromProxyCalls = 0;
int DestroyRPCChunk::s_afterDestroyFromProxyCalls = 0;
int DestroyRPCChunk::s_beforeDestroyFromMasterCalls = 0;
int DestroyRPCChunk::s_afterDestroyFromMasterCalls = 0;

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

class Integ_ReplicaDestroyedInRPC
    : public Integ_SimpleTest
{
public:
    enum
    {
        sHost,
        s2,
        s3,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        ReplicaChunkDescriptorTable::Get().RegisterChunkType<DestroyRPCChunk>();

        // creating 2 replicas on host
        for (int i = 0; i < 2; ++i)
        {
            auto replica = Replica::CreateReplica(nullptr);
            CreateAndAttachReplicaChunk<DestroyRPCChunk>(replica);
            m_repId[i] = m_sessions[sHost].GetReplicaMgr().AddMaster(replica);
        }
    }

    ReplicaId m_repId[2];
};

TEST_F(Integ_ReplicaDestroyedInRPC, ReplicaDestroyedInRPC)
{
    RunTickLoop([this](int tick)->TestStatus
    {
        switch (tick)
        {
        case 10:
        {
            // calling destroy on master
            auto master = m_sessions[sHost].GetReplicaMgr().FindReplica(m_repId[0]);
            auto masterChunk = master->FindReplicaChunk<DestroyRPCChunk>();
            masterChunk->BeforeDestroyFromMaster();
            masterChunk->DestroyFromMaster();
            masterChunk->AfterDestroyFromMaster();

            // calling destroy on proxy
            auto proxy = m_sessions[s2].GetReplicaMgr().FindReplica(m_repId[1]);
            auto proxyChunk = proxy->FindReplicaChunk<DestroyRPCChunk>();
            proxyChunk->BeforeDestroyFromProxy();
            proxyChunk->DestroyFromProxy();
            proxyChunk->AfterDestroyFromProxy();

            break;
        }

        case 20:
        {
            // checking if before destroy RPC was called on every peer
            AZ_TEST_ASSERT(DestroyRPCChunk::s_beforeDestroyFromProxyCalls == nSessions);
            AZ_TEST_ASSERT(DestroyRPCChunk::s_beforeDestroyFromMasterCalls == nSessions);

            // checking if destroy itself was called on every peer
            AZ_TEST_ASSERT(DestroyRPCChunk::s_destroyFromProxyCalls == nSessions);
            AZ_TEST_ASSERT(DestroyRPCChunk::s_destroyFromMasterCalls == nSessions);

            // checking if after destroy RPC was never called
            AZ_TEST_ASSERT(DestroyRPCChunk::s_afterDestroyFromProxyCalls == 0);     // RPCs that arrive via the network after deactivation should be dropped.
            AZ_TEST_ASSERT(DestroyRPCChunk::s_afterDestroyFromMasterCalls == 1);     // RPCs explicitly called on an inactive replica should still be executed.

            return TestStatus::Completed;
        }
        default:
            return TestStatus::Running;
        }

        return TestStatus::Running;
    });
}

class Integ_ReplicaChunkAddWhileReplicated
    : public Integ_SimpleTest
{
public:
    Integ_ReplicaChunkAddWhileReplicated()
        : m_replica(nullptr)
        , m_chunk(nullptr)
        , m_replicaId(0)
    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        // put something on s1 to get it going
        m_replica = Replica::CreateReplica(nullptr);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(m_replica);
    }

    ReplicaPtr m_replica;
    RPCChunk::Ptr m_chunk;
    ReplicaId m_replicaId;
};

TEST_F(Integ_ReplicaChunkAddWhileReplicated, ReplicaChunkAddWhileReplicated)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        // perform some random actions on a timeline
        switch (tick)
        {
        case 10:
        {
            auto repHost = m_sessions[sHost].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(!repHost->FindReplicaChunk<RPCChunk>());
            AZ_TEST_ASSERT(repHost->GetNumChunks() == 1);

            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(!rep->FindReplicaChunk<RPCChunk>());
            AZ_TEST_ASSERT(rep->GetNumChunks() == 1);
            break;
        }

        case 20:
            m_chunk = CreateAndAttachReplicaChunk<RPCChunk>(m_replica);
            break;

        case 40:
        {
            auto repHost = m_sessions[sHost].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(!repHost->FindReplicaChunk<RPCChunk>());
            AZ_TEST_ASSERT(repHost->GetNumChunks() == 1);

            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(!rep->FindReplicaChunk<RPCChunk>());
            AZ_TEST_ASSERT(rep->GetNumChunks() == 1);
            return TestStatus::Completed;
        }
        default:
            return TestStatus::Running;
        }
        return TestStatus::Running;
    });
}


class Integ_ReplicaRPCValues
    : public Integ_SimpleTest
{
public:
    Integ_ReplicaRPCValues()
        : m_replica(nullptr)
        , m_chunk(nullptr)
        , m_replicaId(0)
    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        // put something on s1 to get it going
        m_replica = Replica::CreateReplica(nullptr);
        m_chunk = CreateAndAttachReplicaChunk<RPCChunk>(m_replica);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(m_replica);
    }

    ReplicaPtr m_replica;
    RPCChunk::Ptr m_chunk;
    ReplicaId m_replicaId;
};

TEST_F(Integ_ReplicaRPCValues, ReplicaRPCValues)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        if (tick < 100)
        {
            if (!(tick % 10))
            {
                m_chunk->BroadcastInt(tick);
            }
            else if (!(tick % 20))
            {
                auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
                AZ_TEST_ASSERT(rep->FindReplicaChunk<RPCChunk>()->m_sentData.back() == (tick - 10));
            }
            return TestStatus::Running;
        }
        return TestStatus::Completed;
    });
}

class Integ_FullRPCValues
    : public Integ_SimpleTest
{
public:
    Integ_FullRPCValues()
        : m_replica(nullptr)
        , m_chunk(nullptr)
        , m_replicaId(0)
    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        // put something on s1 to get it going
        m_replica = Replica::CreateReplica(nullptr);
        m_chunk = CreateAndAttachReplicaChunk<FullRPCChunk>(m_replica);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(m_replica);
    }

    ReplicaPtr m_replica;
    FullRPCChunk::Ptr m_chunk;
    ReplicaId m_replicaId;
};

TEST_F(Integ_FullRPCValues, FullRPCValues)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        switch (tick)
        {
        case 10:
            m_chunk->ZeroRPC();
            break;
        case 20:
            m_chunk->OneRPC(11);
            break;
        case 30:
            m_chunk->TwoRPC(21, 22);
            break;
        case 40:
            m_chunk->ThreeRPC(31, 32, 33);
            break;
        case 50:
            m_chunk->FourRPC(41, 42, 43, 44);
            break;
        case 60:
            m_chunk->FiveRPC(51, 52, 53, 54, 55);
            break;
        case 70:
            m_chunk->SixRPC(61, 62, 63, 64, 65, 66);
            break;
        case 80:
            m_chunk->SevenRPC(71, 72, 73, 74, 75, 76, 77);
            break;
        case 90:
            m_chunk->EightRPC(81, 82, 83, 84, 85, 86, 87, 88);
            break;
        case 100:
            m_chunk->NineRPC(91, 92, 93, 94, 95, 96, 97, 98, 99);
            break;
        case 150:
        {
            auto client = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(client);
            auto clientChunk = client->FindReplicaChunk<FullRPCChunk>();
            AZ_TEST_ASSERT(clientChunk);

            AZ_TEST_ASSERT(m_chunk->m_sentData.size() == 10);
            AZ_TEST_ASSERT(clientChunk->m_receivedData.size() == 10);

            for (AZ::u32 i = 0; i <= 9; i++)
            {
                AZ_TEST_ASSERT(m_chunk->m_sentData[i].size() == static_cast<size_t>(i));
                for (AZ::u32 j = 0; j < i; j++)
                {
                    AZ_TEST_ASSERT(m_chunk->m_sentData[i][j] == (i * 10) + (j + 1));
                }
            }

            for (AZ::u32 i = 0; i <= 9; i++)
            {
                AZ_TEST_ASSERT(clientChunk->m_receivedData[i].size() == static_cast<size_t>(i));
                for (AZ::u32 j = 0; j < i; j++)
                {
                    AZ_TEST_ASSERT(clientChunk->m_receivedData[i][j] == (i * 10) + (j + 1));
                }
            }

            return TestStatus::Completed;
        }
        default:
            return TestStatus::Running;
        }
        return TestStatus::Running;
    });
}


class Integ_ReplicaRemoveProxy
    : public Integ_SimpleTest
{
public:
    Integ_ReplicaRemoveProxy()
        : m_replica(nullptr)
        , m_replicaId(0)
    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        // put something on s1 to get it going
        m_replica = Replica::CreateReplica(nullptr);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(m_replica);
    }


    ReplicaPtr m_replica;
    ReplicaId m_replicaId;
};

TEST_F(Integ_ReplicaRemoveProxy, ReplicaRemoveProxy)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        switch (tick)
        {
        case 10:
        {
            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(rep);
            break;
        }
        case 20:
            m_replica->Destroy();
            break;
        case 30:
        {
            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(!rep);
            return TestStatus::Completed;
        }
        default:
            return TestStatus::Running;
        }
        return TestStatus::Running;
    });
}


class Integ_ReplicaChunkEvents
    : public Integ_SimpleTest
{
public:
    Integ_ReplicaChunkEvents()
        : m_replicaId(InvalidReplicaId)
        , m_chunk(nullptr)
        , m_proxyChunk(nullptr)

    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        ReplicaPtr replica = Replica::CreateReplica(nullptr);
        m_chunk = CreateAndAttachReplicaChunk<AllEventChunk>(replica);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(replica);

        AZ_TEST_ASSERT(m_chunk->m_attaches == 1);
        AZ_TEST_ASSERT(m_chunk->m_activates == 1);
        AZ_TEST_ASSERT(m_chunk->m_detaches == 0);
        AZ_TEST_ASSERT(m_chunk->m_deactivates == 0);
    }


    ReplicaId m_replicaId;
    AllEventChunk::Ptr m_chunk;
    AllEventChunk::Ptr m_proxyChunk;
};

TEST_F(Integ_ReplicaChunkEvents, ReplicaChunkEvents)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        switch (tick)
        {
        case 20:
        {
            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(rep);
            m_proxyChunk = rep->FindReplicaChunk<AllEventChunk>();
            AZ_TEST_ASSERT(m_proxyChunk);
            AZ_TEST_ASSERT(m_proxyChunk->m_attaches == 1);
            AZ_TEST_ASSERT(m_proxyChunk->m_activates == 1);
            AZ_TEST_ASSERT(m_proxyChunk->m_detaches == 0);
            AZ_TEST_ASSERT(m_proxyChunk->m_deactivates == 0);
            break;
        }
        case 40:
            m_sessions[sHost].GetReplicaMgr().FindReplica(m_replicaId)->Destroy();
            break;
        case 60:
            AZ_TEST_ASSERT(m_chunk->m_attaches == 1);
            AZ_TEST_ASSERT(m_chunk->m_activates == 1);
            AZ_TEST_ASSERT(m_chunk->m_detaches == 1);
            AZ_TEST_ASSERT(m_chunk->m_deactivates == 1);
            AZ_TEST_ASSERT(m_proxyChunk->m_attaches == 1);
            AZ_TEST_ASSERT(m_proxyChunk->m_activates == 1);
            AZ_TEST_ASSERT(m_proxyChunk->m_detaches == 1);
            AZ_TEST_ASSERT(m_proxyChunk->m_deactivates == 1);
            return TestStatus::Completed;
        default: break;
        }
        return TestStatus::Running;
    });
}


class Integ_ReplicaChunksBeyond32
    : public Integ_SimpleTest
{
public:
    Integ_ReplicaChunksBeyond32()
        : m_replicaId(InvalidReplicaId)
    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        ReplicaPtr replica = Replica::CreateReplica(nullptr);
        for (auto i = 0; i < GM_MAX_CHUNKS_PER_REPLICA; ++i)
        {
            auto chunk = CreateAndAttachReplicaChunk<AllEventChunk>(replica);
            AZ_TEST_ASSERT(chunk);
        }
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(replica);

        auto numChunks = replica->GetNumChunks();
        AZ_TEST_ASSERT(numChunks == GM_MAX_CHUNKS_PER_REPLICA);
    }


    ReplicaId m_replicaId;
};

TEST_F(Integ_ReplicaChunksBeyond32, ReplicaChunksBeyond32)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        switch (tick)
        {
        case 20:
        {
            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(rep);

            auto numChunks = rep->GetNumChunks();
            AZ_TEST_ASSERT(numChunks == GM_MAX_CHUNKS_PER_REPLICA);
            break;
        }
        case 40:
            m_sessions[sHost].GetReplicaMgr().FindReplica(m_replicaId)->Destroy();
            break;
        case 60:
            return TestStatus::Completed;
        default:
            break;
        }
        return TestStatus::Running;
    });
}


class Integ_ReplicaChunkEventsDeactivate
    : public Integ_SimpleTest
{
public:
    Integ_ReplicaChunkEventsDeactivate()
        : m_replica(nullptr)
        , m_replicaId(0)
        , m_chunk(nullptr)
        , m_proxyChunk(nullptr)
    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        m_replica = Replica::CreateReplica(nullptr);
        m_chunk = CreateAndAttachReplicaChunk<AllEventChunk>(m_replica);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(m_replica);

        AZ_TEST_ASSERT(m_chunk->m_attaches == 1);
        AZ_TEST_ASSERT(m_chunk->m_activates == 1);
        AZ_TEST_ASSERT(m_chunk->m_detaches == 0);
        AZ_TEST_ASSERT(m_chunk->m_deactivates == 0);
    }

    ReplicaPtr m_replica;
    ReplicaId m_replicaId;
    AllEventChunk::Ptr m_chunk;
    AllEventChunk::Ptr m_proxyChunk;
};

TEST_F(Integ_ReplicaChunkEventsDeactivate, ReplicaChunkEventsDeactivate)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        switch (tick)
        {
        case 20:
        {
            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(rep);
            m_proxyChunk = rep->FindReplicaChunk<AllEventChunk>();
            AZ_TEST_ASSERT(m_proxyChunk);
            AZ_TEST_ASSERT(m_proxyChunk->m_attaches == 1);
            AZ_TEST_ASSERT(m_proxyChunk->m_activates == 1);
            AZ_TEST_ASSERT(m_proxyChunk->m_detaches == 0);
            AZ_TEST_ASSERT(m_proxyChunk->m_deactivates == 0);
            break;
        }
        case 40:
            m_replica->Destroy();
            AZ_TEST_ASSERT(m_chunk->m_attaches == 1);
            AZ_TEST_ASSERT(m_chunk->m_activates == 1);
            AZ_TEST_ASSERT(m_chunk->m_detaches == 0);
            AZ_TEST_ASSERT(m_chunk->m_deactivates == 1);
            break;
        case 50:
            m_replica = nullptr;
            AZ_TEST_ASSERT(m_chunk->m_attaches == 1);
            AZ_TEST_ASSERT(m_chunk->m_activates == 1);
            AZ_TEST_ASSERT(m_chunk->m_detaches == 1);
            AZ_TEST_ASSERT(m_chunk->m_deactivates == 1);
            break;
        case 60:
            AZ_TEST_ASSERT(m_proxyChunk->m_attaches == 1);
            AZ_TEST_ASSERT(m_proxyChunk->m_activates == 1);
            AZ_TEST_ASSERT(m_proxyChunk->m_detaches == 1);
            AZ_TEST_ASSERT(m_proxyChunk->m_deactivates == 1);
            return TestStatus::Completed;
        default: break;
        }
        return TestStatus::Running;
    });
}


class Integ_ReplicaDriller
    : public Integ_SimpleTest
{
public:
    Integ_ReplicaDriller()
        : m_replicaId(InvalidReplicaId)
    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    class ReplicaDrillerHook
        : public Debug::ReplicaDrillerBus::Handler
    {
    public:
        ReplicaDrillerHook()
            : m_createdReplicas(0)
            , m_destroyedReplicas(0)
            , m_activatedReplicas(0)
            , m_deactivatedReplicas(0)
            , m_attachedChunks(0)
            , m_detachedChunks(0)
            , m_numReplicaBytesSent(0)
            , m_numReplicaBytesReceived(0)
            , m_numRequestChangeOwnership(0)
            , m_numChangedOwnership(0)
            , m_createdChunks(0)
            , m_destroyedChunks(0)
            , m_activatedChunks(0)
            , m_deactivatedChunks(0)
            , m_numChunkBytesSent(0)
            , m_numChunkBytesReceived(0)
            , m_numOutgoingDatasets(0)
            , m_numIncomingDatasets(0)
            , m_numRpcRequests(0)
            , m_numRpcInvokes(0)
            , m_outgoingRpcDataSize(0)
            , m_incomingRpcDataSize(0)
            , m_totalOutgoingBytes(0)
            , m_totalIncomingBytes(0)
            , m_curReplicaSend(nullptr)
            , m_curReplicaChunkSend(nullptr)
            , m_curReplicaChunkIndexSend(GM_MAX_CHUNKS_PER_REPLICA)
            , m_curReplicaReceive(nullptr)
            , m_curReplicaChunkReceive(nullptr)
            , m_curReplicaChunkIndexReceive(GM_MAX_CHUNKS_PER_REPLICA)
        {
        }

        AZStd::size_t m_createdReplicas;
        AZStd::size_t m_destroyedReplicas;
        AZStd::size_t m_activatedReplicas;
        AZStd::size_t m_deactivatedReplicas;
        AZStd::size_t m_attachedChunks;
        AZStd::size_t m_detachedChunks;
        AZStd::size_t m_numReplicaBytesSent;
        AZStd::size_t m_numReplicaBytesReceived;
        AZStd::size_t m_numRequestChangeOwnership;
        AZStd::size_t m_numChangedOwnership;

        AZStd::size_t m_createdChunks;
        AZStd::size_t m_destroyedChunks;
        AZStd::size_t m_activatedChunks;
        AZStd::size_t m_deactivatedChunks;
        AZStd::size_t m_numChunkBytesSent;
        AZStd::size_t m_numChunkBytesReceived;

        AZStd::size_t m_numOutgoingDatasets;
        AZStd::size_t m_numIncomingDatasets;

        AZStd::size_t m_numRpcRequests;
        AZStd::size_t m_numRpcInvokes;
        AZStd::size_t m_outgoingRpcDataSize;
        AZStd::size_t m_incomingRpcDataSize;

        AZStd::size_t m_totalOutgoingBytes;
        AZStd::size_t m_totalIncomingBytes;

        Replica* m_curReplicaSend;
        ReplicaChunkBase* m_curReplicaChunkSend;
        size_t m_curReplicaChunkIndexSend;
        Replica* m_curReplicaReceive;
        ReplicaChunkBase* m_curReplicaChunkReceive;
        AZ::u32 m_curReplicaChunkIndexReceive;

        void OnCreateReplica(Replica* replica) override
        {
            AZ_TEST_ASSERT(replica);
            ++m_createdReplicas;
        }

        void OnDestroyReplica(Replica* replica) override
        {
            AZ_TEST_ASSERT(replica);
            ++m_destroyedReplicas;
        }

        void OnActivateReplica(Replica* replica) override
        {
            AZ_TEST_ASSERT(replica);
            ++m_activatedReplicas;
        }

        void OnDeactivateReplica(Replica* replica) override
        {
            AZ_TEST_ASSERT(replica);
            ++m_deactivatedReplicas;
        }

        void OnAttachReplicaChunk(ReplicaChunkBase* chunk) override
        {
            AZ_TEST_ASSERT(chunk);
            ++m_attachedChunks;
        }

        void OnDetachReplicaChunk(ReplicaChunkBase* chunk) override
        {
            AZ_TEST_ASSERT(chunk);
            ++m_detachedChunks;
        }

        void OnSendReplicaBegin(Replica* replica) override
        {
            AZ_TEST_ASSERT(replica);
            AZ_TEST_ASSERT(m_curReplicaSend == nullptr);
            m_curReplicaSend = replica;
        }

        void OnSendReplicaEnd(Replica* replica, const void* data, size_t len) override
        {
            AZ_TEST_ASSERT(replica);
            AZ_TEST_ASSERT(replica == m_curReplicaSend);
            AZ_TEST_ASSERT(data);
            AZ_TEST_ASSERT(len > 0);
            m_numReplicaBytesSent += len;
            m_curReplicaSend = nullptr;
        }

        void OnReceiveReplicaBegin(Replica* replica, const void* data, size_t len) override
        {
            AZ_TEST_ASSERT(replica);
            AZ_TEST_ASSERT(m_curReplicaReceive == nullptr);
            AZ_TEST_ASSERT(data);
            AZ_TEST_ASSERT(len > 0);
            m_curReplicaReceive = replica;
            m_numReplicaBytesReceived += len;
        }

        void OnReceiveReplicaEnd(Replica* replica) override
        {
            AZ_TEST_ASSERT(replica);
            AZ_TEST_ASSERT(replica == m_curReplicaReceive);
            m_curReplicaReceive = nullptr;
        }

        void OnRequestReplicaChangeOwnership(Replica* replica, PeerId requestor) override
        {
            AZ_TEST_ASSERT(replica);
            AZ_TEST_ASSERT(requestor == (s2 + 1));
            ++m_numRequestChangeOwnership;
        }

        void OnReplicaChangeOwnership(Replica* replica, bool wasMaster) override
        {
            AZ_TEST_ASSERT(replica);
            switch (m_numChangedOwnership)
            {
            case 0: // host loses ownership
                AZ_TEST_ASSERT(replica->IsProxy() && wasMaster == true);
                break;
            case 1: // peer acquires ownership
                AZ_TEST_ASSERT(replica->IsMaster() && wasMaster == false);
                break;
            default:
                AZ_TEST_ASSERT(0);
            }

            ++m_numChangedOwnership;
        }

        void OnCreateReplicaChunk(ReplicaChunkBase* chunk) override
        {
            AZ_TEST_ASSERT(chunk);
            ++m_createdChunks;
        }

        void OnDestroyReplicaChunk(ReplicaChunkBase* chunk) override
        {
            AZ_TEST_ASSERT(chunk);
            ++m_destroyedChunks;
        }

        void OnActivateReplicaChunk(ReplicaChunkBase* chunk) override
        {
            AZ_TEST_ASSERT(chunk);
            ++m_activatedChunks;
        }

        void OnDeactivateReplicaChunk(ReplicaChunkBase* chunk) override
        {
            AZ_TEST_ASSERT(chunk);
            ++m_deactivatedChunks;
        }

        void OnSendReplicaChunkBegin(ReplicaChunkBase* chunk, AZ::u32 chunkIndex, PeerId from, PeerId to) override
        {
            (void)from;
            (void)to;

            AZ_TEST_ASSERT(chunk);
            AZ_TEST_ASSERT(m_curReplicaSend == chunk->GetReplica());
            AZ_TEST_ASSERT(m_curReplicaChunkSend == nullptr);
            AZ_TEST_ASSERT(m_curReplicaChunkIndexSend == GM_MAX_CHUNKS_PER_REPLICA);
            m_curReplicaChunkSend = chunk;
            m_curReplicaChunkIndexSend = chunkIndex;
        }

        void OnSendReplicaChunkEnd(ReplicaChunkBase* chunk, AZ::u32 chunkIndex, const void* data, size_t len) override
        {
            AZ_TEST_ASSERT(chunk);
            AZ_TEST_ASSERT(m_curReplicaSend == chunk->GetReplica());
            AZ_TEST_ASSERT(m_curReplicaChunkSend == chunk);
            AZ_TEST_ASSERT(m_curReplicaChunkIndexSend == chunkIndex);
            AZ_TEST_ASSERT(data);
            AZ_TEST_ASSERT(len > 0);
            m_numChunkBytesSent += len;
            m_curReplicaChunkSend = nullptr;
            m_curReplicaChunkIndexSend = GM_MAX_CHUNKS_PER_REPLICA;

        }

        void OnReceiveReplicaChunkBegin(ReplicaChunkBase* chunk, AZ::u32 chunkIndex, PeerId from, PeerId to, const void* data, size_t len) override
        {
            AZ_TEST_ASSERT(chunk);
            AZ_TEST_ASSERT(m_curReplicaReceive == chunk->GetReplica());
            AZ_TEST_ASSERT(m_curReplicaChunkReceive == nullptr);
            AZ_TEST_ASSERT(m_curReplicaChunkIndexReceive == GM_MAX_CHUNKS_PER_REPLICA);
            AZ_TEST_ASSERT(from);
            AZ_TEST_ASSERT(to);
            AZ_TEST_ASSERT(data);
            AZ_TEST_ASSERT(len > 0);
            m_curReplicaChunkReceive = chunk;
            m_curReplicaChunkIndexReceive = chunkIndex;
            m_numChunkBytesReceived += len;
        }

        void OnReceiveReplicaChunkEnd(ReplicaChunkBase* chunk, AZ::u32 chunkIndex) override
        {
            AZ_TEST_ASSERT(chunk);
            AZ_TEST_ASSERT(m_curReplicaReceive == chunk->GetReplica());
            AZ_TEST_ASSERT(m_curReplicaChunkReceive == chunk);
            AZ_TEST_ASSERT(m_curReplicaChunkIndexReceive == chunkIndex);
            m_curReplicaChunkReceive = nullptr;
            m_curReplicaChunkIndexReceive = GM_MAX_CHUNKS_PER_REPLICA;
        }

        void OnSendDataSet(ReplicaChunkBase* chunk, AZ::u32 chunkIndex, DataSetBase* dataSet, PeerId from, PeerId to, const void* data, size_t len) override
        {
            AZ_TEST_ASSERT(chunk);
            AZ_TEST_ASSERT(m_curReplicaChunkSend == chunk);
            AZ_TEST_ASSERT(m_curReplicaChunkIndexSend == chunkIndex);
            AZ_TEST_ASSERT(dataSet);
            AZ_TEST_ASSERT(from);
            AZ_TEST_ASSERT(to);
            AZ_TEST_ASSERT(data);
            AZ_TEST_ASSERT(len > 0);

            ++m_numOutgoingDatasets;
        }

        void OnReceiveDataSet(ReplicaChunkBase* chunk, AZ::u32 chunkIndex, DataSetBase* dataSet, PeerId from, PeerId to, const void* data, size_t len) override
        {
            AZ_TEST_ASSERT(chunk);
            AZ_TEST_ASSERT(m_curReplicaChunkReceive == chunk);
            AZ_TEST_ASSERT(m_curReplicaChunkIndexReceive == chunkIndex);
            AZ_TEST_ASSERT(dataSet);
            AZ_TEST_ASSERT(from);
            AZ_TEST_ASSERT(to);
            AZ_TEST_ASSERT(data);
            AZ_TEST_ASSERT(len > 0);

            ++m_numIncomingDatasets;
        }

        void OnRequestRpc(ReplicaChunkBase* chunk, Internal::RpcRequest* rpc) override
        {
            (void)chunk;
            (void)rpc;
            ++m_numRpcRequests;
        }

        void OnInvokeRpc(ReplicaChunkBase* chunk, Internal::RpcRequest* rpc) override
        {
            (void)chunk;
            (void)rpc;
            ++m_numRpcInvokes;
        }

        void OnSendRpc(ReplicaChunkBase* chunk, AZ::u32 chunkIndex, Internal::RpcRequest* rpc, PeerId from, PeerId to, const void* data, size_t len) override
        {
            AZ_TEST_ASSERT(chunk);
            AZ_TEST_ASSERT(m_curReplicaChunkSend == chunk);
            AZ_TEST_ASSERT(m_curReplicaChunkIndexSend == chunkIndex);
            AZ_TEST_ASSERT(rpc);
            AZ_TEST_ASSERT(from);
            AZ_TEST_ASSERT(to);
            AZ_TEST_ASSERT(data);
            AZ_TEST_ASSERT(len > 0);
            m_outgoingRpcDataSize += len;
        }

        void OnReceiveRpc(ReplicaChunkBase* chunk, AZ::u32 chunkIndex, Internal::RpcRequest* rpc, PeerId from, PeerId to, const void* data, size_t len) override
        {
            AZ_TEST_ASSERT(chunk);
            AZ_TEST_ASSERT(m_curReplicaChunkReceive == chunk);
            AZ_TEST_ASSERT(m_curReplicaChunkIndexReceive == chunkIndex);
            AZ_TEST_ASSERT(rpc);
            AZ_TEST_ASSERT(from);
            AZ_TEST_ASSERT(to);
            AZ_TEST_ASSERT(data);
            AZ_TEST_ASSERT(len > 0);
            m_incomingRpcDataSize += len;
        }

        void OnSend(PeerId to, const void* data, size_t len, bool isReliable) override
        {
            (void)to; // peerId might not be valid at this point, e.g. handshake (Cmd_Greetings) did not accomplish yet
            (void)isReliable;
            AZ_TEST_ASSERT(data);
            AZ_TEST_ASSERT(len > 0);
            m_totalOutgoingBytes += len;
        }

        void OnReceive(PeerId from, const void* data, size_t len) override
        {
            (void)from; // peerId might not be valid at this point, e.g. handshake (Cmd_Greetings) did not accomplish yet
            AZ_TEST_ASSERT(data);
            AZ_TEST_ASSERT(len > 0);
            m_totalIncomingBytes += len;
        }
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        m_driller.BusConnect();

        ReplicaPtr replica = Replica::CreateReplica(nullptr);
        CreateAndAttachReplicaChunk<DrillerTestChunk>(replica);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(replica);
    }

    ~Integ_ReplicaDriller()
    {
        m_driller.BusDisconnect();
    }

    ReplicaDrillerHook m_driller;
    ReplicaId m_replicaId;
};

TEST_F(Integ_ReplicaDriller, ReplicaDriller)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        switch (tick)
        {
        case 10:
        {
            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(rep);
            AZ_TEST_ASSERT(rep->IsProxy());
            rep->RequestChangeOwnership();
            break;
        }
        case 30:
        {
            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(rep);
            AZ_TEST_ASSERT(rep->IsMaster());
            rep->Destroy();
            break;
        }
        case 40:
            // replicas
            AZ_TEST_ASSERT(m_driller.m_createdReplicas > 0);
            AZ_TEST_ASSERT(m_driller.m_destroyedReplicas > 0);
            AZ_TEST_ASSERT(m_driller.m_activatedReplicas > 0);
            AZ_TEST_ASSERT(m_driller.m_deactivatedReplicas > 0);
            AZ_TEST_ASSERT(m_driller.m_numReplicaBytesSent > 0);
            AZ_TEST_ASSERT(m_driller.m_numReplicaBytesReceived > 0);
            AZ_TEST_ASSERT(m_driller.m_numRequestChangeOwnership == 1);
            AZ_TEST_ASSERT(m_driller.m_numChangedOwnership == 2); // two because one call for host & one for peer

                                                                  // chunks
            AZ_TEST_ASSERT(m_driller.m_createdChunks >= m_driller.m_createdReplicas);
            AZ_TEST_ASSERT(m_driller.m_destroyedChunks >= m_driller.m_destroyedReplicas);
            AZ_TEST_ASSERT(m_driller.m_activatedChunks >= m_driller.m_activatedReplicas);
            AZ_TEST_ASSERT(m_driller.m_deactivatedChunks >= m_driller.m_deactivatedReplicas);
            AZ_TEST_ASSERT(m_driller.m_attachedChunks > 0);
            AZ_TEST_ASSERT(m_driller.m_detachedChunks > 0);
            AZ_TEST_ASSERT(m_driller.m_numChunkBytesReceived > 0);

            AZ_TEST_ASSERT(m_driller.m_numChunkBytesSent > 0);
            AZ_TEST_ASSERT(m_driller.m_numChunkBytesReceived > 0);

            // datasets
            AZ_TEST_ASSERT(m_driller.m_numOutgoingDatasets > 0);
            AZ_TEST_ASSERT(m_driller.m_numIncomingDatasets > 0);

            // rpcs
            AZ_TEST_ASSERT(m_driller.m_numRpcRequests > 0);
            AZ_TEST_ASSERT(m_driller.m_numRpcInvokes > 0);
            AZ_TEST_ASSERT(m_driller.m_outgoingRpcDataSize > 0);
            AZ_TEST_ASSERT(m_driller.m_incomingRpcDataSize > 0);

            // data
            AZ_TEST_ASSERT(m_driller.m_totalOutgoingBytes > 0);
            AZ_TEST_ASSERT(m_driller.m_totalIncomingBytes > 0);
            return TestStatus::Completed;
        default: break;
        }
        return TestStatus::Running;
    });
}


class Integ_DataSetChangedTest
    : public Integ_SimpleTest
{
public:
    Integ_DataSetChangedTest()
        : m_replica(nullptr)
        , m_replicaId(0)
        , m_chunk(nullptr)
    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        m_replica = Replica::CreateReplica(nullptr);
        m_chunk = CreateAndAttachReplicaChunk<DataSetChunk>(m_replica);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(m_replica);
    }


    ReplicaPtr m_replica;
    ReplicaId m_replicaId;
    DataSetChunk::Ptr m_chunk;
};

TEST_F(Integ_DataSetChangedTest, DataSetChangedTest)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        if (tick < 100)
        {
            if (!(tick % 10))
            {
                m_chunk->Data1.Set(tick);
            }
            return TestStatus::Running;
        }

        auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
        auto chunk = rep->FindReplicaChunk<DataSetChunk>();
        AZ_TEST_ASSERT(m_chunk->m_changedData.size() == 0);
        AZ_TEST_ASSERT(chunk->m_changedData.size() == 10);
        int expected = 0;
        for (auto i : chunk->m_changedData)
        {
            AZ_TEST_ASSERT(i == expected);
            expected += 10;
        }

        return TestStatus::Completed;
    });
}


class Integ_CustomHandlerTest
    : public Integ_SimpleTest
{
public:
    Integ_CustomHandlerTest()
        : m_replica(nullptr)
        , m_replicaId(0)
        , m_chunk(nullptr)
    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        m_replica = Replica::CreateReplica(nullptr);
        m_chunk = CreateAndAttachReplicaChunk<CustomHandlerChunk>(m_replica);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(m_replica);
        m_masterHandler.reset(aznew CustomHandler());
        m_proxyHandler.reset(aznew CustomHandler());
        m_chunk->SetHandler(m_masterHandler.get());
    }

    ReplicaPtr m_replica;
    ReplicaId m_replicaId;
    CustomHandlerChunk::Ptr m_chunk;
    AZStd::scoped_ptr<CustomHandler> m_masterHandler;
    AZStd::scoped_ptr<CustomHandler> m_proxyHandler;
};

TEST_F(Integ_CustomHandlerTest, CustomHandlerTest)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        {
            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            if (rep)
            {
                auto chunk = rep->FindReplicaChunk<CustomHandlerChunk>();
                if (chunk && !chunk->GetHandler())
                {
                    chunk->SetHandler(m_proxyHandler.get());
                    m_proxyHandler->m_dataset.push_back(chunk->Data.Get());
                }
            }
        }

        if (tick < 100)
        {
            if (!(tick % 10))
            {
                m_chunk->Data.Set(tick);
            }
            return TestStatus::Running;
        }
        else if (tick < 200)
        {
            if (!(tick % 10))
            {
                m_chunk->RPC(tick);
            }
            return TestStatus::Running;
        }


        AZ_TEST_ASSERT(m_proxyHandler->m_dataset.size() == 10);
        int expected = 0;
        for (auto i : m_proxyHandler->m_dataset)
        {
            AZ_TEST_ASSERT(i == expected);
            expected += 10;
        }
        expected = 100;
        for (auto i : m_proxyHandler->m_rpc)
        {
            AZ_TEST_ASSERT(i == expected);
            expected += 10;
        }
        return TestStatus::Completed;
    });
}


class Integ_NonConstMarshalerTest
    : public Integ_SimpleTest
{
public:
    Integ_NonConstMarshalerTest()
        : m_replica(nullptr)
        , m_replicaId(0)
        , m_chunk(nullptr)
    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        m_replica = Replica::CreateReplica(nullptr);
        m_chunk = CreateAndAttachReplicaChunk<NonConstMarshalerChunk>(m_replica);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(m_replica);
    }

    ReplicaPtr m_replica;
    ReplicaId m_replicaId;
    NonConstMarshalerChunk::Ptr m_chunk;
};

TEST_F(Integ_NonConstMarshalerTest, NonConstMarshalerTest)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        switch (tick)
        {
        case 10:
            m_chunk->RPCTestRPC(1);
            break;

        case 20:
            m_chunk->RPCTestRPC(2);
            break;

        case 30:
            m_chunk->m_data.Set(10);
            break;

        case 40:
            m_chunk->m_data.Set(20);
            break;

        case 50:
        {
            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            auto chunk = rep->FindReplicaChunk<NonConstMarshalerChunk>();

            AZ_TEST_ASSERT(m_chunk->RPCTestRPC.GetMarshalers().m_marshaler.m_valueWritten > 0);
            AZ_TEST_ASSERT(chunk->RPCTestRPC.GetMarshalers().m_marshaler.m_valueRead > 0);

            AZ_TEST_ASSERT(m_chunk->m_data.GetMarshaler().m_valueWritten > 0);
            AZ_TEST_ASSERT(chunk->m_data.GetMarshaler().m_valueRead > 0);

            return TestStatus::Completed;
        }
        default: ;
        }

        return TestStatus::Running;
    });
}


class Integ_SourcePeerTest
    : public Integ_SimpleTest
{
public:
    Integ_SourcePeerTest()
        : m_replica(nullptr)
        , m_replicaId(0)
        , m_chunk(nullptr)
        , m_chunk2(nullptr)
    {
    }

    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        m_replica = Replica::CreateReplica(nullptr);
        m_chunk = CreateAndAttachReplicaChunk<SourcePeerChunk>(m_replica);
        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(m_replica);
    }

    ReplicaPtr m_replica;
    ReplicaId m_replicaId;
    SourcePeerChunk::Ptr m_chunk;
    SourcePeerChunk::Ptr m_chunk2;
};

TEST_F(Integ_SourcePeerTest, SourcePeerTest)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        if (!m_chunk2)
        {
            auto rep = m_sessions[s2].GetReplicaMgr().FindReplica(m_replicaId);
            if (rep)
            {
                m_chunk2 = rep->FindReplicaChunk<SourcePeerChunk>();
            }
        }

        switch (tick)
        {
        case 10:
            m_chunk->PeerRPC();
            break;

        case 20:
            m_chunk2->PeerRPC();
            break;

        case 30:
            m_chunk->ForwardPeerRPC();
            break;

        case 40:
            m_chunk2->ForwardPeerRPC();
            break;

        case 50:
        {
            AZ_TEST_ASSERT(m_chunk->m_peers.size() == 2);
            AZ_TEST_ASSERT(m_chunk2->m_peers.size() == 2);

            AZ_TEST_ASSERT(m_chunk->m_peers[0] == m_sessions[sHost].GetReplicaMgr().GetLocalPeerId());
            AZ_TEST_ASSERT(m_chunk2->m_peers[0] == m_sessions[sHost].GetReplicaMgr().GetLocalPeerId());

            AZ_TEST_ASSERT(m_chunk->m_peers[1] == m_sessions[s2].GetReplicaMgr().GetLocalPeerId());
            AZ_TEST_ASSERT(m_chunk2->m_peers[1] == m_sessions[sHost].GetReplicaMgr().GetLocalPeerId());

            AZ_TEST_ASSERT(m_chunk->m_forwardPeers.size() == 2);
            AZ_TEST_ASSERT(m_chunk2->m_forwardPeers.size() == 2);

            AZ_TEST_ASSERT(m_chunk->m_forwardPeers[0] == m_sessions[sHost].GetReplicaMgr().GetLocalPeerId());
            AZ_TEST_ASSERT(m_chunk2->m_forwardPeers[0] == m_sessions[sHost].GetReplicaMgr().GetLocalPeerId());

            AZ_TEST_ASSERT(m_chunk->m_forwardPeers[1] == m_sessions[s2].GetReplicaMgr().GetLocalPeerId());
            AZ_TEST_ASSERT(m_chunk2->m_forwardPeers[1] == m_sessions[s2].GetReplicaMgr().GetLocalPeerId());

            return TestStatus::Completed;
        }
        default: break;
        }

        return TestStatus::Running;
    });
}


class Integ_SendWithPriority
    : public Integ_SimpleTest
{
public:
    enum
    {
        sHost,
        s2,
        nSessions
    };

    class PriorityChunk
        : public ReplicaChunk
    {
    public:
        GM_CLASS_ALLOCATOR(PriorityChunk);
        typedef AZStd::intrusive_ptr<PriorityChunk> Ptr;
        static const char* GetChunkName() { return "PriorityChunk"; }

        PriorityChunk()
            : m_value("Value")
        {
        }

        bool IsReplicaMigratable() override { return false; }

        DataSet<int> m_value;
    };

    class ReplicaDrillerHook
        : public Debug::ReplicaDrillerBus::Handler
    {
    public:
        ReplicaDrillerHook()
            : m_expectedSendValue(Integ_SendWithPriority::kNumReplicas)
            , m_expectedRecvValue(Integ_SendWithPriority::kNumReplicas)
        {
        }

        void OnReceiveReplicaEnd(Replica* replica) override
        {
            auto chunk = replica->FindReplicaChunk<PriorityChunk>();
            if (chunk && m_expectedRecvValue > 0)
            {
                AZ_TEST_ASSERT(chunk->m_value.Get() == m_expectedRecvValue); // checking reverse order
                --m_expectedRecvValue;
            }
        }

        void OnSendReplicaEnd(Replica* replica, const void* data, size_t len) override
        {
            (void)data;
            (void)len;

            auto chunk = replica->FindReplicaChunk<PriorityChunk>();
            if (chunk && m_expectedSendValue > 0)
            {
                AZ_TEST_ASSERT(chunk->m_value.Get() == m_expectedSendValue); // checking reverse order
                --m_expectedSendValue;
            }
        }

        int m_expectedSendValue;
        int m_expectedRecvValue;
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        ReplicaChunkDescriptorTable::Get().RegisterChunkType<PriorityChunk>();

        m_driller.BusConnect();

        for (unsigned i = 0; i < kNumReplicas; ++i)
        {
            ReplicaPtr replica = Replica::CreateReplica(nullptr);
            m_chunks[i] = CreateAndAttachReplicaChunk<PriorityChunk>(replica);
            m_chunks[i]->m_value.Set(i + 1); // setting dataset values to 1..kNumReplicas
            m_chunks[i]->SetPriority(k_replicaPriorityNormal + static_cast<ReplicaPriority>(i)); // the later created - the higher priorities, so should be sent in reverse order
            m_sessions[sHost].GetReplicaMgr().AddMaster(replica);
        }
    }


    static const size_t kNumReplicas = 5;

    ReplicaDrillerHook m_driller;
    PriorityChunk::Ptr m_chunks[kNumReplicas];
};

TEST_F(Integ_SendWithPriority, SendWithPriority)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        if (tick == 20)
        {
            AZ_TEST_ASSERT(m_driller.m_expectedSendValue == 0); // sent all the replicas in the right order
            AZ_TEST_ASSERT(m_driller.m_expectedRecvValue == 0); // received all the replicas in the right order
            return TestStatus::Completed;
        }

        return TestStatus::Running;
    });
}


class Integ_SuspendUpdatesTest
    : public Integ_SimpleTest
{
public:
    enum
    {
        sHost,
        s2,
        s3,
        nSessions
    };

    class SuspendUpdatesChunk
        : public ReplicaChunk
    {
    public:
        GM_CLASS_ALLOCATOR(SuspendUpdatesChunk);

        SuspendUpdatesChunk()
            : RPCTest("RPCTest")
            , Data1("Data1", 0)
            , m_numDataSetChanged(0)
            , m_numRpcCalled(0)
            , m_enabled(true)
        {
        }

        void OnReplicaActivate(const ReplicaContext& ctx) override
        {
            (void)ctx;
            SuspendUpdatesFromReplica();
        }

        typedef AZStd::intrusive_ptr<SuspendUpdatesChunk> Ptr;
        static const char* GetChunkName() { return "SuspendUpdatesChunk"; }

        bool IsReplicaMigratable() override { return false; }
        bool IsUpdateFromReplicaEnabled() override { return m_enabled; }

        void SuspendUpdatesFromReplica()
        {
            m_enabled = false;
        }

        void ResumeUpdatesFromReplica()
        {
            m_enabled = true;
        }

        void DatasetHandler(const int& val, const TimeContext& ctx)
        {
            (void)val;
            (void)ctx;
            ++m_numDataSetChanged;
        }

        bool RPCHandler(const RpcContext& ctx)
        {
            (void)ctx;
            ++m_numRpcCalled;
            return true;
        }

        Rpc<>::BindInterface<SuspendUpdatesChunk, &SuspendUpdatesChunk::RPCHandler> RPCTest;
        DataSet<int>::BindInterface<SuspendUpdatesChunk, &SuspendUpdatesChunk::DatasetHandler> Data1;

        unsigned int m_numDataSetChanged;
        unsigned int m_numRpcCalled;
        bool m_enabled;
    };

    int GetNumSessions() override { return nSessions; }

    void PreConnect() override
    {
        ReplicaChunkDescriptorTable::Get().RegisterChunkType<SuspendUpdatesChunk>();

        ReplicaPtr replica = Replica::CreateReplica(nullptr);
        m_chunk = CreateAndAttachReplicaChunk<SuspendUpdatesChunk>(replica);
        m_sessions[sHost].GetReplicaMgr().AddMaster(replica);
    }

    SuspendUpdatesChunk::Ptr m_chunk = nullptr;
    unsigned int m_numRpcCalled = 0;
};

TEST_F(Integ_SuspendUpdatesTest, SuspendUpdatesTest)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        if (tick >= 10 && tick < 15)
        {
            m_chunk->Data1.Set(m_chunk->Data1.Get() + 1);
            m_chunk->RPCTest();
            ++m_numRpcCalled;
        }
        else if (tick >= 15 && tick < 20)
        {
            for (int i = sHost + 1; i < nSessions; ++i)
            {
                ReplicaPtr rep = m_sessions[i].GetReplicaMgr().FindReplica(m_chunk->GetReplicaId());
                AZ_Assert(rep, "No replica in the session %d\n", i);

                auto chunkPtr = rep->FindReplicaChunk<SuspendUpdatesChunk>();
                AZ_Assert(chunkPtr, "No SuspendUpdatesChunk is not found on replica\n");

                // rpcs and datasets updates should not be called
                AZ_TEST_ASSERT(chunkPtr->m_numDataSetChanged == 0);
                AZ_TEST_ASSERT(chunkPtr->m_numRpcCalled == 0);
            }
        }
        else if (tick == 20)
        {
            for (int i = sHost + 1; i < nSessions; ++i)
            {
                ReplicaPtr rep = m_sessions[i].GetReplicaMgr().FindReplica(m_chunk->GetReplicaId());
                AZ_Assert(rep, "No replica in the session %d\n", i);
                auto chunkPtr = rep->FindReplicaChunk<SuspendUpdatesChunk>();
                AZ_Assert(chunkPtr, "No SuspendUpdatesChunk is not found on replica\n");

                chunkPtr->ResumeUpdatesFromReplica();
            }
        }
        else if (tick == 25)
        {
            for (int i = sHost + 1; i < nSessions; ++i)
            {
                ReplicaPtr rep = m_sessions[i].GetReplicaMgr().FindReplica(m_chunk->GetReplicaId());
                AZ_Assert(rep, "No replica in the session %d\n", i);

                auto chunkPtr = rep->FindReplicaChunk<SuspendUpdatesChunk>();
                AZ_Assert(chunkPtr, "SuspendUpdatesChunk is not found on replica\n");

                // all rpcs and datasets callback should be called here
                AZ_TEST_ASSERT(chunkPtr->m_numDataSetChanged == 1);
                AZ_TEST_ASSERT(chunkPtr->m_numRpcCalled == m_numRpcCalled);
            }

            return TestStatus::Completed;
        }

        return TestStatus::Running;
    });
}


class Integ_BasicHostChunkDescriptorTest
    : public UnitTest::GridMateMPTestFixture
    , public ::testing::Test
{
public:
    enum
    {
        Host,
        Client,
        nNodes
    };

    class HostChunk : public ReplicaChunk
    {
    public:
        GM_CLASS_ALLOCATOR(HostChunk);

        static const char* GetChunkName() { return "BasicHostChunkDescriptorTest::HostChunk"; }

        bool IsReplicaMigratable() override { return false; }

        void OnReplicaActivate(const GridMate::ReplicaContext&) override
        {
            if (IsMaster())
            {
                nMasterActivations++;
            }
            else
            {
                nProxyActivations++;
            }
        }

        static int nMasterActivations;
        static int nProxyActivations;
    };
};
int Integ_BasicHostChunkDescriptorTest::HostChunk::nMasterActivations = 0;
int Integ_BasicHostChunkDescriptorTest::HostChunk::nProxyActivations = 0;

TEST_F(Integ_BasicHostChunkDescriptorTest, BasicHostChunkDescriptorTest)
{
    AZ_TracePrintf("GridMate", "\n");

    // Register test chunks
    ReplicaChunkDescriptorTable::Get().RegisterChunkType<HostChunk, GridMate::BasicHostChunkDescriptor<HostChunk>>();

    MPSession nodes[nNodes];

    // initialize transport
    int basePort = 4427;
    for (int i = 0; i < nNodes; ++i)
    {
        TestCarrierDesc desc;
        desc.m_port = basePort + i;
        // initialize replica managers
        nodes[i].SetTransport(DefaultCarrier::Create(desc, m_gridMate));
        nodes[i].AcceptConn(true);
        nodes[i].SetClient(i != Host);
        nodes[i].GetReplicaMgr().Init(ReplicaMgrDesc(i + 1, nodes[i].GetTransport(), 0, i == 0 ? ReplicaMgrDesc::Role_SyncHost : 0));
    }

    // connect Client to Host
    nodes[Client].GetTransport()->Connect("127.0.0.1", basePort);

    ReplicaPtr hostReplica;
    ReplicaPtr clientReplica;

    // main test loop
    for (int tick = 0; tick < 1000; ++tick)
    {
        if (tick == 100)
        {
            for (int i = 0; i < nNodes; ++i)
            {
                AZ_TEST_ASSERT(nodes[i].GetReplicaMgr().IsReady());
            }
        }

        if (tick == 200)
        {
            hostReplica = Replica::CreateReplica("HostReplica");
            hostReplica->AttachReplicaChunk(CreateReplicaChunk<HostChunk>());
            nodes[Host].GetReplicaMgr().AddMaster(hostReplica);
        }

        if (tick == 300)
        {
            AZ_TEST_ASSERT(HostChunk::nMasterActivations == 1);
            AZ_TEST_ASSERT(HostChunk::nProxyActivations == 1);
            AZ_TEST_ASSERT(nodes[Client].GetReplicaMgr().FindReplica(hostReplica->GetRepId())->FindReplicaChunk<HostChunk>());

            AZ_TEST_START_TRACE_SUPPRESSION;
            clientReplica = Replica::CreateReplica("ClientReplica");
            clientReplica->AttachReplicaChunk(CreateReplicaChunk<HostChunk>());
            nodes[Client].GetReplicaMgr().AddMaster(clientReplica);
        }

        if (tick == 400)
        {
            AZ_TEST_STOP_TRACE_SUPPRESSION(1);
            AZ_TEST_ASSERT(HostChunk::nMasterActivations == 2);
            AZ_TEST_ASSERT(HostChunk::nProxyActivations == 1);
            AZ_TEST_ASSERT(!nodes[Host].GetReplicaMgr().FindReplica(clientReplica->GetRepId())->FindReplicaChunk<HostChunk>());
        }

        // tick everything
        for (int i = 0; i < nNodes; ++i)
        {
            nodes[i].Update();
            nodes[i].GetReplicaMgr().Unmarshal();
            nodes[i].GetReplicaMgr().UpdateReplicas();
            nodes[i].GetReplicaMgr().UpdateFromReplicas();
            nodes[i].GetReplicaMgr().Marshal();
            nodes[i].GetTransport()->Update();
        }

        AZStd::this_thread::sleep_for(AZStd::chrono::milliseconds(10));
    }

    hostReplica = nullptr;
    clientReplica = nullptr;

    for (int i = 0; i < nNodes; ++i)
    {
        nodes[i].GetReplicaMgr().Shutdown();
        DefaultCarrier::Destroy(nodes[i].GetTransport());
    }
}

/*
 * Create and immedietly destroy master replica
 * Test that it does not result in any network sync
*/
class Integ_CreateDestroyMaster
    : public Integ_SimpleTest
    , public Debug::ReplicaDrillerBus::Handler
{
public:
    enum
    {
        sHost,
        s2,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }


    // ReplicaDrillerBus
    void OnReceive(PeerId from, const void* data, size_t len) override
    {
        (void)from;
        (void)data;
        (void)len;

        AZ_TEST_ASSERT(false); // should not receive any replica data
    }

    void ConnectDriller()
    {
        Debug::ReplicaDrillerBus::Handler::BusConnect();
    }

    void DisconnectDriller()
    {
        Debug::ReplicaDrillerBus::Handler::BusDisconnect();
    }
};

TEST_F(Integ_CreateDestroyMaster, CreateDestroyMaster)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        switch (tick)
        {

        case 10:
        {
            ConnectDriller();
            auto replica = Replica::CreateReplica(nullptr);
            CreateAndAttachReplicaChunk<DataSetChunk>(replica);
            m_sessions[sHost].GetReplicaMgr().AddMaster(replica);

            // Destroying replica right away
            replica->Destroy();
            break;
        }

        case 20:
            DisconnectDriller();
            return TestStatus::Completed;
        default: break;
        }

        return TestStatus::Running;
    });
}

/*
* This test checks that when the carrier ACKs a message it feeds back to the ReplicaTarget.
* The ReplicaTarget will prevent sending more updates.
*/
class ReplicaACKfeedbackTestFixture
    : public Integ_SimpleTest
{
public:
    ReplicaACKfeedbackTestFixture()
        : m_replicaId(InvalidReplicaId)
    {
    }

    enum
    {
        sHost,
        sClient,
        nSessions
    };

    int GetNumSessions() override { return nSessions; }

    static const int NonDefaultValue = 4242;
    static const int k_headerBytes = 12;
    static const int k_updateBytes = k_headerBytes + 10 * sizeof(int);

    void PreConnect() override
    {
        m_driller.BusConnect();

        ReplicaPtr replica = Replica::CreateReplica("ReplicaACKfeedbackTest");
        LargeChunkWithDefaultsMedium* chunk = CreateAndAttachReplicaChunk<LargeChunkWithDefaultsMedium>(replica);
        AZ_TEST_ASSERT(chunk);

        m_replicaId = m_sessions[sHost].GetReplicaMgr().AddMaster(replica);
    }

    ~ReplicaACKfeedbackTestFixture()
    {
        m_driller.BusDisconnect();
    }

    size_t                              m_replicaBytesSentPrev = 0;
    ReplicaId                           m_replicaId;
    Integ_ReplicaDriller::ReplicaDrillerHook  m_driller;
};

TEST_F(ReplicaACKfeedbackTestFixture, ReplicaACKfeedbackTest)
{
    RunTickLoop([this](int tick)-> TestStatus
    {
        if(! ReplicaTarget::IsAckEnabled())
        {
            return TestStatus::Completed;
        }

        //AZ_Printf("GridMateTests", "%d %d\n", tick, m_driller.m_numReplicaBytesSent);
        // Tests the Revision stamp with Carrier ACK feedback
        // result is true on the immediate tick after changing, but false on the next and stays false until next change
        auto CheckHostReplicaChanged = [this](bool result)
        {
            ReplicaPtr replica = m_sessions[sHost].GetReplicaMgr().FindReplica(m_replicaId);
            if (replica)
            {
                for (auto& dst : replica->DebugGetTargets() )
                {
                    const bool targetHasUnAckdData = ReplicaTarget::IsAckEnabled() && dst.HasOldRevision(replica->GetRevision());
                    //AZ_Printf("GridMateTests", "target's stamp %d replica last_change %d : assert(%d)\n",
                    //    dst.GetRevision(), replica->GetRevision(), targetHasUnAckdData == result);
                    AZ_TEST_ASSERT(targetHasUnAckdData == result);
                }
            }
        };
        auto updateDataSets = [](AZStd::intrusive_ptr<LargeChunkWithDefaultsMedium>& chunk, int val)
        {
            AZ_TEST_ASSERT(chunk);
            auto touch = [val](DataSet<int>& dataSet) { dataSet.Set(val); };
            touch(chunk->Data1);
            touch(chunk->Data2);
            touch(chunk->Data3);
            touch(chunk->Data4);
            touch(chunk->Data5);
            touch(chunk->Data6);
            touch(chunk->Data7);
            touch(chunk->Data8);
            touch(chunk->Data9);
            touch(chunk->Data10);
        };

        switch (tick)
        {
        case 6:
        {
            CheckHostReplicaChanged(false);     //Default value sent reliably. Called back immediately. Nothing to ACK.
        }break;
        case 10:
        {
            auto rep = m_sessions[sClient].GetReplicaMgr().FindReplica(m_replicaId);
            AZ_TEST_ASSERT(rep);                //Client has recvd
        }break;
        case 15:
        {
            ReplicaPtr replica = m_sessions[sHost].GetReplicaMgr().FindReplica(m_replicaId);
            auto chunk = replica->FindReplicaChunk<LargeChunkWithDefaultsMedium>();

            updateDataSets(chunk, NonDefaultValue);

            m_replicaBytesSentPrev = m_driller.m_numReplicaBytesSent;
            CheckHostReplicaChanged(false);  //Changed now, but wont know until next prepareData()
        }break;
        case 16:
        {
            AZ_TEST_ASSERT(m_driller.m_numReplicaBytesSent - m_replicaBytesSentPrev == k_updateBytes);
            CheckHostReplicaChanged(true);  //Detected change. ACK feedback on next tick returns to false.
        }break;
        case 20:
        {
            ReplicaPtr replica = m_sessions[sHost].GetReplicaMgr().FindReplica(m_replicaId);
            auto chunk = replica->FindReplicaChunk<LargeChunkWithDefaultsMedium>();

            updateDataSets(chunk, NonDefaultValue + 1);

            m_replicaBytesSentPrev = m_driller.m_numReplicaBytesSent;
            CheckHostReplicaChanged(false);  //Changed now, but wont know until next prepareData()
        }break;
        case 21:
        {
            AZ_TEST_ASSERT(m_driller.m_numReplicaBytesSent - m_replicaBytesSentPrev == k_updateBytes);
            CheckHostReplicaChanged(true);  //Detected change. ACK feedback on next tick returns to false.
        }break;
        case 25:
        {
            CheckHostReplicaChanged(false);
            m_sessions[sHost].GetReplicaMgr().FindReplica(m_replicaId)->Destroy();
        } break;
        case 30:
        {
            return TestStatus::Completed;
        }
        default:
            CheckHostReplicaChanged(false);      //All other ticks leave Replica unchanged!
            break;
        }

        return TestStatus::Running;
    });
}

} // namespace UnitTest