client.cpp

来自「ICE-3.2 一个开源的中间件」· C++ 代码 · 共 905 行 · 第 1/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/Random.h>
#include <IceUtil/IceUtil.h>
#include <Ice/Application.h>
#include <Glacier2/Router.h>
#include <TestCommon.h>
#include <CallbackI.h>

using namespace std;
using namespace Ice;
using namespace Test;

static Ice::InitializationData initData;

class AMI_Callback_initiateConcurrentCallbackI : public AMI_Callback_initiateConcurrentCallback,
                                                 public IceUtil::Monitor<IceUtil::Mutex>
{
public:

    AMI_Callback_initiateConcurrentCallbackI() :
        _haveResponse(false)
    {
    }

    virtual void
    ice_response(Int response)
    {
        Lock sync(*this);
        _haveResponse = true;
        _response = response;
        notify();
    }

    virtual void
    ice_exception(const Exception& e)
    {
        Lock sync(*this);
        _haveResponse = true;
        _ex.reset(e.ice_clone());
        notify();
    }

    int
    waitResponse() const
    {
        Lock sync(*this);
        while(!_haveResponse)
        {
            if(!timedWait(IceUtil::Time::milliSeconds(5000)))
            {
                throw TimeoutException(__FILE__, __LINE__);
            }
        }
        if(_ex.get())
        {
            _ex->ice_throw();
        }
        return _response;
    }

private:

    bool _haveResponse;
    auto_ptr<Exception> _ex;
    Int _response;
};
typedef IceUtil::Handle<AMI_Callback_initiateConcurrentCallbackI> AMI_Callback_initiateConcurrentCallbackIPtr;

class MisbehavedClient : public IceUtil::Thread, public IceUtil::Monitor<IceUtil::Mutex>
{
public:
    
    MisbehavedClient(int id) : _id(id), _callback(false)
    {
    }

    virtual 
    void run()
    {
        CommunicatorPtr communicator = initialize(initData);
        ObjectPrx routerBase = communicator->stringToProxy("Glacier2/router:default -p 12347 -t 10000");
        Glacier2::RouterPrx router = Glacier2::RouterPrx::checkedCast(routerBase);
        communicator->setDefaultRouter(router);

        ostringstream os;
        os << "userid-" << _id;
        Glacier2::SessionPrx session = router->createSession(os.str(), "abc123");
        communicator->getProperties()->setProperty("Ice.PrintAdapterReady", "");
        ObjectAdapterPtr adapter = communicator->createObjectAdapterWithRouter("CallbackReceiverAdapter", router);
        adapter->activate();
        
        string category = router->getCategoryForClient();
        {
            Lock sync(*this);
            _callbackReceiver = new CallbackReceiverI;
            notify();
        }
        
        Identity ident;
        ident.name = "callbackReceiver";
        ident.category = category;
        CallbackReceiverPrx receiver = CallbackReceiverPrx::uncheckedCast(adapter->add(_callbackReceiver, ident));
        
        ObjectPrx base = communicator->stringToProxy("c1/callback:tcp -p 12010 -t 10000");
        base = base->ice_oneway();
        CallbackPrx callback = CallbackPrx::uncheckedCast(base);


        //
        // Block the CallbackReceiver in wait() to prevent the client from
        // processing other incoming calls and wait to receive the callback.
        //
        callback->initiateWaitCallback(receiver);
        test(_callbackReceiver->waitCallbackOK());

        //
        // Notify the main thread that the callback was received.
        //
        {
            Lock sync(*this);
            _callback = true;
            notify();
        }

        //
        // Callback the client with a large payload. This should cause
        // the Glacier2 request queue thread to block trying to send the
        // callback to the client because the client is currently blocked
        // in CallbackReceiverI::waitCallback() and can't process more 
        // requests.
        //
        callback->initiateCallbackWithPayload(receiver);
        test(_callbackReceiver->callbackWithPayloadOK());

        try
        {
            router->destroySession();
            test(false);
        }
        catch(const Ice::ConnectionLostException&)
        {
        }
        catch(const Ice::LocalException&)
        {
            test(false);
        }
        communicator->destroy();
    }

    void
    notifyWaitCallback()
    {
        _callbackReceiver->notifyWaitCallback();
    }

    void
    waitForCallback()
    {
        {
            Lock sync(*this);
            while(!_callback)
            {
                wait();
            }
        }
    }

private:

    int _id;
    CallbackReceiverIPtr _callbackReceiver;
    bool _callback;
};
typedef IceUtil::Handle<MisbehavedClient> MisbehavedClientPtr;

class StressClient : public IceUtil::Thread, public IceUtil::Monitor<IceUtil::Mutex>
{
public:
    
    StressClient(int id) : _id(id), _initialized(false), _notified(false)
    {
    }

    virtual 
    void run()
    {
        CommunicatorPtr communicator = initialize(initData);
        ObjectPrx routerBase = communicator->stringToProxy("Glacier2/router:default -p 12347 -t 30000");
        _router = Glacier2::RouterPrx::checkedCast(routerBase);
        communicator->setDefaultRouter(_router);

        ostringstream os;
        os << "userid-" << _id;
        Glacier2::SessionPrx session = _router->createSession(os.str(), "abc123");
        communicator->getProperties()->setProperty("Ice.PrintAdapterReady", "");
        ObjectAdapterPtr adapter = communicator->createObjectAdapterWithRouter("CallbackReceiverAdapter", _router);
        adapter->activate();
        
        string category = _router->getCategoryForClient();
        _callbackReceiver = new CallbackReceiverI;
        Identity ident;
        ident.name = "callbackReceiver";
        ident.category = category;
        CallbackReceiverPrx receiver = CallbackReceiverPrx::uncheckedCast(adapter->add(_callbackReceiver, ident));
        
        ObjectPrx base = communicator->stringToProxy("c1/callback:tcp -p 12010 -t 10000");
        base = base->ice_oneway();
        CallbackPrx callback = CallbackPrx::uncheckedCast(base);

        {
            Lock sync(*this);
            _initialized = true;
            notifyAll();
        }
        {
            Lock sync(*this);
            while(!_notified)
            {
                wait();
            }
        }
        
        //
        // Stress the router until the connection is closed.
        //
        stress(callback, receiver);
        communicator->destroy();
    }

    virtual void stress(const CallbackPrx& callback, const CallbackReceiverPrx&) = 0;

    void
    notifyThread()
    {
        {
            Lock sync(*this);
            while(!_initialized)
            {
                wait();
            }
        }
        {
            Lock sync(*this);
            _notified = true;
            notify();
        }
    }

    void
    kill()
    {
        try
        {
            _router->destroySession();
            test(false);
        }
        catch(const Ice::ConnectionLostException&)
        {
        }
        catch(const Ice::CommunicatorDestroyedException&)
        {
        }
        catch(const Ice::LocalException& ex)
        {
            cerr << ex << endl;
            test(false);
        }
    }

protected:

    Glacier2::RouterPrx _router;
    int _id;
    CallbackReceiverIPtr _callbackReceiver;
    bool _initialized;
    bool _notified;
};
typedef IceUtil::Handle<StressClient> StressClientPtr;

class PingStressClient : public StressClient
{
public:

    PingStressClient(int id) : StressClient(id) 
    {
    }

    virtual void
    stress(const CallbackPrx& callback, const CallbackReceiverPrx&)
    {
        try
        {
            CallbackPrx cb = CallbackPrx::uncheckedCast(callback->ice_twoway());
            Context context;
            context["_fwd"] = "t";
            while(true)
            {
                cb->ice_ping(context);
                IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(1));
            }
        }
        catch(const Ice::ConnectionLostException&)
        {
        }
        catch(const Ice::CommunicatorDestroyedException&)
        {
        }
        catch(const Ice::Exception& ex)
        {
            cerr << ex << endl;
            test(false);
        }
    }
};

class CallbackStressClient : public StressClient
{
public:

    CallbackStressClient(int id) : StressClient(id) 
    {
    }

    virtual void
    stress(const CallbackPrx& callback, const CallbackReceiverPrx& receiver)
    {
        try
        {
            CallbackPrx cb = CallbackPrx::uncheckedCast(callback->ice_twoway());
            Context context;
            context["_fwd"] = "t";
            while(true)
            {
                cb->initiateCallback(receiver, context);
                test(_callbackReceiver->callbackOK());
                IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(1));
            }
        }
        catch(const Ice::ConnectionLostException&)
        {
            // Session was destroyed.
        }
        catch(const Ice::ObjectNotExistException&)
        {
            // This might be raised by the CallbackI implementation if it can't invoke on the 
            // callback receiver because the session is being destroyed concurrently.
        }
        catch(const Ice::CommunicatorDestroyedException&)
        {
            // This might happen if the retry fails because the communicator is destroyed.
        }
        catch(const Ice::Exception& ex)
        {
            cerr << ex << endl;
            test(false);
        }
    }
};

class CallbackWithPayloadStressClient : public StressClient
{
public:

    CallbackWithPayloadStressClient(int id) : StressClient(id) 
    {
    }

    virtual void
    stress(const CallbackPrx& callback, const CallbackReceiverPrx& receiver)
    {
        try
        {
            CallbackPrx cb = CallbackPrx::uncheckedCast(callback->ice_twoway());
            Context context;
            context["_fwd"] = "t";
            while(true)
            {
                cb->initiateCallbackWithPayload(receiver, context);
                test(_callbackReceiver->callbackWithPayloadOK());
                IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(10));
            }
        }
        catch(const Ice::ConnectionLostException&)
        {
            // Session was destroyed.
        }
        catch(const Ice::ObjectNotExistException&)
        {
            // This might be raised by the CallbackI implementation if it can't invoke on the 
            // callback receiver because the session is being destroyed concurrently.
        }
        catch(const Ice::CommunicatorDestroyedException&)
        {
            // This might happen if the retry fails because the communicator is destroyed.
        }
        catch(const Ice::Exception& ex)
        {
            cerr << ex << endl;
            test(false);
        }
    }
};

class CallbackClient : public Application
{
public:

    virtual int run(int, char*[]);
};

int
main(int argc, char* argv[])
{
    //
    // We must disable connection warnings, because we attempt to ping
    // the router before session establishment, as well as after
    // session destruction. Both will cause a ConnectionLostException.
    //
    initData.properties = Ice::createProperties(argc, argv);
    initData.properties->setProperty("Ice.Warn.Connections", "0");

    CallbackClient app;
    return app.main(argc, argv, initData);
}

int
CallbackClient::run(int argc, char* argv[])
{
    ObjectPrx routerBase;

    {
        cout << "testing stringToProxy for router... " << flush;
        routerBase = communicator()->stringToProxy("Glacier2/router:default -p 12347 -t 10000");
        cout << "ok" << endl;
    }
    
    Glacier2::RouterPrx router;

    {
        cout << "testing checked cast for router... " << flush;
        router = Glacier2::RouterPrx::checkedCast(routerBase);
        test(router);
        cout << "ok" << endl;
    }