alltests.cpp

ICE-3.2 一个开源的中间件

// **********************************************************************
//
// Copyright (c) 2003-2007 ZeroC, Inc. All rights reserved.
//
// This copy of Ice is licensed to you under the terms described in the
// ICE_LICENSE file included in this distribution.
//
// **********************************************************************

#include <Ice/Ice.h>
#include <Ice/BuiltinSequences.h>
#include <Ice/IdentityUtil.h>
#include <Ice/Locator.h>
#include <IceGrid/Query.h>
#include <IceGrid/Registry.h>
#include <IceGrid/Admin.h>
#include <IceGrid/UserAccountMapper.h>
#include <IceUtil/Thread.h>
#include <TestCommon.h>
#include <Test.h>

using namespace std;
using namespace Test;
using namespace IceGrid;

class SessionKeepAliveThread : public IceUtil::Thread, public IceUtil::Monitor<IceUtil::Mutex>
{
public:

    SessionKeepAliveThread(const IceGrid::AdminSessionPrx& session, long timeout) :
        _session(session),
        _timeout(IceUtil::Time::seconds(timeout)),
        _destroy(false)
    {
    }

    virtual void
    run()
    {
        Lock sync(*this);
        while(!_destroy)
        {
            timedWait(_timeout);
            if(_destroy)
            {
                break;
            }
            try
            {
                _session->keepAlive();
            }
            catch(const Ice::Exception&)
            {
                break;
            }
        }
    }

    void
    destroy()
    {
        Lock sync(*this);
        _destroy = true;
        notify();
    }

private:

    IceGrid::AdminSessionPrx _session;
    const IceUtil::Time _timeout;
    bool _destroy;
};
typedef IceUtil::Handle<SessionKeepAliveThread> SessionKeepAliveThreadPtr;

void
waitForServerState(const IceGrid::AdminPrx& admin, const std::string& server, bool up)
{
    int nRetry = 0;
    while(nRetry < 30)
    {
        if(admin->getServerState(server) == (up ? Active : Inactive))
        {
            return;
        } 

        IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(500));
        ++nRetry;
    }
    test(false);
}

void
waitForNodeState(const IceGrid::AdminPrx& admin, const std::string& node, bool up)
{
    int nRetry = 0;
    while(nRetry < 30)
    {
        try
        {
            if(admin->pingNode(node) == up) // Wait for the node to be removed.
            {
                return;
            }
        }
        catch(const NodeNotExistException&)
        {
            if(!up)
            {
                return;
            }
        }
        
        IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(500));
        ++nRetry;
    }
    try
    {
        if(admin->pingNode(node) != up)
        {
            cerr << "node state change timed out:" << endl;
            cerr << "node: " << node << endl;
            cerr << "state: " << up << endl;
        }
    }
    catch(NodeNotExistException&)
    {
        if(up)
        {
            cerr << "node state change timed out:" << endl;
            cerr << "node: " << node << endl;
            cerr << "state: " << up << endl;
        }
    }
}

void
instantiateServer(const AdminPrx& admin, const string& templ, const map<string, string>& params)
{
    ServerInstanceDescriptor desc;
    desc._cpp_template = templ;
    desc.parameterValues = params;
    NodeUpdateDescriptor nodeUpdate;
    nodeUpdate.name = "localnode";
    nodeUpdate.serverInstances.push_back(desc);
    ApplicationUpdateDescriptor update;
    update.name = "Test";
    update.nodes.push_back(nodeUpdate);
    try
    {
        admin->updateApplication(update);
    }
    catch(DeploymentException& ex)
    {
        cerr << ex.reason << endl;
        test(false);
    }
    catch(const Ice::LocalException& ex)
    {
        cerr << ex << endl;
        test(false);
    }
}

void
removeServer(const AdminPrx& admin, const string& id)
{
    try
    {
        admin->stopServer(id);
    }
    catch(const ServerStopException&)
    {
    }
    catch(const NodeUnreachableException&)
    {
    }
    catch(const Ice::UserException& ex)
    {
        cerr << ex << endl;
        test(false);
    }

    NodeUpdateDescriptor nodeUpdate;
    nodeUpdate.name = "localnode";
    nodeUpdate.removeServers.push_back(id);
    ApplicationUpdateDescriptor update;
    update.name = "Test";
    update.nodes.push_back(nodeUpdate);
    try
    {
        admin->updateApplication(update);
    }
    catch(DeploymentException& ex)
    {
        cerr << ex.reason << endl;
        test(false);
    }
}

bool
waitAndPing(const Ice::ObjectPrx& obj)
{
    int nRetry = 0;
    while(nRetry < 30)
    {
        try
        {
            obj->ice_ping();
            return true;
        }
        catch(const Ice::LocalException&)
        {
            IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(500));
            ++nRetry;
        }
    }
    return false;
}

AdminPrx
createAdminSession(const Ice::LocatorPrx& locator, const string& replica)
{
    test(waitAndPing(locator));
    
    string registryStr("TestIceGrid/Registry");
    if(!replica.empty() && replica != "Master")
    {
        registryStr += "-" + replica;
    }
    Ice::ObjectPrx obj = locator->ice_getCommunicator()->stringToProxy(registryStr)->ice_locator(locator);
    RegistryPrx registry = RegistryPrx::checkedCast(obj);
    test(registry);

    AdminSessionPrx session = AdminSessionPrx::checkedCast(registry->createAdminSession("foo", "bar"));
    test(session);
    return session->getAdmin();
}

void
allTests(const Ice::CommunicatorPtr& comm)
{
    RegistryPrx registry = IceGrid::RegistryPrx::checkedCast(comm->stringToProxy("IceGrid/Registry"));
    test(registry);
    AdminSessionPrx session = registry->createAdminSession("foo", "bar");

    SessionKeepAliveThreadPtr keepAlive = new SessionKeepAliveThread(session, registry->getSessionTimeout() / 2);
    keepAlive->start();

    AdminPrx admin = session->getAdmin();
    test(admin);

    map<string, string> params;

    params.clear();
    params["id"] = "Master";
    params["replicaName"] = "";
    params["port"] = "12050";
    instantiateServer(admin, "IceGridRegistry", params);
    
    params.clear();
    params["id"] = "Slave1";
    params["replicaName"] = "Slave1";
    params["port"] = "12051";
    instantiateServer(admin, "IceGridRegistry", params);
    
    params.clear();
    params["id"] = "Slave2";
    params["replicaName"] = "Slave2";
    params["port"] = "12052";
    instantiateServer(admin, "IceGridRegistry", params);

    Ice::LocatorPrx masterLocator = 
        Ice::LocatorPrx::uncheckedCast(comm->stringToProxy("TestIceGrid/Locator-Master:default -p 12050"));
    Ice::LocatorPrx slave1Locator = 
        Ice::LocatorPrx::uncheckedCast(comm->stringToProxy("TestIceGrid/Locator-Slave1:default -p 12051"));
    Ice::LocatorPrx slave2Locator = 
        Ice::LocatorPrx::uncheckedCast(comm->stringToProxy("TestIceGrid/Locator-Slave2:default -p 12052"));

    Ice::LocatorPrx replicatedLocator = 
        Ice::LocatorPrx::uncheckedCast(comm->stringToProxy("TestIceGrid/Locator:default -p 12050:default -p 12051"));

    AdminPrx masterAdmin, slave1Admin, slave2Admin;

    admin->startServer("Master");
    masterAdmin = createAdminSession(masterLocator, "");
    
    admin->startServer("Slave1");
    slave1Admin = createAdminSession(slave1Locator, "Slave1");

    //
    // Test replication and well-known objects:
    //
    // - Locator interface
    // - Query interface
    //
    // - Registry object
    // - RegistryUserAccountMapper
    // - SessionManager/SSLSessionManager
    // - AdminSessionManager/AdminSSLSessionManager
    //
    cout << "testing replicated locator and query interface... " << flush;
    {
        Ice::EndpointSeq endpoints;
        ObjectInfo info;
        
        info = masterAdmin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Locator"));
        ObjectInfo info1 = slave1Admin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Locator"));
        test(slave1Admin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Locator")) == info);
        test(info.type == Ice::Locator::ice_staticId());
        endpoints = info.proxy->ice_getEndpoints();
        test(endpoints.size() == 2);
        test(endpoints[0]->toString().find("-p 12050") != string::npos);
        test(endpoints[1]->toString().find("-p 12051") != string::npos);

        info = masterAdmin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Query"));
        test(slave1Admin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Query")) == info);
        test(info.type == IceGrid::Query::ice_staticId());
        endpoints = info.proxy->ice_getEndpoints();
        test(endpoints.size() == 2);
        test(endpoints[0]->toString().find("-p 12050") != string::npos);
        test(endpoints[1]->toString().find("-p 12051") != string::npos);

        admin->startServer("Slave2");
        slave2Admin = createAdminSession(slave2Locator, "Slave2");

        info = masterAdmin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Locator"));
        // We eventually need to wait here for the update of the replicated objects to propagate to the replica.
        int nRetry = 0;
        while(slave1Admin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Locator")) != info && nRetry < 30)
        {
            IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(500));
            ++nRetry;
        }
        test(slave2Admin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Locator")) == info);
        test(info.type == Ice::Locator::ice_staticId());
        endpoints = info.proxy->ice_getEndpoints();
        test(endpoints.size() == 3);
        test(endpoints[0]->toString().find("-p 12050") != string::npos);
        test(endpoints[1]->toString().find("-p 12051") != string::npos);
        test(endpoints[2]->toString().find("-p 12052") != string::npos);

        info = masterAdmin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Query"));
        // We eventually need to wait here for the update of the replicated objects to propagate to the replica.
        nRetry = 0;
        while(slave1Admin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Query")) != info && nRetry < 30)
        {
            IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(500));
            ++nRetry;
        }
        test(slave2Admin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Query")) == info);
        test(info.type == IceGrid::Query::ice_staticId());
        endpoints = info.proxy->ice_getEndpoints();
        test(endpoints.size() == 3);
        test(endpoints[0]->toString().find("-p 12050") != string::npos);
        test(endpoints[1]->toString().find("-p 12051") != string::npos);
        test(endpoints[2]->toString().find("-p 12052") != string::npos);

        slave2Admin->shutdown();
        waitForServerState(admin, "Slave2", false);

        info = masterAdmin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Locator"));
        // We eventually need to wait here for the update of the replicated objects to propagate to the replica.
        nRetry = 0;
        while(slave1Admin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Locator")) != info && nRetry < 30)
        {
            IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(500));
            ++nRetry;
        }
        test(slave1Admin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Locator")) == info);
        test(info.type == Ice::Locator::ice_staticId());
        endpoints = info.proxy->ice_getEndpoints();
        test(endpoints.size() == 2);
        test(endpoints[0]->toString().find("-p 12050") != string::npos);
        test(endpoints[1]->toString().find("-p 12051") != string::npos);

        info = masterAdmin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Query"));
        nRetry = 0;
        while(slave1Admin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Query")) != info && nRetry < 30)
        {
            IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(500));
            ++nRetry;
        }
        test(slave1Admin->getObjectInfo(comm->stringToIdentity("TestIceGrid/Query")) == info);
        test(info.type == IceGrid::Query::ice_staticId());
        endpoints = info.proxy->ice_getEndpoints();
        test(endpoints.size() == 2);
        test(endpoints[0]->toString().find("-p 12050") != string::npos);
        test(endpoints[1]->toString().find("-p 12051") != string::npos);

        QueryPrx query;
        query = QueryPrx::uncheckedCast(comm->stringToProxy("TestIceGrid/Query:" + endpoints[0]->toString()));
        Ice::ObjectProxySeq objs = query->findAllObjectsByType("::IceGrid::Registry");
        test(objs.size() == 2);
        query = QueryPrx::uncheckedCast(comm->stringToProxy("TestIceGrid/Query:" + endpoints[1]->toString()));