From 80382a8a4dce8cafecf747730bedec8823eac5ae Mon Sep 17 00:00:00 2001 From: Elliott Clark Date: Tue, 12 Apr 2016 12:03:52 -0700 Subject: [PATCH] HBASE-15620 Add on Call serialization Summary: Add on delimited serialization so that request headers and request payloads can be serialized. Test Plan: Add a unit test. Differential Revision: https://reviews.facebook.net/D56757 --- hbase-native-client/core/BUCK | 1 + hbase-native-client/core/client-dispatcher.cc | 2 +- hbase-native-client/core/client-dispatcher.h | 6 +- .../core/client-serialize-handler.cc | 87 +++---------- .../core/client-serialize-handler.h | 23 ++-- hbase-native-client/core/client.h | 2 +- hbase-native-client/core/connection-factory.h | 2 +- hbase-native-client/core/get-request.h | 2 +- hbase-native-client/core/get-result.h | 2 +- hbase-native-client/core/location-cache.cc | 3 +- hbase-native-client/core/location-cache.h | 2 +- hbase-native-client/core/pipeline.h | 4 +- hbase-native-client/core/request.h | 12 +- hbase-native-client/core/response.h | 2 +- hbase-native-client/core/service.h | 2 +- hbase-native-client/core/simple-client.cc | 26 +++- hbase-native-client/core/table-name.h | 2 +- hbase-native-client/serde/BUCK | 42 +++++++ hbase-native-client/serde/client-deserializer.cc | 19 +++ hbase-native-client/serde/client-deserializer.h | 36 ++++++ .../serde/client-serializer-test.cc | 56 +++++++++ hbase-native-client/serde/client-serializer.cc | 137 +++++++++++++++++++++ hbase-native-client/serde/client-serializer.h | 55 +++++++++ 23 files changed, 428 insertions(+), 97 deletions(-) create mode 100644 hbase-native-client/serde/BUCK create mode 100644 hbase-native-client/serde/client-deserializer.cc create mode 100644 hbase-native-client/serde/client-deserializer.h create mode 100644 hbase-native-client/serde/client-serializer-test.cc create mode 100644 hbase-native-client/serde/client-serializer.cc create mode 100644 hbase-native-client/serde/client-serializer.h diff --git a/hbase-native-client/core/BUCK b/hbase-native-client/core/BUCK index 2b00d66..af4db04 100644 --- a/hbase-native-client/core/BUCK +++ b/hbase-native-client/core/BUCK @@ -59,6 +59,7 @@ cxx_library(name="core", ], deps=[ "//if:if", + "//serde:serde", "//third-party:folly", "//third-party:wangle", "//third-party:zookeeper_mt", diff --git a/hbase-native-client/core/client-dispatcher.cc b/hbase-native-client/core/client-dispatcher.cc index d356759..c4c9582 100644 --- a/hbase-native-client/core/client-dispatcher.cc +++ b/hbase-native-client/core/client-dispatcher.cc @@ -34,7 +34,7 @@ void ClientDispatcher::read(Context *ctx, Response in) { p.setValue(in); } -Future ClientDispatcher::operator()(Request arg) { +Future ClientDispatcher::operator()(Request &arg) { auto call_id = ++current_call_id_; arg.set_call_id(call_id); auto &p = requests_[call_id]; diff --git a/hbase-native-client/core/client-dispatcher.h b/hbase-native-client/core/client-dispatcher.h index 4b9d35a..8429b57 100644 --- a/hbase-native-client/core/client-dispatcher.h +++ b/hbase-native-client/core/client-dispatcher.h @@ -27,11 +27,11 @@ namespace hbase { class ClientDispatcher - : public wangle::ClientDispatcherBase { public: void read(Context *ctx, Response in) override; - folly::Future operator()(Request arg) override; + folly::Future operator()(Request &arg) override; folly::Future close(Context *ctx) override; folly::Future close() override; @@ -39,4 +39,4 @@ private: std::unordered_map> requests_; uint32_t current_call_id_ = 1; }; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/core/client-serialize-handler.cc b/hbase-native-client/core/client-serialize-handler.cc index cad1308..1206e90 100644 --- a/hbase-native-client/core/client-serialize-handler.cc +++ b/hbase-native-client/core/client-serialize-handler.cc @@ -19,86 +19,39 @@ #include "core/client-serialize-handler.h" +#include + +#include "core/request.h" +#include "core/response.h" + #include using namespace hbase; using namespace folly; using namespace wangle; -static const std::string PREAMBLE = "HBas"; -static const std::string INTERFACE = "ClientService"; -static const uint8_t RPC_VERSION = 0; -static const uint8_t AUTH_TYPE = 80; - // TODO(eclark): Make this actually do ANYTHING. -void ClientSerializeHandler::read(Context *ctx, std::unique_ptr msg) { - Response received; - ctx->fireRead(received); +void ClientSerializeHandler::read(Context *ctx, std::unique_ptr buf) { + if (buf) { + buf->coalesce(); + std::string data((const char *)buf->data(), buf->length()); + Response received; + LOG(ERROR) << "Got back data" << data; + ctx->fireRead(received); + } } -Future ClientSerializeHandler::write(Context *ctx, Request r) { +Future ClientSerializeHandler::write(Context *ctx, Request &r) { // Keep track of if we have sent the header. if (need_send_header_) { need_send_header_ = false; - // Should this be replacing the IOBuf rather than - // sending several different calls? - write_preamble(ctx); - write_header(ctx); + // Should we be sending just one fireWrite? + auto pre = serializer_.preamble(); + auto header = serializer_.header("elliott"); + pre->appendChain(std::move(header)); + ctx->fireWrite(std::move(pre)); } - // Send out the actual request and not just a test string. - std::string out{"test"}; - return ctx->fireWrite(prepend_length(IOBuf::copyBuffer(out))); -} - -Future ClientSerializeHandler::write_preamble(Context *ctx) { - auto magic = IOBuf::copyBuffer(PREAMBLE); - auto buf = IOBuf::create(2); - buf->append(2); - folly::io::RWPrivateCursor c(buf.get()); - - // Version - c.write(RPC_VERSION); - // Standard security aka Please don't lie to me. - c.write(AUTH_TYPE); - magic->appendChain(std::move(buf)); - return ctx->fireWrite(std::move(magic)); -} - -Future ClientSerializeHandler::write_header(Context *ctx) { - pb::ConnectionHeader h; - - // TODO(eclark): Make this not a total lie. - h.mutable_user_info()->set_effective_user("elliott"); - // The service name that we want to talk to. - // - // Right now we're completely ignoring the service interface. - // That may or may not be the correct thing to do. - // It worked for a while with the java client; until it - // didn't. - h.set_service_name(INTERFACE); - // TODO(eclark): Make this 1 copy. - auto msg = IOBuf::copyBuffer(h.SerializeAsString()); - return ctx->fireWrite(prepend_length(std::move(msg))); -} - -// Our own simple version of LengthFieldPrepender -std::unique_ptr -ClientSerializeHandler::prepend_length(std::unique_ptr msg) { - // Java ints are 4 long. So create a buffer that large - auto len_buf = IOBuf::create(4); - // Then make those bytes visible. - len_buf->append(4); - - io::RWPrivateCursor c(len_buf.get()); - // Get the size of the data to be pushed out the network. - auto size = msg->computeChainDataLength(); - - // Write the length to this IOBuf. - c.writeBE(static_cast(size)); - - // Then attach the origional to the back of len_buf - len_buf->appendChain(std::move(msg)); - return len_buf; + return ctx->fireWrite(serializer_.request(r.call_id(), r.method(), r.msg())); } diff --git a/hbase-native-client/core/client-serialize-handler.h b/hbase-native-client/core/client-serialize-handler.h index 961a03b..3f875f9 100644 --- a/hbase-native-client/core/client-serialize-handler.h +++ b/hbase-native-client/core/client-serialize-handler.h @@ -20,25 +20,24 @@ #include -#include "if/HBase.pb.h" -#include "if/RPC.pb.h" -#include "core/request.h" -#include "core/response.h" +#include "serde/client-serializer.h" + +// Forward decs. +namespace hbase { +class Request; +class Response; +} namespace hbase { class ClientSerializeHandler - : public wangle::Handler, Response, Request, + : public wangle::Handler, Response, Request &, std::unique_ptr> { public: void read(Context *ctx, std::unique_ptr msg) override; - folly::Future write(Context *ctx, Request r) override; + folly::Future write(Context *ctx, Request &r) override; private: - folly::Future write_preamble(Context *ctx); - folly::Future write_header(Context *ctx); - // Our own simple version of LengthFieldPrepender - std::unique_ptr - prepend_length(std::unique_ptr msg); bool need_send_header_ = true; + ClientSerializer serializer_; }; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/core/client.h b/hbase-native-client/core/client.h index 818bc6b..c2dc226 100644 --- a/hbase-native-client/core/client.h +++ b/hbase-native-client/core/client.h @@ -39,4 +39,4 @@ private: LocationCache location_cache; }; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/core/connection-factory.h b/hbase-native-client/core/connection-factory.h index 6f450c2..a27019b 100644 --- a/hbase-native-client/core/connection-factory.h +++ b/hbase-native-client/core/connection-factory.h @@ -38,4 +38,4 @@ public: private: wangle::ClientBootstrap bootstrap_; }; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/core/get-request.h b/hbase-native-client/core/get-request.h index c9113ad..bb755c5 100644 --- a/hbase-native-client/core/get-request.h +++ b/hbase-native-client/core/get-request.h @@ -32,4 +32,4 @@ private: TableName table_name_; std::string key_; }; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/core/get-result.h b/hbase-native-client/core/get-result.h index e021316..a49ad98 100644 --- a/hbase-native-client/core/get-result.h +++ b/hbase-native-client/core/get-result.h @@ -29,4 +29,4 @@ public: private: std::string key_; }; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/core/location-cache.cc b/hbase-native-client/core/location-cache.cc index 34e3236..52e86e3 100644 --- a/hbase-native-client/core/location-cache.cc +++ b/hbase-native-client/core/location-cache.cc @@ -70,8 +70,7 @@ ServerName LocationCache::ReadMetaLocation() { // This needs to be int rather than size_t as that's what ZK expects. int len = sizeof(contents); // TODO(elliott): handle disconnects/reconntion as needed. - int zk_result = - zoo_get(this->zk_, META_LOCATION, 0, contents, &len, nullptr); + int zk_result = zoo_get(this->zk_, META_LOCATION, 0, contents, &len, nullptr); if (zk_result != ZOK || len < 9) { LOG(ERROR) << "Error getting meta location."; throw runtime_error("Error getting meta location"); diff --git a/hbase-native-client/core/location-cache.h b/hbase-native-client/core/location-cache.h index efcfde5..28a1ee1 100644 --- a/hbase-native-client/core/location-cache.h +++ b/hbase-native-client/core/location-cache.h @@ -51,4 +51,4 @@ private: zhandle_t *zk_; }; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/core/pipeline.h b/hbase-native-client/core/pipeline.h index d199d08..435525a 100644 --- a/hbase-native-client/core/pipeline.h +++ b/hbase-native-client/core/pipeline.h @@ -24,11 +24,11 @@ #include "core/response.h" namespace hbase { -using SerializePipeline = wangle::Pipeline; +using SerializePipeline = wangle::Pipeline; class RpcPipelineFactory : public wangle::PipelineFactory { public: SerializePipeline::Ptr newPipeline(std::shared_ptr sock) override; }; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/core/request.h b/hbase-native-client/core/request.h index 39083ed..6f2f179 100644 --- a/hbase-native-client/core/request.h +++ b/hbase-native-client/core/request.h @@ -19,6 +19,8 @@ #pragma once #include +#include +#include namespace hbase { class Request { @@ -26,8 +28,16 @@ public: Request() : call_id_(0) {} uint32_t call_id() { return call_id_; } void set_call_id(uint32_t call_id) { call_id_ = call_id; } + google::protobuf::Message *msg() { return msg_.get(); } + void set_msg(std::unique_ptr &&msg) { + msg_ = std::move(msg); + } + std::string method() { return method_; } + void set_method(std::string method) { method_ = method; } private: uint32_t call_id_; + std::unique_ptr msg_ = nullptr; + std::string method_ = "Get"; }; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/core/response.h b/hbase-native-client/core/response.h index 34a284d..61024af 100644 --- a/hbase-native-client/core/response.h +++ b/hbase-native-client/core/response.h @@ -31,4 +31,4 @@ public: private: uint32_t call_id_; }; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/core/service.h b/hbase-native-client/core/service.h index 880e65f..28bf7bf 100644 --- a/hbase-native-client/core/service.h +++ b/hbase-native-client/core/service.h @@ -23,4 +23,4 @@ namespace hbase { using HBaseService = wangle::Service; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/core/simple-client.cc b/hbase-native-client/core/simple-client.cc index 08e886a..eadb210 100644 --- a/hbase-native-client/core/simple-client.cc +++ b/hbase-native-client/core/simple-client.cc @@ -24,17 +24,27 @@ #include #include +#include #include "core/client.h" #include "core/connection-factory.h" #include "if/ZooKeeper.pb.h" +#include "if/Client.pb.h" using namespace folly; using namespace std; +using namespace std::chrono; using namespace hbase; using namespace hbase::pb; +using namespace google::protobuf; + +// TODO(eclark): remove the need for this. +DEFINE_string(region, "1588230740", "What region to send a get to"); +DEFINE_string(row, "test", "What row to get"); int main(int argc, char *argv[]) { + google::SetUsageMessage( + "Simple client to get a single row from HBase on the comamnd line"); google::ParseCommandLineFlags(&argc, &argv, true); google::InitGoogleLogging(argv[0]); @@ -51,7 +61,21 @@ int main(int argc, char *argv[]) { // Send the request Request r; - conn(r).get(); + + // This is a get request so make that + unique_ptr msg = make_unique(); + + // Set what region + msg->mutable_region()->set_value(FLAGS_region); + // It's always this. + msg->mutable_region()->set_type( + RegionSpecifier_RegionSpecifierType:: + RegionSpecifier_RegionSpecifierType_ENCODED_REGION_NAME); + // What row. + msg->mutable_get()->set_row(FLAGS_row); + // Send it. + r.set_msg(std::move(msg)); + conn(r).get(milliseconds(500)); return 0; } diff --git a/hbase-native-client/core/table-name.h b/hbase-native-client/core/table-name.h index 796115b..37c3461 100644 --- a/hbase-native-client/core/table-name.h +++ b/hbase-native-client/core/table-name.h @@ -29,4 +29,4 @@ public: explicit TableName(std::string tableName); explicit TableName(std::string namespaceName, std::string tableName); }; -} // namespace hbase +} // namespace hbase diff --git a/hbase-native-client/serde/BUCK b/hbase-native-client/serde/BUCK new file mode 100644 index 0000000..0ee109b --- /dev/null +++ b/hbase-native-client/serde/BUCK @@ -0,0 +1,42 @@ +## +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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. + +cxx_library(name="serde", + exported_headers=[ + "client-serializer.h", + "client-deserializer.h", + ], + srcs=[ + "client-serializer.cc", + "client-deserializer.cc", + ], + deps=[ + "//if:if", + "//third-party:folly", + "//third-party:wangle", + ], + visibility=[ + 'PUBLIC', + ], ) + +cxx_test(name="client-serializer-test", + srcs=[ + "client-serializer-test.cc", + ], + deps=[ + ":serde", + ],) diff --git a/hbase-native-client/serde/client-deserializer.cc b/hbase-native-client/serde/client-deserializer.cc new file mode 100644 index 0000000..06c0e33 --- /dev/null +++ b/hbase-native-client/serde/client-deserializer.cc @@ -0,0 +1,19 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * + */ + diff --git a/hbase-native-client/serde/client-deserializer.h b/hbase-native-client/serde/client-deserializer.h new file mode 100644 index 0000000..e1aec2b --- /dev/null +++ b/hbase-native-client/serde/client-deserializer.h @@ -0,0 +1,36 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * + */ +#pragma once + +#include + +// Forward +namespace google { +namespace protobuf { +class Message; +} +} + +namespace hbase { +class CliendDeserializer { +public: + bool parse_delimited(folly::IOBuf *buf, google::protobuf::Message *header); +} + +} // namespace hbase diff --git a/hbase-native-client/serde/client-serializer-test.cc b/hbase-native-client/serde/client-serializer-test.cc new file mode 100644 index 0000000..46c1aa7 --- /dev/null +++ b/hbase-native-client/serde/client-serializer-test.cc @@ -0,0 +1,56 @@ +#include + +#include +#include + +#include "serde/client-serializer.h" +#include "if/HBase.pb.h" +#include "if/RPC.pb.h" + +using namespace hbase; +using namespace hbase::pb; +using namespace folly; +using namespace folly::io; + +TEST(ClientSerializerTest, PreambleIncludesHBas) { + ClientSerializer ser; + auto buf = ser.preamble(); + const char *p = reinterpret_cast(buf->data()); + // Take the first for chars and make sure they are the + // magic string + EXPECT_EQ("HBas", std::string(p, 4)); + + EXPECT_EQ(6, buf->computeChainDataLength()); +} + +TEST(ClientSerializerTest, PreambleIncludesVersion) { + ClientSerializer ser; + auto buf = ser.preamble(); + EXPECT_EQ(0, static_cast(buf->data())[4]); + EXPECT_EQ(80, static_cast(buf->data())[5]); +} + +TEST(ClientSerializerTest, TestHeaderLengthPrefixed) { + ClientSerializer ser; + auto header = ser.header("elliott"); + + // The header should be prefixed by 4 bytes of length. + EXPECT_EQ(4, header->length()); + EXPECT_TRUE(header->length() < header->computeChainDataLength()); + EXPECT_TRUE(header->isChained()); + + // Now make sure the length is correct. + Cursor cursor(header.get()); + auto prefixed_len = cursor.readBE(); + EXPECT_EQ(prefixed_len, header->next()->length()); +} + +TEST(ClientSerializerTest, TestHeaderDecode) { + ClientSerializer ser; + auto buf = ser.header("elliott"); + auto header_buf = buf->next(); + ConnectionHeader h; + + EXPECT_TRUE(h.ParseFromArray(header_buf->data(), header_buf->length())); + EXPECT_EQ("elliott", h.user_info().effective_user()); +} diff --git a/hbase-native-client/serde/client-serializer.cc b/hbase-native-client/serde/client-serializer.cc new file mode 100644 index 0000000..a89482e --- /dev/null +++ b/hbase-native-client/serde/client-serializer.cc @@ -0,0 +1,137 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 "serde/client-serializer.h" + +#include +#include +#include + +#include "if/HBase.pb.h" +#include "if/RPC.pb.h" +#include "if/Client.pb.h" + +using namespace hbase; + +using folly::IOBuf; +using folly::io::RWPrivateCursor; +using google::protobuf::Message; +using google::protobuf::io::ArrayOutputStream; +using google::protobuf::io::CodedOutputStream; +using google::protobuf::io::ZeroCopyOutputStream; +using std::string; +using std::unique_ptr; + +static const std::string PREAMBLE = "HBas"; +static const std::string INTERFACE = "ClientService"; +static const uint8_t RPC_VERSION = 0; +static const uint8_t DEFAULT_AUTH_TYPE = 80; + +ClientSerializer::ClientSerializer() : auth_type_(DEFAULT_AUTH_TYPE) {} + +unique_ptr ClientSerializer::preamble() { + auto magic = IOBuf::copyBuffer(PREAMBLE, 0, 2); + magic->append(2); + RWPrivateCursor c(magic.get()); + c.skip(4); + // Version + c.write(RPC_VERSION); + // Standard security aka Please don't lie to me. + c.write(auth_type_); + return magic; +} + +unique_ptr ClientSerializer::header(string user) { + pb::ConnectionHeader h; + + // TODO(eclark): Make this not a total lie. + h.mutable_user_info()->set_effective_user(user); + // The service name that we want to talk to. + // + // Right now we're completely ignoring the service interface. + // That may or may not be the correct thing to do. + // It worked for a while with the java client; until it + // didn't. + h.set_service_name(INTERFACE); + return prepend_length(serialize_message(h)); +} + +unique_ptr ClientSerializer::request(uint32_t call_id, string method, Message* msg) { + pb::RequestHeader rq; + rq.set_method_name(method); + rq.set_call_id(call_id); + rq.set_request_param(msg!= nullptr); + auto ser_header = serialize_delimited(rq); + if (msg != nullptr) { + auto ser_req = serialize_delimited(*msg); + ser_header->appendChain(std::move(ser_req)); + } + + return prepend_length(std::move(ser_header)); +} + +unique_ptr ClientSerializer::prepend_length(unique_ptr msg) { + // Java ints are 4 long. So create a buffer that large + auto len_buf = IOBuf::create(4); + // Then make those bytes visible. + len_buf->append(4); + + RWPrivateCursor c(len_buf.get()); + // Get the size of the data to be pushed out the network. + auto size = msg->computeChainDataLength(); + + // Write the length to this IOBuf. + c.writeBE(static_cast(size)); + + // Then attach the origional to the back of len_buf + len_buf->appendChain(std::move(msg)); + return len_buf; +} + +unique_ptr ClientSerializer::serialize_delimited(Message &msg) { + // Get the buffer size needed for just the message. + int msg_size = msg.ByteSize(); + int buf_size = CodedOutputStream::VarintSize32(msg_size) + msg_size; + + // Create a buffer big enough to hold the varint and the object. + auto buf = IOBuf::create(buf_size); + buf->append(buf_size); + + // Create the array output stream. + ArrayOutputStream aos{buf->writableData(), static_cast(buf->length())}; + // Wrap the ArrayOuputStream in the coded output stream to allow writing + // Varint32 + CodedOutputStream cos{&aos}; + + // Write out the size. + cos.WriteVarint32(msg_size); + + // Now write the rest out. + // We're using the protobuf output streams here to keep track + // of where in the output array we are rather than IOBuf. + msg.SerializeWithCachedSizesToArray( + cos.GetDirectBufferForNBytesAndAdvance(msg_size)); + + // Return the buffer. + return buf; +} +// TODO(eclark): Make this 1 copy. +unique_ptr ClientSerializer::serialize_message(Message &msg) { + auto buf = IOBuf::copyBuffer(msg.SerializeAsString()); + return buf; +} diff --git a/hbase-native-client/serde/client-serializer.h b/hbase-native-client/serde/client-serializer.h new file mode 100644 index 0000000..c1049a8 --- /dev/null +++ b/hbase-native-client/serde/client-serializer.h @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * + */ +#pragma once + +#include +#include +#include + +// Forward +namespace google { +namespace protobuf { +class Message; +} +} +namespace hbase { +class Request; +} + +namespace hbase { +class ClientSerializer { +public: + ClientSerializer(); + std::unique_ptr preamble(); + std::unique_ptr header(std::string user); + std::unique_ptr request(uint32_t call_id, std::string method, google::protobuf::Message * msg); + +private: + std::unique_ptr + serialize_message(google::protobuf::Message &msg); + + std::unique_ptr + serialize_delimited(google::protobuf::Message &msg); + + std::unique_ptr + prepend_length(std::unique_ptr msg); + + uint8_t auth_type_; +}; +} // namespace hbase -- 2.8.0-rc2