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 <Ice/Ice.h>
#include <IceGrid/Registry.h>
#include <IceGrid/Query.h>
#include <IceGrid/Session.h>
#include <IceGrid/Admin.h>
#include <IceGrid/Observer.h>
#include <Glacier2/Router.h>
#include <TestCommon.h>

using namespace std;
using namespace IceGrid;

void 
addProperty(const CommunicatorDescriptorPtr& communicator, const string& name, const string& value)
{
    PropertyDescriptor prop;
    prop.name = name;
    prop.value = value;
    communicator->propertySet.properties.push_back(prop);
}

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<AdminSessionPrx>::iterator p = _sessions.begin();
                while(p != _sessions.end())
                {
                    try
                    {
                        (*p)->keepAlive();
                        ++p;
                    }
                    catch(const Ice::Exception&)
                    {
                        p = _sessions.erase(p);
                    }
                }
            }
        }
    }

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

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

private:

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

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

    ObserverBase(const string& name) : _name(name), _updated(0)
    {
        _observers.insert(make_pair(name, this));
    }

    virtual ~ObserverBase()
    {
        _observers.erase(_name);
    }

    static void
    printStack()
    {
        map<string, ObserverBase*>::const_iterator p;
        for(p = _observers.begin(); p != _observers.end(); ++p)
        {
            vector<string>::const_iterator q = p->second->_stack.begin();
            if(p->second->_stack.size() > 10)
            {
                q = p->second->_stack.begin() + p->second->_stack.size() - 10;
            }
            cerr << "Last 10 updates of observer `" << p->second->_name << "':" << endl;
            for(; q != p->second->_stack.end(); ++q)
            {
                cerr << "  " << *q << endl;
            }
            p->second->_stack.clear();
        }
    }

    void
    trace(const string& msg)
    {
        _stack.push_back(msg);
    }

    void
    waitForUpdate(const char* file, int line)
    {
        Lock sync(*this);

        ostringstream os;
        os << "wait for update from line " << line;
        trace(os.str());

        while(!_updated)
        {
            if(!timedWait(IceUtil::Time::seconds(10)))
            {
                cerr << "timeout: " << file << ":" << line << endl;
                ObserverBase::printStack();
                test(false); // Timeout
            }
        }
        --_updated;
    }

protected:

    void
    updated(const string& update)
    {
        trace(update);
        ++_updated;
        notifyAll();
    }

    string _name;
    vector<string> _stack;
    int _updated;
    static map<string, ObserverBase*> _observers;
};
map<string, ObserverBase*> ObserverBase::_observers;

class ApplicationObserverI : public ApplicationObserver, public ObserverBase
{
public:

    ApplicationObserverI(const string& name) : ObserverBase(name)                                            
    {
    }
    
    virtual void 
    applicationInit(int serial, const ApplicationInfoSeq& apps, const Ice::Current&)
    {
        Lock sync(*this);
        for(ApplicationInfoSeq::const_iterator p = apps.begin(); p != apps.end(); ++p)
        {
            if(p->descriptor.name != "Test") // Ignore the test application from application.xml!
            {
                this->applications.insert(make_pair(p->descriptor.name, *p));
            }
        }
        updated(updateSerial(serial, "init update"));
    }

    virtual void
    applicationAdded(int serial, const ApplicationInfo& app, const Ice::Current&)
    {
        Lock sync(*this);
        this->applications.insert(make_pair(app.descriptor.name, app));
        updated(updateSerial(serial, "application added `" + app.descriptor.name + "'"));
    }

    virtual void 
    applicationRemoved(int serial, const std::string& name, const Ice::Current&)
    {
        Lock sync(*this);
        this->applications.erase(name);
        updated(updateSerial(serial, "application removed `" + name + "'"));
    }

    virtual void 
    applicationUpdated(int serial, const ApplicationUpdateInfo& info, const Ice::Current&)
    {
        Lock sync(*this);
        const ApplicationUpdateDescriptor& desc = info.descriptor;
        for(Ice::StringSeq::const_iterator q = desc.removeVariables.begin(); q != desc.removeVariables.end(); ++q)
        {
            this->applications[desc.name].descriptor.variables.erase(*q);
        }
        for(map<string, string>::const_iterator p = desc.variables.begin(); p != desc.variables.end(); ++p)
        {
            this->applications[desc.name].descriptor.variables[p->first] = p->second;
        }
        updated(updateSerial(serial, "application updated `" + desc.name + "'"));
    }

    int serial;
    map<string, ApplicationInfo> applications;

private:

    string
    updateSerial(int serial, const string& update)
    {
        this->serial = serial;
        ostringstream os;
        os << update << " (serial = " << serial << ")";
        return os.str();
    }
};
typedef IceUtil::Handle<ApplicationObserverI> ApplicationObserverIPtr;

class AdapterObserverI : public AdapterObserver, public ObserverBase
{
public:

    AdapterObserverI(const string& name) : ObserverBase(name)
    {
    }
    
    virtual void 
    adapterInit(const AdapterInfoSeq& adapters, const Ice::Current&)
    {
        Lock sync(*this);
        for(AdapterInfoSeq::const_iterator q = adapters.begin(); q != adapters.end(); ++q)
        {
            this->adapters.insert(make_pair(q->id, *q));
        }
        updated(updateSerial(0, "init update"));
    }

    void
    adapterAdded(const AdapterInfo& info, const Ice::Current&)
    {
        Lock sync(*this);
        this->adapters.insert(make_pair(info.id, info));
        updated(updateSerial(0, "adapter added `" + info.id + "'"));
    }

    void
    adapterUpdated(const AdapterInfo& info, const Ice::Current&)
    {
        Lock sync(*this);
        this->adapters[info.id] = info;
        updated(updateSerial(0, "adapter updated `" + info.id + "'"));
    }

    void
    adapterRemoved(const string& id, const Ice::Current&)
    {
        Lock sync(*this);
        this->adapters.erase(id);
        updated(updateSerial(0, "adapter removed `" + id + "'"));
    }

    int serial;
    map<string, AdapterInfo> adapters;

private:

    string
    updateSerial(int serial, const string& update)
    {
        this->serial = serial;
        ostringstream os;
        os << update << " (serial = " << serial << ")";
        return os.str();
    }
};
typedef IceUtil::Handle<AdapterObserverI> AdapterObserverIPtr;

class ObjectObserverI : public ObjectObserver, public ObserverBase
{
public:

    ObjectObserverI(const string& name) : ObserverBase(name)
    {
    }
    
    virtual void 
    objectInit(const ObjectInfoSeq& objects, const Ice::Current&)
    {
        Lock sync(*this);
        for(ObjectInfoSeq::const_iterator r = objects.begin(); r != objects.end(); ++r)
        {
            this->objects.insert(make_pair(r->proxy->ice_getIdentity(), *r));
        }
        updated(updateSerial(0, "init update"));
    }

    void
    objectAdded(const ObjectInfo& info, const Ice::Current&)
    {
        Lock sync(*this);
        this->objects.insert(make_pair(info.proxy->ice_getIdentity(), info));
        updated(updateSerial(0, "object added `" + info.proxy->ice_toString() + "'"));
    }

    void
    objectUpdated(const ObjectInfo& info, const Ice::Current&)
    {
        Lock sync(*this);
        this->objects[info.proxy->ice_getIdentity()] = info;
        updated(updateSerial(0, "object updated `" + info.proxy->ice_toString() + "'"));
    }

    void
    objectRemoved(const Ice::Identity& id, const Ice::Current& current)
    {
        Lock sync(*this);
        this->objects.erase(id);
        updated(updateSerial(0, "object removed `" + 
                             current.adapter->getCommunicator()->identityToString(id) + "'"));
    }

    int serial;
    map<Ice::Identity, ObjectInfo> objects;

private:

    string
    updateSerial(int serial, const string& update)
    {
        this->serial = serial;
        ostringstream os;
        os << update << " (serial = " << serial << ")";
        return os.str();
    }
};
typedef IceUtil::Handle<ObjectObserverI> ObjectObserverIPtr;

class NodeObserverI : public NodeObserver, public ObserverBase
{
public:

    NodeObserverI(const string& name) : ObserverBase(name)
    {
    }

    virtual void 
    nodeInit(const NodeDynamicInfoSeq& info, const Ice::Current& current)
    {
        Lock sync(*this);
        for(NodeDynamicInfoSeq::const_iterator p = info.begin(); p != info.end(); ++p)
        {
            this->nodes[p->info.name] = filter(*p);
        }
        updated("init");
    }

    virtual void
    nodeUp(const NodeDynamicInfo& info, const Ice::Current& current)
    {
        Lock sync(*this);
        this->nodes[info.info.name] = filter(info);
        updated("node `" + info.info.name + "' up");
    }

    virtual void
    nodeDown(const string& name, const Ice::Current& current)
    {
        Lock sync(*this);