You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@brpc.apache.org by ja...@apache.org on 2019/05/14 09:27:46 UTC
[incubator-brpc] 06/06: Add reader and writer for binary records
This is an automated email from the ASF dual-hosted git repository.
jamesge pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-brpc.git
commit a68868077ca830853de952db7e488cd93aac0ea4
Author: gejun <ge...@bilibili.com>
AuthorDate: Tue May 14 17:26:23 2019 +0800
Add reader and writer for binary records
---
Makefile | 1 +
src/butil/recordio.cc | 359 +++++++++++++++++++++++++++++++++++++++++++++
src/butil/recordio.h | 131 +++++++++++++++++
test/Makefile | 1 +
test/recordio_unittest.cpp | 254 ++++++++++++++++++++++++++++++++
5 files changed, 746 insertions(+)
diff --git a/Makefile b/Makefile
index 21e3081..cdd1ff5 100644
--- a/Makefile
+++ b/Makefile
@@ -146,6 +146,7 @@ BUTIL_SOURCES = \
src/butil/containers/case_ignored_flat_map.cpp \
src/butil/iobuf.cpp \
src/butil/binary_printer.cpp \
+ src/butil/recordio.cc \
src/butil/popen.cpp
ifeq ($(SYSTEM), Linux)
diff --git a/src/butil/recordio.cc b/src/butil/recordio.cc
new file mode 100755
index 0000000..f4e2b19
--- /dev/null
+++ b/src/butil/recordio.cc
@@ -0,0 +1,359 @@
+#include <gflags/gflags.h>
+#include "butil/logging.h"
+#include "butil/recordio.h"
+#include "butil/sys_byteorder.h"
+
+namespace butil {
+
+DEFINE_int64(recordio_max_record_size, 67108864,
+ "Records exceeding this size will be rejected");
+
+#define BRPC_RECORDIO_MAGIC "RDIO"
+
+const size_t MAX_NAME_SIZE = 256;
+
+// 8-bit CRC using the polynomial x^8+x^6+x^3+x^2+1, 0x14D.
+// Chosen based on Koopman, et al. (0xA6 in his notation = 0x14D >> 1):
+// http://www.ece.cmu.edu/~koopman/roses/dsn04/koopman04_crc_poly_embedded.pdf
+//
+// This implementation is reflected, processing the least-significant bit of the
+// input first, has an initial CRC register value of 0xff, and exclusive-or's
+// the final register value with 0xff. As a result the CRC of an empty string,
+// and therefore the initial CRC value, is zero.
+//
+// The standard description of this CRC is:
+// width=8 poly=0x4d init=0xff refin=true refout=true xorout=0xff check=0xd8
+// name="CRC-8/KOOP"
+static unsigned char const crc8_table[] = {
+ 0xea, 0xd4, 0x96, 0xa8, 0x12, 0x2c, 0x6e, 0x50, 0x7f, 0x41, 0x03, 0x3d,
+ 0x87, 0xb9, 0xfb, 0xc5, 0xa5, 0x9b, 0xd9, 0xe7, 0x5d, 0x63, 0x21, 0x1f,
+ 0x30, 0x0e, 0x4c, 0x72, 0xc8, 0xf6, 0xb4, 0x8a, 0x74, 0x4a, 0x08, 0x36,
+ 0x8c, 0xb2, 0xf0, 0xce, 0xe1, 0xdf, 0x9d, 0xa3, 0x19, 0x27, 0x65, 0x5b,
+ 0x3b, 0x05, 0x47, 0x79, 0xc3, 0xfd, 0xbf, 0x81, 0xae, 0x90, 0xd2, 0xec,
+ 0x56, 0x68, 0x2a, 0x14, 0xb3, 0x8d, 0xcf, 0xf1, 0x4b, 0x75, 0x37, 0x09,
+ 0x26, 0x18, 0x5a, 0x64, 0xde, 0xe0, 0xa2, 0x9c, 0xfc, 0xc2, 0x80, 0xbe,
+ 0x04, 0x3a, 0x78, 0x46, 0x69, 0x57, 0x15, 0x2b, 0x91, 0xaf, 0xed, 0xd3,
+ 0x2d, 0x13, 0x51, 0x6f, 0xd5, 0xeb, 0xa9, 0x97, 0xb8, 0x86, 0xc4, 0xfa,
+ 0x40, 0x7e, 0x3c, 0x02, 0x62, 0x5c, 0x1e, 0x20, 0x9a, 0xa4, 0xe6, 0xd8,
+ 0xf7, 0xc9, 0x8b, 0xb5, 0x0f, 0x31, 0x73, 0x4d, 0x58, 0x66, 0x24, 0x1a,
+ 0xa0, 0x9e, 0xdc, 0xe2, 0xcd, 0xf3, 0xb1, 0x8f, 0x35, 0x0b, 0x49, 0x77,
+ 0x17, 0x29, 0x6b, 0x55, 0xef, 0xd1, 0x93, 0xad, 0x82, 0xbc, 0xfe, 0xc0,
+ 0x7a, 0x44, 0x06, 0x38, 0xc6, 0xf8, 0xba, 0x84, 0x3e, 0x00, 0x42, 0x7c,
+ 0x53, 0x6d, 0x2f, 0x11, 0xab, 0x95, 0xd7, 0xe9, 0x89, 0xb7, 0xf5, 0xcb,
+ 0x71, 0x4f, 0x0d, 0x33, 0x1c, 0x22, 0x60, 0x5e, 0xe4, 0xda, 0x98, 0xa6,
+ 0x01, 0x3f, 0x7d, 0x43, 0xf9, 0xc7, 0x85, 0xbb, 0x94, 0xaa, 0xe8, 0xd6,
+ 0x6c, 0x52, 0x10, 0x2e, 0x4e, 0x70, 0x32, 0x0c, 0xb6, 0x88, 0xca, 0xf4,
+ 0xdb, 0xe5, 0xa7, 0x99, 0x23, 0x1d, 0x5f, 0x61, 0x9f, 0xa1, 0xe3, 0xdd,
+ 0x67, 0x59, 0x1b, 0x25, 0x0a, 0x34, 0x76, 0x48, 0xf2, 0xcc, 0x8e, 0xb0,
+ 0xd0, 0xee, 0xac, 0x92, 0x28, 0x16, 0x54, 0x6a, 0x45, 0x7b, 0x39, 0x07,
+ 0xbd, 0x83, 0xc1, 0xff
+};
+
+static uint8_t SizeChecksum(uint32_t input) {
+ uint8_t crc = 0;
+ crc = crc8_table[crc ^ (input & 0xFF)];
+ crc = crc8_table[crc ^ ((input >> 8) & 0xFF)];
+ crc = crc8_table[crc ^ ((input >> 16) & 0xFF)];
+ crc = crc8_table[crc ^ ((input >> 24) & 0xFF)];
+ return crc;
+}
+
+const butil::IOBuf* Record::Meta(const char* name) const {
+ for (size_t i = 0; i < _metas.size(); ++i) {
+ if (_metas[i].name == name) {
+ return _metas[i].data.get();
+ }
+ }
+ return NULL;
+}
+
+butil::IOBuf* Record::MutableMeta(const char* name_cstr, bool null_on_found) {
+ const butil::StringPiece name = name_cstr;
+ for (size_t i = 0; i < _metas.size(); ++i) {
+ if (_metas[i].name == name) {
+ return null_on_found ? NULL : _metas[i].data.get();
+ }
+ }
+ if (name.size() > MAX_NAME_SIZE) {
+ LOG(ERROR) << "Too long name=" << name;
+ return NULL;
+ } else if (name.empty()) {
+ LOG(ERROR) << "Empty name";
+ return NULL;
+ }
+ NamedMeta p;
+ name.CopyToString(&p.name);
+ p.data = std::make_shared<butil::IOBuf>();
+ _metas.push_back(p);
+ return p.data.get();
+}
+
+butil::IOBuf* Record::MutableMeta(const std::string& name, bool null_on_found) {
+ for (size_t i = 0; i < _metas.size(); ++i) {
+ if (_metas[i].name == name) {
+ return null_on_found ? NULL : _metas[i].data.get();
+ }
+ }
+ if (name.size() > MAX_NAME_SIZE) {
+ LOG(ERROR) << "Too long name" << name;
+ return NULL;
+ } else if (name.empty()) {
+ LOG(ERROR) << "Empty name";
+ return NULL;
+ }
+ NamedMeta p;
+ p.name = name;
+ p.data = std::make_shared<butil::IOBuf>();
+ _metas.push_back(p);
+ return p.data.get();
+}
+
+bool Record::RemoveMeta(const butil::StringPiece& name) {
+ for (size_t i = 0; i < _metas.size(); ++i) {
+ if (_metas[i].name == name) {
+ _metas[i] = _metas.back();
+ _metas.pop_back();
+ return true;
+ }
+ }
+ return false;
+}
+
+void Record::Clear() {
+ _payload.clear();
+ _metas.clear();
+}
+
+size_t Record::ByteSize() const {
+ size_t n = 9 + _payload.size();
+ for (size_t i = 0; i < _metas.size(); ++i) {
+ const NamedMeta& m = _metas[i];
+ n += 5 + m.name.size() + m.data->size();
+ }
+ return n;
+}
+
+RecordReader::RecordReader(IReader* reader)
+ : _reader(reader)
+ , _cutter(&_portal)
+ , _ncut(0)
+ , _last_error(0) {
+}
+
+bool RecordReader::ReadNext(Record* out) {
+ const size_t MAX_READ = 1024 * 1024;
+ do {
+ const int rc = CutRecord(out);
+ if (rc > 0) {
+ _last_error = 0;
+ return true;
+ } else if (rc < 0) {
+ while (!CutUntilNextRecordCandidate()) {
+ const ssize_t nr = _portal.append_from_reader(_reader, MAX_READ);
+ if (nr <= 0) {
+ _last_error = (nr < 0 ? errno : END_OF_READER);
+ return false;
+ }
+ }
+ } else { // rc == 0, not enough data to parse
+ const ssize_t nr = _portal.append_from_reader(_reader, MAX_READ);
+ if (nr <= 0) {
+ _last_error = (nr < 0 ? errno : END_OF_READER);
+ return false;
+ }
+ }
+ } while (true);
+}
+
+bool RecordReader::CutUntilNextRecordCandidate() {
+ const size_t old_ncut = _ncut;
+ // Skip beginning magic
+ char magic[4];
+ if (_cutter.copy_to(magic, sizeof(magic)) != sizeof(magic)) {
+ return false;
+ }
+ if (*(const uint32_t*)magic == *(const uint32_t*)BRPC_RECORDIO_MAGIC) {
+ _cutter.pop_front(sizeof(magic));
+ _ncut += sizeof(magic);
+ }
+ char buf[512];
+ do {
+ const size_t nc = _cutter.copy_to(buf, sizeof(buf));
+ if (nc < sizeof(magic)) {
+ return false;
+ }
+ const size_t m = nc + 1 - sizeof(magic);
+ for (size_t i = 0; i < m; ++i) {
+ if (*(const uint32_t*)(buf + i) == *(const uint32_t*)BRPC_RECORDIO_MAGIC) {
+ _cutter.pop_front(i);
+ _ncut += i;
+ LOG(INFO) << "Found record candidate after " << _ncut - old_ncut << " bytes";
+ return true;
+ }
+ }
+ _cutter.pop_front(m);
+ _ncut += m;
+ if (nc < sizeof(buf)) {
+ return false;
+ }
+ } while (true);
+}
+
+int RecordReader::CutRecord(Record* rec) {
+ uint8_t headbuf[9];
+ if (_cutter.copy_to(headbuf, sizeof(headbuf)) != sizeof(headbuf)) {
+ return 0;
+ }
+ if (*(const uint32_t*)headbuf != *(const uint32_t*)BRPC_RECORDIO_MAGIC) {
+ LOG(ERROR) << "Invalid magic_num="
+ << butil::PrintedAsBinary(std::string((char*)headbuf, 4))
+ << ", offset=" << read_bytes();
+ return -1;
+ }
+ uint32_t tmp = NetToHost32(*(const uint32_t*)(headbuf + 4));
+ const uint8_t checksum = SizeChecksum(tmp);
+ bool has_meta = (tmp & 0x80000000);
+ // NOTE: use size_t rather than uint32_t for sizes to avoid potential
+ // addition overflows
+ const size_t data_size = (tmp & 0x7FFFFFFF);
+ if (checksum != headbuf[8]) {
+ LOG(ERROR) << "Unmatched checksum of 0x"
+ << std::hex << tmp << std::dec
+ << "(metabit=" << has_meta
+ << " size=" << data_size
+ << " offset=" << read_bytes()
+ << "), expected=" << (unsigned)headbuf[8]
+ << " actual=" << (unsigned)checksum;
+ return -1;
+ }
+ if (data_size > (size_t)FLAGS_recordio_max_record_size) {
+ LOG(ERROR) << "data_size=" << data_size
+ << " is larger than -recordio_max_record_size="
+ << FLAGS_recordio_max_record_size
+ << ", offset=" << read_bytes();
+ return -1;
+ }
+ if (_cutter.remaining_bytes() < data_size) {
+ return 0;
+ }
+ rec->Clear();
+ _cutter.pop_front(sizeof(headbuf));
+ _ncut += sizeof(headbuf);
+ size_t consumed_bytes = 0;
+ while (has_meta) {
+ char name_size_buf = 0;
+ CHECK(_cutter.cut1(&name_size_buf));
+ const size_t name_size = (uint8_t)name_size_buf;
+ std::string name;
+ _cutter.cutn(&name, name_size);
+ _cutter.cutn(&tmp, 4);
+ tmp = NetToHost32(tmp);
+ has_meta = (tmp & 0x80000000);
+ const size_t meta_size = (tmp & 0x7FFFFFFF);
+ _ncut += 5 + name_size;
+ if (consumed_bytes + 5 + name_size + meta_size > data_size) {
+ LOG(ERROR) << name << ".meta_size=" << meta_size
+ << " is inconsistent with its data_size=" << data_size
+ << ", offset=" << read_bytes();
+ return -1;
+ }
+ butil::IOBuf* meta = rec->MutableMeta(name, true/*null_on_found*/);
+ if (meta == NULL) {
+ LOG(ERROR) << "Fail to add meta=" << name
+ << ", offset=" << read_bytes();
+ return -1;
+ }
+ _cutter.cutn(meta, meta_size);
+ _ncut += meta_size;
+ consumed_bytes += 5 + name_size + meta_size;
+ }
+ _cutter.cutn(rec->MutablePayload(), data_size - consumed_bytes);
+ _ncut += data_size - consumed_bytes;
+ return 1;
+}
+
+RecordWriter::RecordWriter(IWriter* writer)
+ :_writer(writer) {
+}
+
+int RecordWriter::WriteWithoutFlush(const Record& rec) {
+ const size_t old_size = _buf.size();
+ uint8_t headbuf[9];
+ const IOBuf::Area headarea = _buf.reserve(sizeof(headbuf));
+ for (size_t i = 0; i < rec.MetaCount(); ++i) {
+ auto& s = rec.MetaAt(i);
+ if (s.name.size() > MAX_NAME_SIZE) {
+ LOG(ERROR) << "Too long name=" << s.name;
+ _buf.pop_back(_buf.size() - old_size);
+ return -1;
+ }
+ char metabuf[s.name.size() + 5];
+ char* p = metabuf;
+ *p = s.name.size();
+ ++p;
+ memcpy(p, s.name.data(), s.name.size());
+ p += s.name.size();
+ if (s.data->size() > 0x7FFFFFFFULL) {
+ LOG(ERROR) << "Meta named `" << s.name << "' is too long, size="
+ << s.data->size();
+ _buf.pop_back(_buf.size() - old_size);
+ return -1;
+ }
+ uint32_t tmp = s.data->size() & 0x7FFFFFFF;
+ if (i < rec.MetaCount() - 1) {
+ tmp |= 0x80000000;
+ }
+ *(uint32_t*)p = HostToNet32(tmp);
+ _buf.append(metabuf, sizeof(metabuf));
+ _buf.append(*s.data.get());
+ }
+ if (!rec.Payload().empty()) {
+ _buf.append(rec.Payload());
+ }
+ *(uint32_t*)headbuf = *(const uint32_t*)BRPC_RECORDIO_MAGIC;
+ const size_t data_size = _buf.size() - old_size - sizeof(headbuf);
+ if (data_size > 0x7FFFFFFFULL) {
+ LOG(ERROR) << "data_size=" << data_size << " is too long";
+ _buf.pop_back(_buf.size() - old_size);
+ return -1;
+ }
+ uint32_t tmp = (data_size & 0x7FFFFFFF);
+ if (rec.MetaCount() > 0) {
+ tmp |= 0x80000000;
+ }
+ *(uint32_t*)(headbuf + 4) = HostToNet32(tmp);
+ headbuf[8] = SizeChecksum(tmp);
+ _buf.unsafe_assign(headarea, headbuf);
+ return 0;
+}
+
+int RecordWriter::Flush() {
+ size_t total_nw = 0;
+ do {
+ const ssize_t nw = _buf.cut_into_writer(_writer);
+ if (nw > 0) {
+ total_nw += nw;
+ } else {
+ if (total_nw) {
+ // We've flushed sth., return as success.
+ return 0;
+ }
+ if (nw == 0) {
+ return _buf.empty() ? 0 : EAGAIN;
+ } else {
+ return errno;
+ }
+ }
+ } while (true);
+}
+
+int RecordWriter::Write(const Record& record) {
+ const int rc = WriteWithoutFlush(record);
+ if (rc) {
+ return rc;
+ }
+ return Flush();
+}
+
+
+} // namespace butil
diff --git a/src/butil/recordio.h b/src/butil/recordio.h
new file mode 100755
index 0000000..e1a711c
--- /dev/null
+++ b/src/butil/recordio.h
@@ -0,0 +1,131 @@
+// recordio - A binary format to transport data from end to end.
+//
+// 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.
+
+// Author: Ge,Jun (jge666@gmail.com)
+// Date: Thu Nov 22 13:57:56 CST 2012
+
+#ifndef BUTIL_RECORDIO_H
+#define BUTIL_RECORDIO_H
+
+#include "butil/iobuf.h"
+#include <memory>
+
+namespace butil {
+
+class Record {
+public:
+ struct NamedMeta {
+ std::string name;
+ std::shared_ptr<butil::IOBuf> data;
+ };
+
+ // Number of meta
+ size_t MetaCount() const { return _metas.size(); }
+
+ // Get i-th Meta, out-of-range accesses may crash.
+ const NamedMeta& MetaAt(size_t i) const { return _metas[i]; }
+
+ // Get meta by name.
+ const butil::IOBuf* Meta(const char* name) const;
+
+ // Add meta.
+ // Returns a modifiable pointer to the meta with the name.
+ // If null_on_found is true and meta with the name is present, NULL is returned.
+ butil::IOBuf* MutableMeta(const char* name, bool null_on_found = false);
+ butil::IOBuf* MutableMeta(const std::string& name, bool null_on_found = false);
+
+ // Remove meta with the name.
+ // Returns true on erased.
+ bool RemoveMeta(const butil::StringPiece& name);
+
+ // Get the payload.
+ const butil::IOBuf& Payload() const { return _payload; }
+
+ // Get a modifiable pointer to the payload.
+ butil::IOBuf* MutablePayload() { return &_payload; }
+
+ // Clear payload and remove all meta.
+ void Clear();
+
+ // Byte size of serialized form of this record.
+ size_t ByteSize() const;
+
+private:
+ butil::IOBuf _payload;
+ std::vector<NamedMeta> _metas;
+};
+
+// Parse records from the IReader, corrupted records will be skipped.
+// Example:
+// RecordReader rd(ireader);
+// Record rec;
+// while (rd.ReadNext(&rec)) {
+// HandleRecord(rec);
+// }
+// if (rd.last_error() != RecordReader::END_OF_READER) {
+// LOG(FATAL) << "Critical error occurred";
+// }
+class RecordReader {
+public:
+ static const int END_OF_READER = -1;
+
+ explicit RecordReader(IReader* reader);
+
+ // Returns true on success and `out' is overwritten by the record.
+ // False otherwise, check last_error() for the error which is treated as permanent.
+ bool ReadNext(Record* out);
+
+ // 0 means no error.
+ // END_OF_READER means all bytes in the input reader are consumed and
+ // turned into records.
+ int last_error() const { return _last_error; }
+
+ // Total bytes of all read records.
+ size_t read_bytes() const { return _ncut; }
+
+private:
+ bool CutUntilNextRecordCandidate();
+ int CutRecord(Record* rec);
+
+private:
+ IReader* _reader;
+ IOPortal _portal;
+ IOBufCutter _cutter;
+ size_t _ncut;
+ int _last_error;
+};
+
+// Write records into the IWriter.
+class RecordWriter {
+public:
+ explicit RecordWriter(IWriter* writer);
+
+ // Serialize the record into internal buffer and NOT flush into the IWriter.
+ int WriteWithoutFlush(const Record&);
+
+ // Serialize the record into internal buffer and flush into the IWriter.
+ int Write(const Record&);
+
+ // Flush internal buffer into the IWriter.
+ // Returns 0 on success, error code otherwise.
+ int Flush();
+
+private:
+ IOBuf _buf;
+ IWriter* _writer;
+};
+
+} // namespace butil
+
+#endif // BUTIL_RECORDIO_H
diff --git a/test/Makefile b/test/Makefile
index 969535a..db4d7d8 100644
--- a/test/Makefile
+++ b/test/Makefile
@@ -110,6 +110,7 @@ TEST_BUTIL_SOURCES = \
flat_map_unittest.cpp \
crc32c_unittest.cc \
iobuf_unittest.cpp \
+ recordio_unittest.cpp \
test_switches.cc \
scoped_locale.cc \
popen_unittest.cpp \
diff --git a/test/recordio_unittest.cpp b/test/recordio_unittest.cpp
new file mode 100755
index 0000000..d4a2c42
--- /dev/null
+++ b/test/recordio_unittest.cpp
@@ -0,0 +1,254 @@
+#include <gtest/gtest.h>
+#include "butil/recordio.h"
+#include "butil/fast_rand.h"
+#include "butil/string_printf.h"
+#include "butil/file_util.h"
+
+namespace {
+
+class StringReader : public butil::IReader {
+public:
+ StringReader(const std::string& str)
+ : _str(str), _offset(0) {}
+
+ ssize_t ReadV(const iovec* iov, int iovcnt) override {
+ size_t total_nc = 0;
+ for (int i = 0; i < iovcnt; ++i) {
+ void* dst = iov[i].iov_base;
+ size_t len = iov[i].iov_len;
+ size_t remain = _str.size() - _offset;
+ size_t nc = std::min(len, remain);
+ memcpy(dst, _str.data() + _offset, nc);
+ _offset += nc;
+ total_nc += nc;
+ if (_offset == _str.size()) {
+ break;
+ }
+ }
+ return total_nc;
+ }
+private:
+ std::string _str;
+ size_t _offset;
+};
+
+class StringWriter : public butil::IWriter {
+public:
+ ssize_t WriteV(const iovec* iov, int iovcnt) override {
+ const size_t old_size = _str.size();
+ for (int i = 0; i < iovcnt; ++i) {
+ _str.append((char*)iov[i].iov_base, iov[i].iov_len);
+ }
+ return _str.size() - old_size;
+ }
+ const std::string& str() const { return _str; }
+private:
+ std::string _str;
+};
+
+TEST(RecordIOTest, empty_record) {
+ butil::Record r;
+ ASSERT_EQ((size_t)0, r.MetaCount());
+ ASSERT_TRUE(r.Meta("foo") == NULL);
+ ASSERT_FALSE(r.RemoveMeta("foo"));
+ ASSERT_TRUE(r.Payload().empty());
+ ASSERT_TRUE(r.MutablePayload()->empty());
+}
+
+TEST(RecordIOTest, manipulate_record) {
+ butil::Record r1;
+ ASSERT_EQ((size_t)0, r1.MetaCount());
+ butil::IOBuf* foo_val = r1.MutableMeta("foo");
+ ASSERT_EQ((size_t)1, r1.MetaCount());
+ ASSERT_TRUE(foo_val->empty());
+ foo_val->append("foo_data");
+ ASSERT_EQ(foo_val, r1.MutableMeta("foo"));
+ ASSERT_EQ((size_t)1, r1.MetaCount());
+ ASSERT_EQ("foo_data", *foo_val);
+ ASSERT_EQ(foo_val, r1.Meta("foo"));
+
+ butil::IOBuf* bar_val = r1.MutableMeta("bar");
+ ASSERT_EQ((size_t)2, r1.MetaCount());
+ ASSERT_TRUE(bar_val->empty());
+ bar_val->append("bar_data");
+ ASSERT_EQ(bar_val, r1.MutableMeta("bar"));
+ ASSERT_EQ((size_t)2, r1.MetaCount());
+ ASSERT_EQ("bar_data", *bar_val);
+ ASSERT_EQ(bar_val, r1.Meta("bar"));
+
+ butil::Record r2 = r1;
+
+ ASSERT_TRUE(r1.RemoveMeta("foo"));
+ ASSERT_EQ((size_t)1, r1.MetaCount());
+ ASSERT_TRUE(r1.Meta("foo") == NULL);
+
+ ASSERT_EQ(foo_val, r2.Meta("foo"));
+ ASSERT_EQ("foo_data", *foo_val);
+}
+
+TEST(RecordIOTest, invalid_name) {
+ char name[258];
+ for (size_t i = 0; i < sizeof(name); ++i) {
+ name[i] = 'a';
+ }
+ name[sizeof(name) - 1] = 0;
+ butil::Record r;
+ ASSERT_EQ(NULL, r.MutableMeta(name));
+}
+
+TEST(RecordIOTest, write_read_basic) {
+ StringWriter sw;
+ butil::RecordWriter rw(&sw);
+
+ butil::Record src;
+ ASSERT_EQ(0, rw.Write(src));
+
+ butil::IOBuf* foo_val = src.MutableMeta("foo");
+ foo_val->append("foo_data");
+ ASSERT_EQ(0, rw.Write(src));
+
+ butil::IOBuf* bar_val = src.MutableMeta("bar");
+ bar_val->append("bar_data");
+ ASSERT_EQ(0, rw.Write(src));
+
+ src.MutablePayload()->append("payload_data");
+ ASSERT_EQ(0, rw.Write(src));
+
+ ASSERT_EQ(0, rw.Flush());
+ std::cout << "len=" << sw.str().size()
+ << " content=" << butil::PrintedAsBinary(sw.str(), 256) << std::endl;
+
+ StringReader sr(sw.str());
+ butil::RecordReader rr(&sr);
+ butil::Record r1;
+ ASSERT_TRUE(rr.ReadNext(&r1));
+ ASSERT_EQ(0, rr.last_error());
+ ASSERT_EQ((size_t)0, r1.MetaCount());
+ ASSERT_TRUE(r1.Payload().empty());
+
+ butil::Record r2;
+ ASSERT_TRUE(rr.ReadNext(&r2));
+ ASSERT_EQ(0, rr.last_error());
+ ASSERT_EQ((size_t)1, r2.MetaCount());
+ ASSERT_EQ("foo", r2.MetaAt(0).name);
+ ASSERT_EQ("foo_data", *r2.MetaAt(0).data);
+ ASSERT_TRUE(r2.Payload().empty());
+
+ butil::Record r3;
+ ASSERT_TRUE(rr.ReadNext(&r3));
+ ASSERT_EQ(0, rr.last_error());
+ ASSERT_EQ((size_t)2, r3.MetaCount());
+ ASSERT_EQ("foo", r3.MetaAt(0).name);
+ ASSERT_EQ("foo_data", *r3.MetaAt(0).data);
+ ASSERT_EQ("bar", r3.MetaAt(1).name);
+ ASSERT_EQ("bar_data", *r3.MetaAt(1).data);
+ ASSERT_TRUE(r3.Payload().empty());
+
+ butil::Record r4;
+ ASSERT_TRUE(rr.ReadNext(&r4));
+ ASSERT_EQ(0, rr.last_error());
+ ASSERT_EQ((size_t)2, r4.MetaCount());
+ ASSERT_EQ("foo", r4.MetaAt(0).name);
+ ASSERT_EQ("foo_data", *r4.MetaAt(0).data);
+ ASSERT_EQ("bar", r4.MetaAt(1).name);
+ ASSERT_EQ("bar_data", *r4.MetaAt(1).data);
+ ASSERT_EQ("payload_data", r4.Payload());
+
+ ASSERT_FALSE(rr.ReadNext(NULL));
+ ASSERT_EQ((int)butil::RecordReader::END_OF_READER, rr.last_error());
+ ASSERT_EQ(sw.str().size(), rr.read_bytes());
+}
+
+static std::string rand_string(int min_len, int max_len) {
+ const int len = butil::fast_rand_in(min_len, max_len);
+ std::string str;
+ str.reserve(len);
+ for (int i = 0; i < len; ++i) {
+ str.push_back(butil::fast_rand_in('a', 'Z'));
+ }
+ return str;
+}
+
+TEST(RecordIOTest, write_read_random) {
+ StringWriter sw;
+ butil::RecordWriter rw(&sw);
+
+ const int N = 1024;
+ std::vector<std::pair<std::string, std::string>> name_value_list;
+ size_t nbytes = 0;
+ std::map<int, size_t> breaking_offsets;
+ for (int i = 0; i < N; ++i) {
+ butil::Record src;
+ std::string value = rand_string(10, 20);
+ std::string name = butil::string_printf("name_%d_", i) + value;
+ src.MutableMeta(name)->append(value);
+ ASSERT_EQ(0, rw.Write(src));
+ if (butil::fast_rand_less_than(70) == 0) {
+ breaking_offsets[i] = nbytes;
+ } else {
+ name_value_list.push_back(std::make_pair(name, value));
+ }
+ nbytes += src.ByteSize();
+ }
+ ASSERT_EQ(0, rw.Flush());
+ std::string str = sw.str();
+ ASSERT_EQ(nbytes, str.size());
+ // break some records
+ int break_idx = 0;
+ for (auto it = breaking_offsets.begin(); it != breaking_offsets.end(); ++it) {
+ switch (break_idx++ % 10) {
+ case 0:
+ str[it->second] = 'r';
+ break;
+ case 1:
+ str[it->second + 1] = 'd';
+ break;
+ case 2:
+ str[it->second + 2] = 'i';
+ break;
+ case 3:
+ str[it->second + 3] = 'o';
+ break;
+ case 4:
+ ++str[it->second + 4];
+ break;
+ case 5:
+ str[it->second + 4] = 8;
+ break;
+ case 6:
+ ++str[it->second + 5];
+ break;
+ case 7:
+ ++str[it->second + 6];
+ break;
+ case 8:
+ ++str[it->second + 7];
+ break;
+ case 9:
+ ++str[it->second + 8];
+ break;
+ default:
+ ASSERT_TRUE(false) << "never";
+ }
+ }
+ ASSERT_EQ((size_t)N - breaking_offsets.size(), name_value_list.size());
+ std::cout << "sw.size=" << str.size()
+ << " nbreak=" << breaking_offsets.size() << std::endl;
+
+ StringReader sr(str);
+ ASSERT_LT(0, butil::WriteFile(butil::FilePath("recordio_ref.io"), str.data(), str.size()));
+ butil::RecordReader rr(&sr);
+ size_t j = 0;
+ butil::Record r;
+ for (; rr.ReadNext(&r); ++j) {
+ ASSERT_LT(j, name_value_list.size());
+ ASSERT_EQ((size_t)1, r.MetaCount());
+ ASSERT_EQ(name_value_list[j].first, r.MetaAt(0).name) << j;
+ ASSERT_EQ(name_value_list[j].second, *r.MetaAt(0).data);
+ }
+ ASSERT_EQ((int)butil::RecordReader::END_OF_READER, rr.last_error());
+ ASSERT_EQ(str.size(), rr.read_bytes());
+ ASSERT_EQ(j, name_value_list.size());
+}
+
+} // namespace
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org