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 <IceUtil/Thread.h>
#include <IceUtil/Random.h>
#include <Ice/Ice.h>
#include <IceGrid/Registry.h>
#include <IceGrid/Admin.h>
#include <IceGrid/Query.h>
#include <Glacier2/Router.h>
#include <TestCommon.h>
#include <Test.h>

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

class Callback : public IceUtil::Monitor<IceUtil::Mutex>
{
public:

    Callback() : _response(false), _exception(false)
    {
    }

    void
    response(const Ice::ObjectPrx& obj)
    {
        Lock sync(*this);
        _response = true;
        _obj = obj;
        notify();
    }

    void
    exception()
    {
        Lock sync(*this);
        _exception = true;
        notify();
    }

    void
    waitResponse(const char* file, int line)
    {
        Lock sync(*this);
        while(!_response && !_exception)
        {
            if(!timedWait(IceUtil::Time::seconds(3)))
            {
                cerr << "timeout: " << file << ":" << line << endl;
                test(false); // Timeout
            }
        }
    }

    bool
    hasResponse(Ice::ObjectPrx& obj)
    {
        Lock sync(*this);
        obj = _obj;
        return _response;
    }

    bool
    hasException()
    {
        Lock sync(*this);
        return _exception;
    }

private:

    bool _response;
    bool _exception;
    Ice::ObjectPrx _obj;
};

class AllocateObjectByIdCallback : public AMI_Session_allocateObjectById, public Callback
{
public:

    virtual void ice_response(const Ice::ObjectPrx& obj) { response(obj); }
    virtual void ice_exception(const Ice::Exception&) { exception(); }
};
typedef IceUtil::Handle<AllocateObjectByIdCallback> AllocateObjectByIdCallbackPtr;

class AllocateObjectByTypeCallback : public AMI_Session_allocateObjectByType, public Callback
{
public:

    virtual void ice_response(const Ice::ObjectPrx& obj) { response(obj); }
    virtual void ice_exception(const Ice::Exception&) { exception(); }
};
typedef IceUtil::Handle<AllocateObjectByTypeCallback> AllocateObjectByTypeCallbackPtr;

class StressClient : public IceUtil::Thread, public IceUtil::Monitor<IceUtil::Mutex>
{
public:
    
    StressClient(int id, const RegistryPrx& registry, bool destroySession) : 
        _communicator(registry->ice_getCommunicator()),
        _id(id),
        _registry(registry),
        _notified(false),
        _terminated(false),
        _destroySession(destroySession)
    {
    }

    StressClient(int id, const SessionPrx& session) : 
        _communicator(session->ice_getCommunicator()),
        _id(id),
        _session(session),
        _notified(false),
        _terminated(false),
        _destroySession(false)
    {
    }

    virtual 
    void run()
    {
        {
            Lock sync(*this);
            while(!_notified)
            {
                wait();
            }
        }
        
        SessionPrx session;
        while(true)
        {
            {
                Lock sync(*this);
                if(_terminated)
                {
                    if(!_session && session)
                    {
                        session->destroy();
                    }
                    return;
                }
            }
            
            if(!session)
            {
                ostringstream os;
                os << "Client-" << _id;
                if(_session)
                {
                    session = _session;
                }
                else
                {
                    session = _registry->createSession(os.str(), "");
                    session->setAllocationTimeout(IceUtil::random(200)); // 200ms timeout
                }
            }

            assert(session);
            session->keepAlive();

            Ice::ObjectPrx object;
            switch(IceUtil::random(_destroySession ? 4 : 2))
            {
            case 0:
                object = allocate(session);
                break;
            case 1:
                object = allocateByType(session);
                break;
            case 2:
                assert(!_session);
                allocateAndDestroy(session);
                session = 0;
                break;
            case 3:
                assert(!_session);
                allocateByTypeAndDestroy(session);
                session = 0;
                break;
            }

            if(object)
            {
                IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(IceUtil::random(20)));
                switch(IceUtil::random(_destroySession ? 2 : 1))
                {
                case 0:
                    session->releaseObject(object->ice_getIdentity());
                    break;
                case 1:
                    assert(!_session);
                    session->destroy();
                    session = 0;
                    break;
                }
            }
        }
    }

    Ice::ObjectPrx
    allocate(const SessionPrx& session)
    {
        ostringstream os;
        os << "stress-" << IceUtil::random(6) + 1;
        try
        {
            return session->allocateObjectById(_communicator->stringToIdentity(os.str()));
        }
        catch(const AllocationTimeoutException&)
        {
        }
        catch(const AllocationException&)
        {
            // This can only happen if we are using the common session 
            // and the object is already allocated.
            test(_session);
        }
        return 0;
    }

    Ice::ObjectPrx
    allocateByType(const SessionPrx& session)
    {
        try
        {
            return session->allocateObjectByType("::StressTest");
        }
        catch(const AllocationTimeoutException&)
        {
        }
        return 0;
    }

    void
    allocateAndDestroy(const SessionPrx& session)
    {
        ostringstream os;
        os << "stress-" << IceUtil::random(3);
        session->allocateObjectById_async(new AllocateObjectByIdCallback(), _communicator->stringToIdentity(os.str()));
        session->destroy();
    }

    void
    allocateByTypeAndDestroy(const SessionPrx& session)
    {
        session->allocateObjectByType_async(new AllocateObjectByTypeCallback(), "::StressTest");
        session->destroy();
    }

    void
    notifyThread()
    {
        Lock sync(*this);
        _notified = true;
        notify();
    }

    void
    terminate()
    {
        Lock sync(*this);
        _terminated = true;
        notify();
    }

protected:

    const Ice::CommunicatorPtr _communicator;
    const int _id;
    const RegistryPrx _registry;
    const SessionPrx _session;
    bool _notified;
    bool _terminated;
    const bool _destroySession;
};
typedef IceUtil::Handle<StressClient> StressClientPtr;

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

    SessionKeepAliveThread(const Ice::LoggerPtr& logger, const IceUtil::Time& timeout) :
        _logger(logger),
        _timeout(timeout),
        _terminated(false)
    {
    }

    virtual void
    run()
    {
        Lock sync(*this);
        while(!_terminated)
        {
            timedWait(_timeout);
            if(!_terminated)
            {
                vector<SessionPrx>::iterator p = _sessions.begin();
                while(p != _sessions.end())
                {
                    try
                    {
                        (*p)->keepAlive();
                        ++p;
                    }
                    catch(const Ice::Exception&)
                    {
                        p = _sessions.erase(p);
                    }
                }

                vector<AdminSessionPrx>::iterator q = _adminSessions.begin();
                while(q != _adminSessions.end())
                {
                    try
                    {
                        (*q)->keepAlive();
                        ++q;
                    }
                    catch(const Ice::Exception&)
                    {
                        q = _adminSessions.erase(q);
                    }
                }
            }
        }
    }

    void 
    add(const SessionPrx& session)
    {
        Lock sync(*this);
        _sessions.push_back(session);
    }

    void 
    add(const AdminSessionPrx& session)
    {
        Lock sync(*this);
        _adminSessions.push_back(session);
    }

    void
    terminate()
    {
        Lock sync(*this);
        _terminated = true;
        notify();
    }

private:

    const Ice::LoggerPtr _logger;
    vector<SessionPrx> _sessions;
    vector<AdminSessionPrx> _adminSessions;
    const IceUtil::Time _timeout;
    bool _terminated;
};
typedef IceUtil::Handle<SessionKeepAliveThread> SessionKeepAliveThreadPtr;

void 
allTests(const Ice::CommunicatorPtr& communicator)
{
    SessionKeepAliveThreadPtr keepAlive = new SessionKeepAliveThread(
        communicator->getLogger(), IceUtil::Time::seconds(5));
    keepAlive->start();

    RegistryPrx registry = IceGrid::RegistryPrx::checkedCast(communicator->stringToProxy("IceGrid/Registry"));
    test(registry);
    AdminSessionPrx session = registry->createAdminSession("foo", "bar");

    keepAlive->add(session);

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

    cout << "starting router... " << flush;
    try
    {
        admin->startServer("Glacier2");