|| || || || || || || ||

---

ServerProxy.h

Go to the documentation of this file.

#ifndef Impala_Core_VideoJob_ServerProxy_h
#define Impala_Core_VideoJob_ServerProxy_h

#include <map>
#include <sstream>

#include "Basis/ILog.h"
#include "Util/StringParser.h"
#include "Util/ChannelProxy.h"
#include "Util/Sleep.h"
#include "Job/State.h"
#include "Core/VideoJob/Data.h"

namespace Impala
{
namespace Core
{
namespace VideoJob
{

/***********************************************************************
Instances of this class are responsible for dispatching jobs to a job
server and monitoring job status.
************************************************************************/

class ServerProxy
{
    typedef Job::State::StateType StateType;

public:

    ServerProxy(CString socketName, CString passwordFile, int requestIntervalSec)
    {
        int colonPos = socketName.find(":");
        if (colonPos <= 0)
        {
            ILOG_ERROR("Not a valid server address (port number is missing): "
                       << socketName);
            return;
        }
        Init(socketName.substr(0, colonPos),
             atol(socketName.substr(colonPos + 1)),
             passwordFile, requestIntervalSec);
    }

    ServerProxy(CString serverAddress, int port, CString passwordFile,
                int requestIntervalSec)
    {
        Init(serverAddress, port, passwordFile, requestIntervalSec);
    }

    virtual
    ~ServerProxy()
    {
        Disconnect();
    }

    bool
    IsConnected()
    {
        return (mChannelProxy && mChannelProxy->ChannelIsValid());
    }

    // returns server assigned job id if successful, zero otherwise
    int
    ScheduleJob(CString cmdLine, int realVideoNr, bool delayFirst = false)
    {
        if (!IsConnected())
            return 0;

        if (delayFirst)
        {
            ILOG_DEBUG("Delaying job dispatch");
            Util::Sleep(mRequestIntervalSec);
        }

        ILOG_INFO("Dispatching job to server (actual video nr = " <<
                  realVideoNr << ") : " << cmdLine );
        String request = "AcceptJob:cmdLine=" + cmdLine;
        String response = mChannelProxy->Send(request);
        int jobId = 0;
        if (response.substr(0, 18) == "JobAccepted:jobId=")
        {
            jobId = atoi(response.substr(18));
            if (jobId > 0)
            {
                ILOG_DEBUG("Received assigned job ID: " << jobId);
                mJobIds.push_back(jobId);
            }
            else
            {
                ILOG_ERROR("Received invalid job ID: " << jobId);
            }
        }
        else if (response == "MaxJobs")
        {
            ILOG_WARN("Job dispatch rejected: job max reached");
        }
        else
        {
            ILOG_ERROR("Job dispatch failed: " << response);
        }
            
        return jobId;
    }

    // Polls the job server for state changes of dispatched jobs and
    // returns immediately on any reported state change.
    std::map<int, StateType>& 
    MonitorJobState(Data& vpData)
    {
        static std::map<int, StateType> jobsWithStateChanged;
        jobsWithStateChanged.clear();

        std::ostringstream oss;
        std::map<int, Data::VideoJob*>::iterator jobIter =
            vpData.mJobsScheduled.begin();
        for ( ; jobIter != vpData.mJobsScheduled.end(); jobIter++ )
        {
            Data::VideoJob& job = *(jobIter->second);
            oss << job.id << ':' << job.state << ' ';
        }
        String request = "ProvideStatus:reference=" + oss.str();

        do
        {
            String response = mChannelProxy->Send(request);
            Util::StringParser parser(response);
            String responseType = parser.GetString(':', false);
            if (responseType == "NoChange")
            {
                Util::Sleep(mRequestIntervalSec);
            }
            else if (responseType == "JobStatus" &&
                     parser.GetString('=') == "changed")
            {
                int jobId = parser.GetInt(':', false, true);
                while (jobId > 0)
                {
                    StateType jobState = (StateType) parser.GetInt(' ', true,
                                                                   true);
                    jobsWithStateChanged[jobId] = jobState;
                    jobId = parser.GetInt(':', false, true);
                }
            }
            else
            {
                ILOG_ERROR("Job report request failed: " << response);
                jobsWithStateChanged.clear();
                break;
            }
        }
        while (jobsWithStateChanged.empty());

        return jobsWithStateChanged;
    }

    int
    GetJobState(int jobId, StateType& state, int& exitCode, String& errorLog)
    {
        String request = "ProvideStatus:jobId=" + MakeString(jobId);
        String response = mChannelProxy->Send(request);
        Util::StringParser parser(response);
        if (parser.GetString(':') == "JobStatus" &&
            parser.GetString('=') == "jobId" &&
            parser.GetInt(';', true, true) == jobId &&
            parser.GetString('=') == "state")
        {
            state = (StateType) parser.GetInt(';', true, true);
            if (parser.GetString('=') == "exitCode")
            {
                exitCode = parser.GetInt(';', true, true);
                if (parser.GetString('=') == "errorLog")
                    errorLog = response.substr(parser.Position());
                return 0;
            }
        }
        ILOG_ERROR("Job report request failed: " << response);
        return -1;
    }

private:

    void
    Init(CString serverAddress, int port, CString passwordFile,
         int requestIntervalSec)
    {
        mServerAddress = serverAddress;
        mPort = port;
        mPasswordFile = passwordFile;
        mRequestIntervalSec = requestIntervalSec;

        mChannelProxy = 0;

        if (! Connect(mServerAddress, mPort, mPasswordFile))
        {
            ILOG_ERROR("Cannot connect to job server at " << mServerAddress <<
                       ':' << mPort);
        }
    }

    bool
    Connect(CString serverAddress, int port, CString passwordFile,
            int timeOutSec = -1)
    {
        mChannelProxy = new Util::ChannelProxy(serverAddress, port,
                                               passwordFile);
        if (!mChannelProxy->GetChannel())
        {
            ILOG_ERROR("Failed to connect to job server");
            Disconnect();
            return false;
        }

        return true;
    }

    void
    Disconnect()
    {
        if (mChannelProxy)
        {
            //mChannelProxy->CloseChannel();
            delete mChannelProxy;
            mChannelProxy = 0;
        }
    }

    ILOG_VAR_DECL;

    String mServerAddress;
    int    mPort;
    String mPasswordFile;
    int    mRequestIntervalSec;

    Util::ChannelProxy* mChannelProxy;

    std::vector<int> mJobIds;

}; // class

ILOG_VAR_INIT(ServerProxy, Impala.Core.VideoJob);

} // namespace 
} // namespace 
} // namespace 

#endif

---