Permalink
Browse files

Stream operations: file access, iteration, splitting.

Summary: Intended to complement and replace strings::byLine.

Test Plan: stream_test

Reviewed By: delong.j@fb.com

FB internal diff: D463341
  • Loading branch information...
1 parent 39786ac commit eed46f426a62a15ffae8a2e73ca48617754f0a76 @tudor tudor committed with jdelong May 2, 2012
View
@@ -187,7 +187,8 @@ class Range : private boost::totally_ordered<Range<Iter> > {
// Allow implicit conversion from Range<const char*> (aka StringPiece) to
// Range<const unsigned char*> (aka ByteRange), as they're both frequently
- // used to represent ranges of bytes.
+ // used to represent ranges of bytes. Allow explicit conversion in the other
+ // direction.
template <class OtherIter, typename std::enable_if<
(std::is_same<Iter, const unsigned char*>::value &&
std::is_same<OtherIter, const char*>::value), int>::type = 0>
@@ -196,6 +197,14 @@ class Range : private boost::totally_ordered<Range<Iter> > {
e_(reinterpret_cast<const unsigned char*>(other.end())) {
}
+ template <class OtherIter, typename std::enable_if<
+ (std::is_same<Iter, const char*>::value &&
+ std::is_same<OtherIter, const unsigned char*>::value), int>::type = 0>
+ explicit Range(const Range<OtherIter>& other)
+ : b_(reinterpret_cast<const char*>(other.begin())),
+ e_(reinterpret_cast<const char*>(other.end())) {
+ }
+
void clear() {
b_ = Iter();
e_ = Iter();
@@ -0,0 +1,98 @@
+/*
+ * Copyright 2012 Facebook, Inc.
+ *
+ * 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.
+ */
+
+#ifndef FOLLY_IO_STREAM_H_
+#error This file may only be included from Stream.h
+#endif
+
+#include <string.h>
+
+#include <glog/logging.h>
+
+namespace folly {
+
+template <class Stream>
+InputByteStreamSplitter<Stream>::InputByteStreamSplitter(
+ char delimiter, Stream stream)
+ : done_(false),
+ delimiter_(delimiter),
+ stream_(std::move(stream)) {
+}
+
+template <class Stream>
+bool InputByteStreamSplitter<Stream>::operator()(ByteRange& chunk) {
+ DCHECK_EQ(buffer_->length(), 0);
+ chunk.clear();
+ if (rest_.empty()) {
+ if (done_) {
+ return false;
+ } else if (!stream_(rest_)) {
+ done_ = true;
+ return false;
+ }
+ }
+
+ auto p = static_cast<const unsigned char*>(memchr(rest_.data(), delimiter_,
+ rest_.size()));
+ if (p) {
+ chunk.assign(rest_.data(), p);
+ rest_.assign(p + 1, rest_.end());
+ return true;
+ }
+
+ // Incomplete line read, copy to buffer
+ if (!buffer_) {
+ static const size_t kDefaultLineSize = 256;
+ // Arbitrarily assume that we have half of a line in rest_, and
+ // get enough room for twice that.
+ buffer_ = IOBuf::create(std::max(kDefaultLineSize, 2 * rest_.size()));
+ } else {
+ buffer_->reserve(0, rest_.size());
+ }
+ memcpy(buffer_->writableTail(), rest_.data(), rest_.size());
+ buffer_->append(rest_.size());
+
+ while (stream_(rest_)) {
+ auto p = static_cast<const unsigned char*>(
+ memchr(rest_.data(), delimiter_, rest_.size()));
+ if (p) {
+ // Copy everything up to the delimiter and return it
+ size_t n = p - rest_.data();
+ buffer_->reserve(0, n);
+ memcpy(buffer_->writableTail(), rest_.data(), n);
+ buffer_->append(n);
+ chunk.reset(buffer_->data(), buffer_->length());
+ buffer_->trimStart(buffer_->length());
+ rest_.assign(p + 1, rest_.end());
+ return true;
+ }
+
+ // Nope, copy the entire chunk that we read
+ buffer_->reserve(0, rest_.size());
+ memcpy(buffer_->writableTail(), rest_.data(), rest_.size());
+ buffer_->append(rest_.size());
+ }
+
+ // Incomplete last line
+ done_ = true;
+ rest_.clear();
+ chunk.reset(buffer_->data(), buffer_->length());
+ buffer_->trimStart(buffer_->length());
+ return true;
+}
+
+} // namespace folly
+
@@ -0,0 +1,98 @@
+/*
+ * Copyright 2012 Facebook, Inc.
+ *
+ * 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.
+ */
+
+#include "folly/experimental/io/Stream.h"
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+
+#include <stdexcept>
+#include <system_error>
+
+#include "folly/String.h"
+
+namespace folly {
+
+FileInputByteStream::FileInputByteStream(int fd, bool ownsFd, size_t bufferSize)
+ : fd_(fd),
+ ownsFd_(ownsFd),
+ buffer_(IOBuf::create(bufferSize)) {
+}
+
+FileInputByteStream::FileInputByteStream(int fd, bool ownsFd,
+ std::unique_ptr<IOBuf>&& buffer)
+ : fd_(fd),
+ ownsFd_(ownsFd),
+ buffer_(std::move(buffer)) {
+ buffer_->clear();
+}
+
+bool FileInputByteStream::operator()(ByteRange& chunk) {
+ ssize_t n = ::read(fd_, buffer_->writableTail(), buffer_->capacity());
+ if (n == -1) {
+ throw std::system_error(errno, std::system_category(), "read failed");
+ }
+ chunk.reset(buffer_->tail(), n);
+ return (n != 0);
+}
+
+FileInputByteStream::FileInputByteStream(FileInputByteStream&& other)
+ : fd_(other.fd_),
+ ownsFd_(other.ownsFd_),
+ buffer_(std::move(other.buffer_)) {
+ other.fd_ = -1;
+ other.ownsFd_ = false;
+}
+
+FileInputByteStream& FileInputByteStream::operator=(
+ FileInputByteStream&& other) {
+ if (&other != this) {
+ closeNoThrow();
+ fd_ = other.fd_;
+ ownsFd_ = other.ownsFd_;
+ buffer_ = std::move(other.buffer_);
+ other.fd_ = -1;
+ other.ownsFd_ = false;
+ }
+ return *this;
+}
+
+FileInputByteStream::~FileInputByteStream() {
+ closeNoThrow();
+}
+
+void FileInputByteStream::closeNoThrow() {
+ if (!ownsFd_) {
+ return;
+ }
+ ownsFd_ = false;
+ if (::close(fd_) == -1) {
+ PLOG(ERROR) << "close failed";
+ }
+}
+
+InputByteStreamSplitter<FileInputByteStream> byLine(
+ const char* fileName, char delim) {
+ int fd = ::open(fileName, O_RDONLY);
+ if (fd == -1) {
+ throw std::system_error(errno, std::system_category(), "open failed");
+ }
+ return makeInputByteStreamSplitter(delim, FileInputByteStream(fd, true));
+}
+
+} // namespace folly
+
Oops, something went wrong.

0 comments on commit eed46f4

Please sign in to comment.