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

---

ReadIOBufferFromChannel.h

Go to the documentation of this file.

#ifndef Impala_Util_ReadIOBufferFromChannel_h
#define Impala_Util_ReadIOBufferFromChannel_h

#include "Util/IOBufferFile.h"

namespace Impala
{
namespace Util
{


inline IOBuffer*
ReadIOBufferFromChannel(Channel* channel, std::string fileName,
                        std::string useLocalFile)
{
    ILOG_VAR(Impala.Util.ReadIOBufferFromChannel);
    char* buf = channel->Buffer();
    sprintf(buf, "openfilebuffer \"%s\" 1\0", fileName.c_str());
    channel->SendRequest(strlen(buf)+1);
    if (channel->LastSendHadError())
        return 0;
    int id = 0;
    Int64 bufSize = 0;
    sscanf(buf, "%d %lld", &id, &bufSize);
    ILOG_DEBUG(fileName << ", id=" << id << ", size=" << bufSize << " from "
               << channel->GetServerInfo());

    IOBuffer* ioBuf = 0;
    if (useLocalFile.empty())
        ioBuf = new IOBuffer(bufSize);
    else
        ioBuf = new IOBufferFile(useLocalFile, false, false);
    Int64 nrRead = 0;
    while (nrRead < bufSize)
    {
        sprintf(buf, "readfilebuffer %d %d\0", id, Channel::DATA_BUFFER_SIZE);
        int len = channel->SendRequest(strlen(buf)+1);
        if (channel->LastSendHadError())
            break;
        ioBuf->Write(buf, len);
        nrRead += len;
        ILOG_DEBUG("  wrote " << len << " bytes, total = " << nrRead);
    }
    sprintf(buf, "closefilebuffer %d\0", id);
    channel->SendRequest(strlen(buf)+1);
    if (useLocalFile.empty())
    {
        ioBuf->SetPosition(0); // to allow for subsequent reading by client
    }
    else
    {
        delete ioBuf;
        ioBuf = new IOBufferFile(useLocalFile, true, false);
    }
    return ioBuf;
}

} // namespace Util
} // namespace Impala

#endif

---