cimlistener.cpp

Pegasus is an open-source implementationof the DMTF CIM and WBEM standards. It is designed to be por

            pthread_testcancel();
#endif
            svc->runForever();
        }
    }
    catch(...)
    {
        Tracer::trace(TRC_SERVER, Tracer::LEVEL2,
                      "Unknown exception thrown in _listener_routine.");
    }

    // CAUTION: deleting the service also deletes the monitor whose tickle()
    // method may still be executing in another thread. This line of code was
    // most likely reached when the CIMListenerService::shutdown() method set
    // _dieNow to true and called Monitor::tickle(). We must wait until we
    // can obtain the _monitorMutex, indicating that we are no longer inside
    // Monitor::tickle().
    svc->_monitorMutex.lock();
    svc->_monitorMutex.unlock();
    delete svc;

    return 0;
}

////////////////////////////////////////////////////////////////////////////////
//
// CIMListenerRep
//
////////////////////////////////////////////////////////////////////////////////

class CIMListenerRep
{
public:
    CIMListenerRep(Uint32 portNumber, SSLContext * sslContext = NULL);
    ~CIMListenerRep();

    Uint32 getPortNumber() const;

    SSLContext *getSSLContext() const;
    void setSSLContext(SSLContext * sslContext);

    void start();
    void stop();

    Boolean isAlive();

    Boolean addConsumer(CIMIndicationConsumer * consumer);
    Boolean removeConsumer(CIMIndicationConsumer * consumer);

private:
    Boolean waitForPendingRequests(Uint32 shutdownTimeout);

    Uint32 _portNumber;
    SSLContext *_sslContext;

    CIMListenerIndicationDispatcher *_dispatcher;
    ThreadPool *_thread_pool;
    CIMListenerService *_svc;
    Semaphore *_listener_sem;
};

CIMListenerRep::CIMListenerRep(
    Uint32 portNumber, 
    SSLContext * sslContext)
    :
    _portNumber(portNumber), 
    _sslContext(sslContext),
    _dispatcher(new CIMListenerIndicationDispatcher()), 
    _thread_pool(NULL),
    _svc(NULL), 
    _listener_sem(NULL)
{
}

CIMListenerRep::~CIMListenerRep()
{
    // if port is alive, clean up the port
    if (_thread_pool != 0)
    {
        // Block incoming export requests and unbind the port
        _svc->stopClientConnection();

        // Wait until pending export requests in the server are done.
        waitForPendingRequests(10);

        // Shutdown the CIMListenerService
        _svc->shutdown();
    }

    delete _sslContext;
    delete _dispatcher;
    delete _thread_pool;
    delete _listener_sem;

    // don't delete _svc, this is deleted by _listener_routine
}

Uint32 CIMListenerRep::getPortNumber() const
{
    Uint32 portNumber;

    if (_svc == 0)
        portNumber = _portNumber;
    else
        portNumber = _svc->getPortNumber();

    return portNumber;
}

SSLContext *CIMListenerRep::getSSLContext() const
{
    return _sslContext;
}

void CIMListenerRep::setSSLContext(SSLContext * sslContext)
{
    delete _sslContext;
    _sslContext = sslContext;
}

void CIMListenerRep::start()
{
    // spawn a thread to do this
    if (_thread_pool == 0)
    {
        AutoPtr < CIMListenerService >
            svc(new CIMListenerService(_portNumber, _sslContext));

        svc->setIndicationDispatcher(_dispatcher);
        svc->init();

        struct timeval deallocateWait = { 15, 0 };
        AutoPtr < ThreadPool >
            threadPool(new ThreadPool(0, "Listener", 0, 1, deallocateWait));
        AutoPtr < Semaphore > sem(new Semaphore(0));

        if (threadPool->allocate_and_awaken(
            svc.get(), CIMListenerService::_listener_routine, sem.get()) 
            != PEGASUS_THREAD_OK)
        {
            Logger::put(
                Logger::STANDARD_LOG, System::CIMLISTENER,
                Logger::TRACE,
                "Not enough threads to start CIMListernerService.");

            Tracer::trace(
                TRC_SERVER, 
                Tracer::LEVEL2,
                "Could not allocate thread for "
                "CIMListenerService::_listener_routine.");
            throw
                Exception(MessageLoaderParms(
                    "Listener.CIMListener.CANNOT_ALLOCATE_THREAD",
                    "Could not allocate thread."));
        }

        Logger::put(
            Logger::STANDARD_LOG, 
            System::CIMLISTENER,
            Logger::INFORMATION, 
            "CIMListener started");

        _svc = svc.release();
        _thread_pool = threadPool.release();
        _listener_sem = sem.release();
    }
}

void CIMListenerRep::stop()
{
    if (_thread_pool != NULL)
    {
        // 
        // Graceful shutdown of the listener service
        // 

        // Block incoming export requests and unbind the port
        _svc->stopClientConnection();

        // Wait until pending export requests in the server are done.
        waitForPendingRequests(10);

        // Shutdown the CIMListenerService
        _svc->shutdown();

        // Wait for the _listener_routine thread to exit.
        // The thread could be delivering an export, so give it 3sec.
        // Note that _listener_routine deletes the CIMListenerService,
        // so no need to delete _svc.
        try
        {
            _listener_sem->time_wait(3000);
        }
        catch(const TimeOut &)
        {
            // No need to do anything, the thread pool will be deleted below
            // to cancel the _listener_routine thread if it is still running.
        }

        delete _listener_sem;
        _listener_sem = NULL;

        // Delete the thread pool.  This cancels the listener thread if it is 
        // still
        // running.
        delete _thread_pool;
        _thread_pool = NULL;

        Logger::put(
            Logger::STANDARD_LOG, System::CIMLISTENER,
            Logger::INFORMATION, "CIMListener stopped");
    }
}

Boolean CIMListenerRep::isAlive()
{
    return (_thread_pool != NULL) ? true : false;
}

Boolean CIMListenerRep::addConsumer(CIMIndicationConsumer * consumer)
{
    return _dispatcher->addConsumer(consumer);
}

Boolean CIMListenerRep::removeConsumer(CIMIndicationConsumer * consumer)
{
    return _dispatcher->removeConsumer(consumer);
}

Boolean CIMListenerRep::waitForPendingRequests(Uint32 shutdownTimeout)
{
    // Wait for 10 sec max
    Uint32 reqCount;
    Uint32 countDown = shutdownTimeout * 10;

    for (; countDown > 0; countDown--)
    {
        reqCount = _svc->getOutstandingRequestCount();
        if (reqCount > 0)
            Threads::sleep(100);
        else
            return true;
    }

    return false;
}

/////////////////////////////////////////////////////////////////////////////
//
// CIMListener
//
/////////////////////////////////////////////////////////////////////////////

CIMListener::CIMListener(
    Uint32 portNumber, 
    SSLContext * sslContext) 
    :
    _rep(new CIMListenerRep(portNumber, sslContext))
{
}

CIMListener::~CIMListener()
{
    if (_rep != NULL)
        delete static_cast < CIMListenerRep * >(_rep);
    _rep = NULL;
}

Uint32 CIMListener::getPortNumber() const
{
    return static_cast < CIMListenerRep * >(_rep)->getPortNumber();
}

SSLContext *CIMListener::getSSLContext() const
{
    return static_cast < CIMListenerRep * >(_rep)->getSSLContext();
}

void CIMListener::setSSLContext(SSLContext * sslContext)
{
    static_cast < CIMListenerRep * >(_rep)->setSSLContext(sslContext);
}

void CIMListener::start()
{
    static_cast < CIMListenerRep * >(_rep)->start();
}

void CIMListener::stop()
{
    static_cast < CIMListenerRep * >(_rep)->stop();
}

Boolean CIMListener::isAlive() const
{
    return static_cast < CIMListenerRep * >(_rep)->isAlive();
}

Boolean CIMListener::addConsumer(CIMIndicationConsumer * consumer)
{
    return static_cast < CIMListenerRep * >(_rep)->addConsumer(consumer);
}

Boolean CIMListener::removeConsumer(CIMIndicationConsumer * consumer)
{
    return static_cast < CIMListenerRep * >(_rep)->removeConsumer(consumer);
}

PEGASUS_NAMESPACE_END