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

---

mainDataServer.cpp

Go to the documentation of this file.

#include "Basis/CmdOptions.h"

#include "Persistency/FileSystem.h"
#include "Util/Channel.h"
#include "Util/ChannelServer.h"
#include "Util/Database.h"
#include "Util/SimpleMap.h"

#include "Core/Array/Arrays.h"

#include "Core/Stream/RgbDataSrcFactory.h"
#include "Core/Stream/RgbDataDstFactory.h"

#include "Link/DiskImage/DiskImageFuncs.h"

// since we are not using libraries:
#ifdef OGL_USED
#include "OglGui/OglLib.cpp"
#endif
#include "Link/ImpalaLib.cpp"

// cannot use log4cpp as library on sunos due to linker bug:
#ifdef LOG4CPP_USED
#ifdef sun
#define LOG4CPPHACK
#endif
#ifdef LOG4CPPHACK
#include "Link/Log4cppLib.cpp"
#endif
#endif

namespace Impala
{
namespace Application
{


using namespace Util;
using namespace Link;

/********************************************************************
 Instances of this class serve data to e.g. video processing clients.
*********************************************************************/
class DataServer : public ChannelServer
{

public:

    DataServer(int port, int nrPorts, CString passwordFile, bool useDstTypeWin) 
        : ChannelServer(port, nrPorts, true, passwordFile)
    {
        Init(useDstTypeWin);
    }

    virtual
    ~DataServer()
    {
    }

    void 
    HandleIdle()
    {
        ChannelServer::HandleIdle();

        if (mDst == 0)
            return;
        int done = 0;
        mDst->WindowManage(done, GetStatusBuffer());
    }


protected:

    virtual int
    AcceptRequest(char* buf, int len, int bufSize, CString conn, int port)
    {
        int rcBase = ChannelServer::AcceptRequest(buf, len, bufSize, conn, port);
        if (rcBase >= 0)
            return rcBase;

        String curCon = ConnectionDescr();
#ifndef REPOSITORY_USED // Here comes the deprecated stuff
        Database& db = Database::GetInstance();
#else // REPOSITORY_USED
        Persistency::FileSystem db("", "");
#endif // REPOSITORY_USED
        ILogErrors& errors = ILogErrors::GetInstance();
        errors.Mark();

        if (strncmp(buf, "domkdirpath", 11) == 0)
        {
            ILOG_DEBUG("Doing domkdirpath for " << curCon);
            int index = 11;
            String dir = ReadQuotedString(buf, index, len);
            String dataPath = ReadQuotedString(buf, index, len);
            ILOG_DEBUG("dir : [" << dir << "]");
            if (! AllowWrite(port))
            {
                ILOG_ERROR("Write is not authorized for dir " << dir);
            }
            else
            {
                db.SetDataPath(dataPath);
                String res = db.MakeDir(dir);
                ILOG_DEBUG("res : [" << res << "]");
                sprintf(buf, "%s\0", res.c_str());
            }
        }

        else if (strncmp(buf, "getreadablefile", 15) == 0)
        {
            ILOG_DEBUG("Doing getreadablefile for " << curCon);
            int index = 15;
            String fileName = ReadQuotedString(buf, index, len);
            String dataPath = ReadQuotedString(buf, index, len);
            int silent = 0;
            if (index < len)
                sscanf(buf + index, "%d", &silent);
            ILOG_DEBUG("name : [" << fileName << "], data : [" << dataPath <<
                       "], silent = " << silent);
            db.SetDataPath(dataPath);
            String res = db.GetFilePath(fileName, false, (silent != 0));
            ILOG_DEBUG("res : [" << res << "]");
            sprintf(buf, "%s\0", res.c_str());
        }

        else if (strncmp(buf, "getwritablefile", 15) == 0)
        {
            ILOG_DEBUG("Doing getwritablefile for " << curCon);
            int index = 15;
            String fileName = ReadQuotedString(buf, index, len);
            String dataPath = ReadQuotedString(buf, index, len);
            if (! AllowWrite(port))
            {
                ILOG_ERROR("Write is not authorized for " << fileName);
            }
            else
            {
                int silent = 0;
                int override = 0;
                if (index < len)
                    sscanf(buf + index, "%d %d", &silent, &override);
                ILOG_DEBUG("name : [" << fileName << "], data : [" << dataPath
                           << "], silent = " << silent << ", override = "
                           << override);
                db.SetDataPath(dataPath);
                db.SetOverride(override != 0);
                String res = db.GetFilePath(fileName, true, (silent != 0));
                ILOG_DEBUG("res : [" << res << "]");
                sprintf(buf, "%s\0", res.c_str());
            }
        }

        else if (strncmp(buf, "openfilebuffer", 14) == 0)
        {
            ILOG_DEBUG("Doing openfilebuffer for " << curCon);
            int index = 14;
            String fileName = ReadQuotedString(buf, index, len);
            int readMode = 123;
            if (index < len)
                sscanf(buf + index, "%d", &readMode);
            ILOG_DEBUG("name : [" << fileName << "], readMode = " << readMode);
            if ((! readMode) && (! AllowWrite(port)))
            {
                ILOG_ERROR("Write is not authorized for " << fileName);
            }
            else if (DiskImage::DiskImageUsed() &&
                     fileName.substr(0, 10) == "diskimage:")
            {
                int posAfterDiskImageFileName = fileName.find("://", 10);
                String realFilePath =
                    fileName.substr(10, posAfterDiskImageFileName-10);
                String filePathInImage =
                    fileName.substr(posAfterDiskImageFileName+1);
                DiskImage::FileDescriptor* desc =
                    DiskImage::OpenFile(realFilePath, filePathInImage);
                if (!desc)
                    return -1;

                sprintf(buf, "%d %lld\0", mFileBuffersId,
                        DiskImage::GetFileSize(desc));
                BufInf inf = BufInfCtor(port, mFileBuffersId++);
                mFileBuffersDiskImage.Add(inf, desc);
            }
            else
            {
                IOBufferFile* ioBuf = new IOBufferFile(fileName, readMode!=0,
                                                       false);
                if (!ioBuf->Valid())
                {
                    ILOG_ERROR("Unable to open file [" << fileName << "]");
                }
                else
                {
                    sprintf(buf, "%d %lld\0", mFileBuffersId, ioBuf->Size());
                    BufInf inf = BufInfCtor(port, mFileBuffersId++);
                    mFileBuffers.Add(inf, ioBuf);
                }
            }
        }

        else if (strncmp(buf, "readfilebuffer", 14) == 0)
        {
            ILOG_DEBUG("Doing readfilebuffer for " << curCon);
            int id;
            int nrToRead;
            sscanf(buf, "readfilebuffer %d %d", &id, &nrToRead);
            ILOG_DEBUG("id = " << id << ", nrToRead = " << nrToRead);
            nrToRead = Min(Channel::DATA_BUFFER_SIZE, nrToRead);
            BufInf inf = BufInfCtor(port, id);
            IOBufferFile* ioBuf;
            if (mFileBuffers.Get(inf, ioBuf))
            {
                int nrToSend = ioBuf->Read(buf, nrToRead);
                return nrToSend; // buffer used for binary data, so return 
            }
            if (DiskImage::DiskImageUsed())
            {
                DiskImage::FileDescriptor* desc;
                if (mFileBuffersDiskImage.Get(inf, desc))
                {
                    int bytesRead = DiskImage::ReadFile(desc, buf, nrToRead);
                    return bytesRead; // buffer used for binary data, so return 
                }
            }
            ILOG_ERROR("No file buffer " << id);
            return -1;
        }

        else if (strncmp(buf, "writefilebuffer", 15) == 0)
        {
            ILOG_DEBUG("Doing writefilebuffer for " << curCon);
            int id;
            sscanf(buf, "writefilebuffer %d", &id);
            ILOG_DEBUG("id = " << id);
            IOBufferFile* ioBuf;
            BufInf inf = BufInfCtor(port, id);
            if (!mFileBuffers.Get(inf, ioBuf))
            {
                ILOG_ERROR("No file buffer " << id);
            }
            else
            {
                int used = strlen(buf) + 1;
                ioBuf->Write(buf+used, len-used);
                sprintf(buf, "OK\0");
            }
        }

        else if (strncmp(buf, "seekfilebuffer", 14) == 0)
        {
            ILOG_DEBUG("Doing seekfilebuffer for " << curCon);
            int id;
            Int64 offset;
            int whence;
            sscanf(buf, "seekfilebuffer %d %lld %d", &id, &offset, &whence);
            ILOG_DEBUG("id = " << id << ", offset = " << offset << ", whence = "
                       << whence);
            BufInf inf = BufInfCtor(port, id);
            IOBufferFile* ioBuf;
            if (mFileBuffers.Get(inf, ioBuf))
            {
                sprintf(buf, "%lld\0", ioBuf->Seek(offset, whence));
            }
            else 
            {
                DiskImage::FileDescriptor* desc;
                if (DiskImage::DiskImageUsed() &&
                    mFileBuffersDiskImage.Get(inf, desc))
                {
                    Int64 newPosition = DiskImage::SeekFile(desc, offset,
                                                            whence);
                    sprintf(buf, "%lld\0", newPosition);
                }
                else
                {
                    ILOG_ERROR("No file buffer " << id);
                    return -1;
                }
            }
        }

        else if (strncmp(buf, "closefilebuffer", 15) == 0)
        {
            ILOG_DEBUG("Doing closefilebuffer for " << curCon);
            int id;
            sscanf(buf, "closefilebuffer %d", &id);
            ILOG_DEBUG("id = " << id);
            BufInf inf = BufInfCtor(port, id);
            IOBufferFile* ioBuf;
            if (mFileBuffers.Get(inf, ioBuf))
            {
                mFileBuffers.Remove(inf);
                sprintf(buf, "OK\0");
            }
            else
            {
                DiskImage::FileDescriptor* desc;
                if (DiskImage::DiskImageUsed() &&
                    mFileBuffersDiskImage.Get(inf, desc))
                {
                    DiskImage::CloseFile(desc);
                    mFileBuffersDiskImage.Remove(inf);
                    sprintf(buf, "OK\0");
                }
                else
                {
                    ILOG_ERROR("No file buffer " << id);
                    return -1;
                }
            }
        }

        else if (strncmp(buf, "testdata", 8) == 0)
        {
            ILOG_DEBUG("Doing testdata for " << curCon);
            int nrBytes;
            sscanf(buf, "testdata %d", &nrBytes);
            nrBytes = Min(Channel::DATA_BUFFER_SIZE, nrBytes);
            ILOG_DEBUG("nrBytes = " << nrBytes);
            static int val = 100;
            memset(buf, val, nrBytes);
            val += 10;
            if (val > 255)
                val = 0;
            return nrBytes;
        }

        else if (strncmp(buf, "opensrc", 7) == 0)
        {
            using Core::Stream::RgbDataSrcFactory;
            ILOG_DEBUG("Doing opensrc for " << curCon);
            if (mSrc)
            {
                delete mSrc;
                mSrc = 0;
            }
            int index = 7;
            String srcName = ReadQuotedString(buf, index, len);
            ILOG_DEBUG("name : [" << srcName << "]");
            int src = RgbDataSrcFactory::Instance().SuggestSrc(srcName);
#ifndef REPOSITORY_USED // Here comes the deprecated stuff
            mSrc = RgbDataSrcFactory::Instance().Construct(src, srcName);
#else // REPOSITORY_USED
            Persistency::RgbDataSrcLocator loc("file:", ".", srcName);
            mSrc = RgbDataSrcFactory::Instance().Construct(src, loc);
#endif // REPOSITORY_USED
            if ((mSrc == 0) || (! mSrc->Valid()))
            {
                ILOG_ERROR("RgbDataSrc failed");
            }
            else
            {
                sprintf(buf, "%d %d\0", mSrc->FrameWidth(),
                        mSrc->FrameHeight());
            }
        }

        else if (strncmp(buf, "nextframe", 9) == 0)
        {
            ILOG_DEBUG("Doing nextframe for " << curCon);
            if ((mSrc == 0) || (! mSrc->Valid()))
            {
                ILOG_ERROR("no RgbDataSrc");
            }
            else
            {
                mSrc->NextFrame();
                if (mSrc->DataPtr() == 0)
                {
                    ILOG_ERROR("RgbDataSrc::nextFrame has no data");
                }
                else
                {
                    int dataSize = mSrc->FrameWidth() * mSrc->FrameHeight() * 3;
                    dataSize = Min(Channel::DATA_BUFFER_SIZE, dataSize);
                    memcpy(buf, mSrc->DataPtr(), dataSize);
                    return dataSize;
                }
            }
        }

#ifdef OGL_USED
        else if (strncmp(buf, "opendst", 7) == 0)
        {
            ILOG_DEBUG("Doing opendst for " << curCon);
            if (mDst != 0)
            {
                ILOG_ERROR("dst already open");
            }
            else
            {
                String dstName("ogl");
                sscanf(buf, "opendst %d %d", &mDstWidth, &mDstHeight);
                mDst = Core::Stream::RgbDataDstFactory::Instance().Construct
                    (mDstType, dstName, mDstWidth, mDstHeight);
                if ((mDst == 0) || (! mDst->Valid()))
                {
                    ILOG_ERROR("RgbDataDst failed");
                }
                else
                {
                    mDstBuf = new unsigned char[mDstWidth * mDstHeight * 3];
                    HandleIdle();
                    sprintf(buf, "OK\0");
                }
            }
        }

        else if (strncmp(buf, "show", 4) == 0)
        {
            ILOG_DEBUG("Doing show for " << curCon);
            if ((mDst == 0) || (! mDst->Valid()))
            {
                ILOG_ERROR("no RgbDataDst");
            }
            else
            {
                mFramesDone++;
                int dataSize = mDstWidth * mDstHeight * 3;
                memcpy(mDstBuf, buf+5, dataSize);
                mDst->NextFrame(mDstBuf);
                HandleIdle();
                sprintf(buf, "OK\0");
            }
        }
#endif

#ifndef REPOSITORY_USED // Here comes the deprecated stuff
        else if (strncmp(buf, "readvideoindex", 14) == 0)
        {
            ILOG_DEBUG("Doing readvideoindex for " << curCon);
            int index = 14;
            String videoFileName = ReadQuotedString(buf, index, len);
            String dataPath = ReadQuotedString(buf, index, len);
            ILOG_DEBUG("videoFileName : [" << videoFileName << "], data : ["
                       << dataPath << "]");

            if (!dataPath.empty())
                db.SetDataPath(dataPath);

            Int64 bufferUsed = WriteVideoAsHashes(buf, videoFileName, db);
            if (bufferUsed >= 0)
                sprintf(buf + bufferUsed, "\0");
        }
#endif // REPOSITORY_USED

        else
        {
            return -1;
        }

        if (errors.GetNrErrorsSinceMark() > 0)
        {
            String last = errors.GetLastErrorSinceMark();
            sprintf(buf, "ERROR : last error at server : %s\0", last.c_str());
        }

        return strlen(buf) + 1; // use this exit for textual responses only!
    }

    virtual void
    AcceptDisconnect(int port)
    {
        std::vector<BufInf> v = mFileBuffers.GetAllIdx();
        for (int i=0 ; i<v.size() ; i++)
        {
            BufInf inf = v[i];
            int p = BufInfPort(inf);
            if (p == port)
            {
                ILOG_DEBUG("cleaning buffer " << BufInfPort(inf) << " " <<
                           BufInfBufferId(inf));
                mFileBuffers.Remove(inf);
            }
        }

        std::vector<BufInf> w = mFileBuffersDiskImage.GetAllIdx();
        for (int i=0 ; i<w.size() ; i++)
        {
            BufInf inf = w[i];
            int p = BufInfPort(inf);
            if (p == port)
            {
                ILOG_DEBUG("cleaning buffer " << BufInfPort(inf) << " " <<
                           BufInfBufferId(inf));
                mFileBuffersDiskImage.Remove(inf);
            }
        }
    }

    // overrides base member
    virtual void 
    UpdateStatusBuf()
    {
        double timeVal = SplitTime();
        double fps = (double) mFramesDone / timeVal;
        std::ostrstream statusStream;
        statusStream << "Did " << mFramesDone << " frames in " << timeVal
                     << " sec = " << fps << " fps" << std::ends;
        strcpy(GetStatusBuffer(), statusStream.str());
        statusStream.freeze(false); // prevents mem leak
    }


private:

    void 
    Init(bool useDstTypeWin)
    {
        mDstType = Core::Stream::RgbDataDstFactory::DST_OGL;
        if (useDstTypeWin)
            mDstType = Core::Stream::RgbDataDstFactory::DST_WIN;
        mFramesDone = 0;
        mSrc = 0;
        mDst = 0;
        mDstBuf = 0;
        mFileBuffersId = 1000;
    }

    typedef UInt64 BufInf;
    
    BufInf
    BufInfCtor(int port, int bufferId)
    {
        return (((BufInf) port) << 32) | ((BufInf) bufferId);
    }

    int
    BufInfPort(BufInf inf)
    {
        return (inf & 0xFFFFFFFF00000000LL) >> 32;
    }

    int
    BufInfBufferId(BufInf inf)
    {
        return (inf & 0x00000000FFFFFFFFLL);
    }

#ifndef REPOSITORY_USED // Here comes the deprecated stuff
    Int64
    WriteVideoAsHashes(char* buf, String videoFileName, Database& db)
    {
#ifdef AVC_USED
        String videoPath = db.GetFilePath(videoFileName, false, false);
        if (videoPath.empty())
            return -1;
        Core::Stream::RgbDataSrcInfo* video = 
            new Core::Stream::RgbDataSrcLavc(
            Core::Stream::RgbDataSrcFactory::SRC_LAVC, videoPath, &db, READIDX);
        if (!video)
            return -1;
        int bufUsed = video->DumpIndexHashes(buf, Channel::DATA_BUFFER_SIZE);
        if (bufUsed < 0)
            return -1;
        return bufUsed;
#else
        return -1;
#endif
    }
#endif // REPOSITORY_USED

    int mFramesDone;
    Core::Stream::RgbDataSrc* mSrc;
    Core::Stream::RgbDataDst* mDst;
    int mDstType;
    int mDstWidth;
    int mDstHeight;
    unsigned char* mDstBuf;

    SimpleMapDelete<BufInf, Util::IOBufferFile*> mFileBuffers;
    SimpleMapDelete<BufInf, DiskImage::FileDescriptor*> mFileBuffersDiskImage;
    int mFileBuffersId;

    ILOG_VAR_DECL;

}; // class

ILOG_VAR_INIT(DataServer, Impala.Application);

int
mainServer(int argc, char* argv[])
{
    CmdOptions& options = CmdOptions::GetInstance();
    options.Initialise(true, true, true);
    options.AddOption(0, "dstWin", "", "0");
    options.AddOption(0, "noIdle", "", "0");
    if (! options.ParseArgs(argc, argv, "port [nrPorts]", 1))
        return 1;

    bool useDstTypeWin = options.GetBool("dstWin");
    String passwordFile = options.GetString("passwordFile");
    int port = atol(options.GetArg(0));
    int nrPorts = 100;
    if (options.GetNrArg() > 1)
        nrPorts = atol(options.GetArg(1));

    Application::DataServer dataServer(port, nrPorts, passwordFile, useDstTypeWin);
    bool doIdle = ! options.GetBool("noIdle");
    dataServer.Start(doIdle);

    return 0;
}

} // namespace Application
} // namespace Impala

int
main(int argc, char* argv[])
{
    return Impala::Application::mainServer(argc, argv);
}

---