Subversion Repositories SmartDukaan

//  Copyright (c) 2007-2008 Facebook

//

//  Licensed under the Apache License, Version 2.0 (the "License");

//  you may not use this file except in compliance with the License.

//  You may obtain a copy of the License at

//

//      http://www.apache.org/licenses/LICENSE-2.0

//

//  Unless required by applicable law or agreed to in writing, software

//  distributed under the License is distributed on an "AS IS" BASIS,

//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

//  See the License for the specific language governing permissions and

//  limitations under the License.

//

// See accompanying file LICENSE or visit the Scribe site at:

// http://developers.facebook.com/scribe/

//

// @author Bobby Johnson

// @author Jason Sobel

// @author Avinash Lakshman

#include "common.h"

#include "file.h"

#include "HdfsFile.h"

// INITIAL_BUFFER_SIZE must always be >= UINT_SIZE

#define INITIAL_BUFFER_SIZE 4096

#define UINT_SIZE 4

using namespace std;

using boost::shared_ptr;

boost::shared_ptr<FileInterface> FileInterface::createFileInterface(const std::string& type,

                                                                    const std::string& name,

                                                                    bool framed) {

  if (0 == type.compare("std")) {

    return shared_ptr<FileInterface>(new StdFile(name, framed));

  } else if (0 == type.compare("hdfs")) {

    return shared_ptr<FileInterface>(new HdfsFile(name));

  } else {

    return shared_ptr<FileInterface>();

  }

}

std::vector<std::string> FileInterface::list(const std::string& path, const std::string &fsType) {

  std::vector<std::string> files;

  shared_ptr<FileInterface> concrete_file = createFileInterface(fsType, path);

  if (concrete_file) {

    concrete_file->listImpl(path, files);

  }

  return files;

}

FileInterface::FileInterface(const std::string& name, bool frame)

  : framed(frame), filename(name) {

}

FileInterface::~FileInterface() {

}

StdFile::StdFile(const std::string& name, bool frame)

  : FileInterface(name, frame), inputBuffer(NULL), bufferSize(0) {

}

StdFile::~StdFile() {

  if (inputBuffer) {

    delete[] inputBuffer;

    inputBuffer = NULL;

  }

}

bool StdFile::openRead() {

  return open(fstream::in);

}

bool StdFile::openWrite() {

  // open file for write in append mode

  ios_base::openmode mode = fstream::out | fstream::app;

  return open(mode);

}

bool StdFile::openTruncate() {

  // open an existing file for write and truncate its contents

  ios_base::openmode mode = fstream::out | fstream::app | fstream::trunc;

  return open(mode);

}

bool StdFile::open(ios_base::openmode mode) {

  if (file.is_open()) {

    return false;

  }

  file.open(filename.c_str(), mode);

  return file.good();

}

bool StdFile::isOpen() {

  return file.is_open();

}

void StdFile::close() {

  if (file.is_open()) {

    file.close();

  }

}

string StdFile::getFrame(unsigned data_length) {

  if (framed) {

    char buf[UINT_SIZE];

    serializeUInt(data_length, buf);

    return string(buf, UINT_SIZE);

  } else {

    return string();

  }

}

bool StdFile::write(const std::string& data) {

  if (!file.is_open()) {

    return false;

  }

  file << data;

  if (file.bad()) {

    return false;

  }

  return true;

}

void StdFile::flush() {

  if (file.is_open()) {

    file.flush();

  }

}

bool StdFile::readNext(std::string& _return) {

  if (!inputBuffer) {

    bufferSize = INITIAL_BUFFER_SIZE;

    inputBuffer = new char[bufferSize];

  }

  if (framed) {

    unsigned size;

    file.read(inputBuffer, UINT_SIZE);  // assumes INITIAL_BUFFER_SIZE > UINT_SIZE

    if (file.good() && (size = unserializeUInt(inputBuffer))) {

      // check if size is larger than half the max uint size

      if (size >= (((unsigned)1) << (UINT_SIZE*8 - 1))) {

        LOG_OPER("WARNING: attempting to read message of size %d bytes", size);

        // Do not try to make bufferSize any larger than this or you might overflow

        bufferSize = size;

      }

      while (size > bufferSize) {

        bufferSize = 2 * bufferSize;

        delete[] inputBuffer;

        inputBuffer = new char[bufferSize];

      }

      file.read(inputBuffer, size);

      if (file.good()) {

        _return.assign(inputBuffer, size);

        return true;

      } else {

        int offset = file.tellg();

        LOG_OPER("ERROR: Failed to read file %s at offset %d",

                 filename.c_str(), offset);

        return false;

      }

    }

  } else {

    file.getline(inputBuffer, bufferSize);

    if (file.good()) {

      _return = inputBuffer;

      return true;

    }

  }

  return false;

}

unsigned long StdFile::fileSize() {

  unsigned long size = 0;

  try {

    size = boost::filesystem::file_size(filename.c_str());

  } catch(std::exception const& e) {

    LOG_OPER("Failed to get size for file <%s> error <%s>", filename.c_str(), e.what());

    size = 0;

  }

  return size;

}

void StdFile::listImpl(const std::string& path, std::vector<std::string>& _return) {

  try {

    if (boost::filesystem::exists(path)) {

      boost::filesystem::directory_iterator dir_iter(path), end_iter;

      for ( ; dir_iter != end_iter; ++dir_iter) {

        _return.push_back(dir_iter->filename());

      }

    }

  } catch (std::exception const& e) {

    LOG_OPER("exception <%s> listing files in <%s>",

             e.what(), path.c_str());

  }

}

void StdFile::deleteFile() {

  boost::filesystem::remove(filename);

}

bool StdFile::createDirectory(std::string path) {

  try {

    boost::filesystem::create_directories(path);

  } catch(std::exception const& e) {

    LOG_OPER("Exception < %s > in StdFile::createDirectory for path %s ",

      e.what(),path.c_str());

    return false;

  }

  return true;

}

bool StdFile::createSymlink(std::string oldpath, std::string newpath) {

  if (symlink(oldpath.c_str(), newpath.c_str()) == 0) {

    return true;

  }

  return false;

}

// Buffer had better be at least UINT_SIZE long!

unsigned FileInterface::unserializeUInt(const char* buffer) {

  unsigned retval = 0;

  int i;

  for (i = 0; i < UINT_SIZE; ++i) {

    retval |= (unsigned char)buffer[i] << (8 * i);

  }

  return retval;

}

void FileInterface::serializeUInt(unsigned data, char* buffer) {

  int i;

  for (i = 0; i < UINT_SIZE; ++i) {

    buffer[i] = (unsigned char)((data >> (8 * i)) & 0xFF);

  }

}