You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by wh...@apache.org on 2015/07/20 23:27:37 UTC
[1/2] hadoop git commit: HDFS-8764. Generate Hadoop RPC stubs from
protobuf definitions. Contributed by Haohui Mai.
Repository: hadoop
Updated Branches:
refs/heads/HDFS-8707 6416b38ca -> 172623de9
HDFS-8764. Generate Hadoop RPC stubs from protobuf definitions. Contributed by Haohui Mai.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/928a9a14
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/928a9a14
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/928a9a14
Branch: refs/heads/HDFS-8707
Commit: 928a9a14285c6f71d1913144a5f8bc264f4d49e0
Parents: 6416b38
Author: Haohui Mai <wh...@apache.org>
Authored: Mon Jul 13 16:53:13 2015 -0700
Committer: Haohui Mai <wh...@apache.org>
Committed: Wed Jul 15 16:59:04 2015 -0700
----------------------------------------------------------------------
.../native/libhdfspp/lib/proto/CMakeLists.txt | 46 +++++++++-
.../native/libhdfspp/lib/proto/cpp_helpers.h | 82 +++++++++++++++++
.../libhdfspp/lib/proto/protoc_gen_hrpc.cc | 93 ++++++++++++++++++++
3 files changed, 220 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/928a9a14/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/CMakeLists.txt
index 156a7f4..3f703b2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/CMakeLists.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/CMakeLists.txt
@@ -18,4 +18,48 @@ protobuf_generate_cpp(PROTO_SRCS PROTO_HDRS
${COMMON_PROTO_DIR}/Security.proto
)
-add_library(proto ${PROTO_SRCS} ${PROTO_HDRS})
+add_executable(protoc-gen-hrpc protoc_gen_hrpc.cc)
+target_link_libraries(protoc-gen-hrpc ${PROTOBUF_PROTOC_LIBRARY} ${PROTOBUF_LIBRARY})
+
+function(GEN_HRPC SRCS)
+ if(NOT ARGN)
+ message(SEND_ERROR "Error: GEN_HRPC() called without any proto files")
+ return()
+ endif()
+
+ if(DEFINED PROTOBUF_IMPORT_DIRS)
+ foreach(DIR ${PROTOBUF_IMPORT_DIRS})
+ get_filename_component(ABS_PATH ${DIR} ABSOLUTE)
+ list(FIND _protobuf_include_path ${ABS_PATH} _contains_already)
+ if(${_contains_already} EQUAL -1)
+ list(APPEND _protobuf_include_path -I ${ABS_PATH})
+ endif()
+ endforeach()
+ endif()
+
+ set(${SRCS})
+
+ foreach(FIL ${ARGN})
+ get_filename_component(ABS_FIL ${FIL} ABSOLUTE)
+ get_filename_component(FIL_WE ${FIL} NAME_WE)
+
+ list(APPEND ${SRCS} "${CMAKE_CURRENT_BINARY_DIR}/${FIL_WE}.hrpc.inl")
+
+ add_custom_command(
+ OUTPUT "${CMAKE_CURRENT_BINARY_DIR}/${FIL_WE}.hrpc.inl"
+ COMMAND ${PROTOBUF_PROTOC_EXECUTABLE}
+ ARGS --plugin=protoc-gen-hrpc=${CMAKE_CURRENT_BINARY_DIR}/protoc-gen-hrpc --hrpc_out=${CMAKE_CURRENT_BINARY_DIR} ${_protobuf_include_path} ${ABS_FIL}
+ DEPENDS ${ABS_FIL} ${PROTOBUF_PROTOC_EXECUTABLE} protoc-gen-hrpc
+ COMMENT "Running HRPC protocol buffer compiler on ${FIL}"
+ VERBATIM )
+ endforeach()
+
+ set_source_files_properties(${${SRCS}} PROPERTIES GENERATED TRUE)
+ set(${SRCS} ${${SRCS}} PARENT_SCOPE)
+endfunction()
+
+gen_hrpc(HRPC_SRCS
+ ${CLIENT_PROTO_DIR}/ClientNamenodeProtocol.proto
+)
+
+add_library(proto ${PROTO_SRCS} ${PROTO_HDRS} ${HRPC_SRCS})
http://git-wip-us.apache.org/repos/asf/hadoop/blob/928a9a14/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/cpp_helpers.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/cpp_helpers.h b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/cpp_helpers.h
new file mode 100644
index 0000000..6f380ad
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/cpp_helpers.h
@@ -0,0 +1,82 @@
+// Protocol Buffers - Google's data interchange format
+// Copyright 2008 Google Inc. All rights reserved.
+// https://developers.google.com/protocol-buffers/
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+// * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+// * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+// * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+// Author: kenton@google.com (Kenton Varda)
+// Based on original Protocol Buffers design by
+// Sanjay Ghemawat, Jeff Dean, and others.
+
+#ifndef LIBHDFSPP_PROTO_CPP_HELPERS_H_
+#define LIBHDFSPP_PROTO_CPP_HELPERS_H_
+
+#include <string>
+
+/**
+ * The functions in this file are derived from the original implementation of
+ *the protobuf library from Google.
+ **/
+
+static inline std::string StripProto(const std::string &str) {
+ static const std::string kExtension = ".proto";
+ if (str.size() >= kExtension.size() &&
+ str.compare(str.size() - kExtension.size(), kExtension.size(),
+ kExtension) == 0) {
+ return str.substr(0, str.size() - kExtension.size());
+ } else {
+ return str;
+ }
+}
+
+static inline std::string ToCamelCase(const std::string &input) {
+ bool cap_next_letter = true;
+ std::string result;
+ // Note: I distrust ctype.h due to locales.
+ for (size_t i = 0; i < input.size(); i++) {
+ if ('a' <= input[i] && input[i] <= 'z') {
+ if (cap_next_letter) {
+ result += input[i] + ('A' - 'a');
+ } else {
+ result += input[i];
+ }
+ cap_next_letter = false;
+ } else if ('A' <= input[i] && input[i] <= 'Z') {
+ // Capital letters are left as-is.
+ result += input[i];
+ cap_next_letter = false;
+ } else if ('0' <= input[i] && input[i] <= '9') {
+ result += input[i];
+ cap_next_letter = true;
+ } else {
+ cap_next_letter = true;
+ }
+ }
+ return result;
+}
+
+#endif
http://git-wip-us.apache.org/repos/asf/hadoop/blob/928a9a14/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/protoc_gen_hrpc.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/protoc_gen_hrpc.cc b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/protoc_gen_hrpc.cc
new file mode 100644
index 0000000..80522e7
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/lib/proto/protoc_gen_hrpc.cc
@@ -0,0 +1,93 @@
+/**
+ * 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 "cpp_helpers.h"
+
+#include <google/protobuf/compiler/code_generator.h>
+#include <google/protobuf/compiler/plugin.h>
+#include <google/protobuf/descriptor.h>
+#include <google/protobuf/io/printer.h>
+#include <google/protobuf/io/zero_copy_stream.h>
+#include <google/protobuf/stubs/common.h>
+
+using ::google::protobuf::FileDescriptor;
+using ::google::protobuf::MethodDescriptor;
+using ::google::protobuf::ServiceDescriptor;
+using ::google::protobuf::compiler::CodeGenerator;
+using ::google::protobuf::compiler::GeneratorContext;
+using ::google::protobuf::io::Printer;
+using ::google::protobuf::io::ZeroCopyOutputStream;
+
+class StubGenerator : public CodeGenerator {
+public:
+ virtual bool Generate(const FileDescriptor *file, const std::string &,
+ GeneratorContext *ctx,
+ std::string *error) const override;
+
+private:
+ void EmitService(const ServiceDescriptor *service, Printer *out) const;
+ void EmitMethod(const MethodDescriptor *method, Printer *out) const;
+};
+
+bool StubGenerator::Generate(const FileDescriptor *file, const std::string &,
+ GeneratorContext *ctx, std::string *) const {
+ namespace pb = ::google::protobuf;
+ std::unique_ptr<ZeroCopyOutputStream> os(
+ ctx->Open(StripProto(file->name()) + ".hrpc.inl"));
+ Printer out(os.get(), '$');
+ for (int i = 0; i < file->service_count(); ++i) {
+ const ServiceDescriptor *service = file->service(i);
+ EmitService(service, &out);
+ }
+ return true;
+}
+
+void StubGenerator::EmitService(const ServiceDescriptor *service,
+ Printer *out) const {
+ out->Print("\n// GENERATED AUTOMATICALLY. DO NOT MODIFY.\n"
+ "class $service$ {\n"
+ "private:\n"
+ " ::hdfs::RpcEngine *const engine_;\n"
+ "public:\n"
+ " typedef std::function<void(const ::hdfs::Status &)> Callback;\n"
+ " typedef ::google::protobuf::MessageLite Message;\n"
+ " inline $service$(::hdfs::RpcEngine *engine)\n"
+ " : engine_(engine) {}\n",
+ "service", service->name());
+ for (int i = 0; i < service->method_count(); ++i) {
+ const MethodDescriptor *method = service->method(i);
+ EmitMethod(method, out);
+ }
+ out->Print("};\n");
+}
+
+void StubGenerator::EmitMethod(const MethodDescriptor *method,
+ Printer *out) const {
+ out->Print(
+ "\n inline void $camel_method$(const Message *req, "
+ "const std::shared_ptr<Message> &resp, "
+ "Callback &&handler) {\n"
+ " engine_->AsyncRpc(\"$method$\", req, resp, std::move(handler));\n"
+ " }\n",
+ "camel_method", ToCamelCase(method->name()), "method", method->name());
+}
+
+int main(int argc, char *argv[]) {
+ StubGenerator generator;
+ return google::protobuf::compiler::PluginMain(argc, argv, &generator);
+}
[2/2] hadoop git commit: HDFS-8788. Implement unit tests for remote
block reader in libhdfspp. Contributed by Haohui Mai.
Posted by wh...@apache.org.
HDFS-8788. Implement unit tests for remote block reader in libhdfspp. Contributed by Haohui Mai.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/172623de
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/172623de
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/172623de
Branch: refs/heads/HDFS-8707
Commit: 172623de91dcc6e27d2b97d37ccdd8fb9089ab0d
Parents: 928a9a1
Author: Haohui Mai <wh...@apache.org>
Authored: Wed Jul 15 16:58:42 2015 -0700
Committer: Haohui Mai <wh...@apache.org>
Committed: Wed Jul 15 16:59:05 2015 -0700
----------------------------------------------------------------------
hadoop-hdfs-project/hadoop-hdfs-client/pom.xml | 13 ++
.../src/main/native/CMakeLists.txt | 2 +
.../src/main/native/libhdfspp/CMakeLists.txt | 1 +
.../main/native/libhdfspp/tests/CMakeLists.txt | 22 ++
.../native/libhdfspp/tests/mock_connection.cc | 25 +++
.../native/libhdfspp/tests/mock_connection.h | 64 ++++++
.../libhdfspp/tests/remote_block_reader_test.cc | 213 +++++++++++++++++++
7 files changed, 340 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/172623de/hadoop-hdfs-project/hadoop-hdfs-client/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/pom.xml b/hadoop-hdfs-project/hadoop-hdfs-client/pom.xml
index 1a03cbd..9f7070e 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/pom.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/pom.xml
@@ -137,6 +137,19 @@ http://maven.apache.org/xsd/maven-4.0.0.xsd">
</target>
</configuration>
</execution>
+ <execution>
+ <id>native_tests</id>
+ <phase>test</phase>
+ <goals><goal>run</goal></goals>
+ <configuration>
+ <skip>${skipTests}</skip>
+ <target>
+ <exec executable="make" dir="${project.build.directory}/native" failonerror="true">
+ <arg line="test"/>
+ </exec>
+ </target>
+ </configuration>
+ </execution>
</executions>
</plugin>
</plugins>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/172623de/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/CMakeLists.txt
index ef14183..309e99f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/CMakeLists.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/CMakeLists.txt
@@ -18,4 +18,6 @@
cmake_minimum_required(VERSION 2.8 FATAL_ERROR)
+enable_testing()
+
add_subdirectory(libhdfspp)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/172623de/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/CMakeLists.txt
index cae786c..51e3122 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/CMakeLists.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/CMakeLists.txt
@@ -51,3 +51,4 @@ include_directories(
add_subdirectory(third_party/gmock-1.7.0)
add_subdirectory(lib)
+add_subdirectory(tests)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/172623de/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/CMakeLists.txt
new file mode 100644
index 0000000..cd5e1b1
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/CMakeLists.txt
@@ -0,0 +1,22 @@
+#
+# 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.
+#
+
+add_library(test_common OBJECT mock_connection.cc)
+add_executable(remote_block_reader_test remote_block_reader_test.cc $<TARGET_OBJECTS:test_common>)
+target_link_libraries(remote_block_reader_test reader proto common ${PROTOBUF_LIBRARIES} gmock_main)
+add_test(remote_block_reader remote_block_reader_test)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/172623de/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/mock_connection.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/mock_connection.cc b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/mock_connection.cc
new file mode 100644
index 0000000..e1dfdc7
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/mock_connection.cc
@@ -0,0 +1,25 @@
+/**
+ * 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 "mock_connection.h"
+
+namespace hdfs {
+
+MockConnectionBase::~MockConnectionBase() {}
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/172623de/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/mock_connection.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/mock_connection.h b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/mock_connection.h
new file mode 100644
index 0000000..e917e9d
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/mock_connection.h
@@ -0,0 +1,64 @@
+/**
+ * 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.
+ */
+#ifndef LIBHDFSPP_TEST_MOCK_CONNECTION_H_
+#define LIBHDFSPP_TEST_MOCK_CONNECTION_H_
+
+#include <asio/error_code.hpp>
+#include <asio/buffer.hpp>
+#include <asio/streambuf.hpp>
+#include <gmock/gmock.h>
+
+namespace hdfs {
+
+class MockConnectionBase {
+public:
+ virtual ~MockConnectionBase();
+ typedef std::pair<asio::error_code, std::string> ProducerResult;
+ template <class MutableBufferSequence, class Handler>
+ void async_read_some(const MutableBufferSequence &buf, Handler &&handler) {
+ if (produced_.size() == 0) {
+ ProducerResult r = Produce();
+ if (r.first) {
+ handler(r.first, 0);
+ }
+ asio::mutable_buffers_1 data = produced_.prepare(r.second.size());
+ asio::buffer_copy(data, asio::buffer(r.second));
+ produced_.commit(r.second.size());
+ }
+
+ size_t len = std::min(asio::buffer_size(buf), produced_.size());
+ asio::buffer_copy(buf, produced_.data());
+ produced_.consume(len);
+ handler(asio::error_code(), len);
+ }
+
+ template <class ConstBufferSequence, class Handler>
+ void async_write_some(const ConstBufferSequence &buf, Handler &&handler) {
+ // CompletionResult res = OnWrite(buf);
+ handler(asio::error_code(), asio::buffer_size(buf));
+ }
+
+protected:
+ virtual ProducerResult Produce() = 0;
+
+private:
+ asio::streambuf produced_;
+};
+}
+
+#endif
http://git-wip-us.apache.org/repos/asf/hadoop/blob/172623de/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/remote_block_reader_test.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/remote_block_reader_test.cc b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/remote_block_reader_test.cc
new file mode 100644
index 0000000..92cbc8f
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/remote_block_reader_test.cc
@@ -0,0 +1,213 @@
+/**
+ * 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 "mock_connection.h"
+
+#include "datatransfer.pb.h"
+#include "common/util.h"
+#include "reader/block_reader.h"
+
+#include <google/protobuf/io/coded_stream.h>
+#include <google/protobuf/io/zero_copy_stream_impl.h>
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+using namespace hdfs;
+
+using ::hadoop::common::TokenProto;
+using ::hadoop::hdfs::BlockOpResponseProto;
+using ::hadoop::hdfs::ChecksumProto;
+using ::hadoop::hdfs::ExtendedBlockProto;
+using ::hadoop::hdfs::PacketHeaderProto;
+using ::hadoop::hdfs::ReadOpChecksumInfoProto;
+
+using ::asio::buffer;
+using ::asio::error_code;
+using ::asio::mutable_buffers_1;
+using ::testing::Return;
+using std::make_pair;
+using std::string;
+
+namespace pb = ::google::protobuf;
+namespace pbio = pb::io;
+
+namespace hdfs {
+
+class MockDNConnection : public MockConnectionBase {
+public:
+ MOCK_METHOD0(Produce, ProducerResult());
+};
+}
+
+static inline string ToDelimitedString(const pb::MessageLite *msg) {
+ string res;
+ res.reserve(hdfs::DelimitedPBMessageSize(msg));
+ pbio::StringOutputStream os(&res);
+ pbio::CodedOutputStream out(&os);
+ out.WriteVarint32(msg->ByteSize());
+ msg->SerializeToCodedStream(&out);
+ return res;
+}
+
+static inline std::pair<error_code, string> Produce(const std::string &s) {
+ return make_pair(error_code(), s);
+}
+
+static inline std::pair<error_code, string>
+ProducePacket(const std::string &data, const std::string &checksum,
+ int offset_in_block, int seqno, bool last_packet) {
+ PacketHeaderProto proto;
+ proto.set_datalen(data.size());
+ proto.set_offsetinblock(offset_in_block);
+ proto.set_seqno(seqno);
+ proto.set_lastpacketinblock(last_packet);
+
+ char prefix[6];
+ *reinterpret_cast<unsigned *>(prefix) =
+ htonl(data.size() + checksum.size() + sizeof(int));
+ *reinterpret_cast<short *>(prefix + sizeof(int)) = htons(proto.ByteSize());
+ std::string payload(prefix, sizeof(prefix));
+ payload.reserve(payload.size() + proto.ByteSize() + checksum.size() +
+ data.size());
+ proto.AppendToString(&payload);
+ payload += checksum;
+ payload += data;
+ return std::make_pair(error_code(), std::move(payload));
+}
+
+static std::shared_ptr<RemoteBlockReader<MockDNConnection>>
+ReadContent(MockDNConnection *conn, TokenProto *token,
+ const ExtendedBlockProto &block, uint64_t length, uint64_t offset,
+ const mutable_buffers_1 &buf, Status *status, size_t *transferred) {
+ BlockReaderOptions options;
+ auto reader =
+ std::make_shared<RemoteBlockReader<MockDNConnection>>(options, conn);
+ Status result;
+ reader->async_connect(
+ "libhdfs++", token, &block, length, offset,
+ [buf, reader, status, transferred](const Status &stat) {
+ if (!stat.ok()) {
+ *status = stat;
+ } else {
+ reader->async_read_some(
+ buf, [status, transferred](const Status &stat, size_t t) {
+ *transferred = t;
+ *status = stat;
+ });
+ }
+ });
+ return reader;
+}
+
+TEST(RemoteBlockReaderTest, TestReadWholeBlock) {
+ static const size_t kChunkSize = 512;
+ static const string kChunkData(kChunkSize, 'a');
+ MockDNConnection conn;
+ BlockOpResponseProto block_op_resp;
+
+ block_op_resp.set_status(::hadoop::hdfs::Status::SUCCESS);
+
+ EXPECT_CALL(conn, Produce())
+ .WillOnce(Return(Produce(ToDelimitedString(&block_op_resp))))
+ .WillOnce(Return(ProducePacket(kChunkData, "", 0, 1, true)));
+
+ ExtendedBlockProto block;
+ std::string data(kChunkSize, 0);
+ size_t transferred = 0;
+ Status stat;
+ ReadContent(&conn, nullptr, block, kChunkSize, 0,
+ buffer(const_cast<char *>(data.c_str()), data.size()), &stat,
+ &transferred);
+ ASSERT_TRUE(stat.ok());
+ ASSERT_EQ(kChunkSize, transferred);
+ ASSERT_EQ(kChunkData, data);
+}
+
+TEST(RemoteBlockReaderTest, TestReadWithinChunk) {
+ static const size_t kChunkSize = 1024;
+ static const size_t kLength = kChunkSize / 4 * 3;
+ static const size_t kOffset = kChunkSize / 4;
+ static const string kChunkData = string(kOffset, 'a') + string(kLength, 'b');
+
+ MockDNConnection conn;
+ BlockOpResponseProto block_op_resp;
+ ReadOpChecksumInfoProto *checksum_info =
+ block_op_resp.mutable_readopchecksuminfo();
+ checksum_info->set_chunkoffset(0);
+ ChecksumProto *checksum = checksum_info->mutable_checksum();
+ checksum->set_type(::hadoop::hdfs::ChecksumTypeProto::CHECKSUM_NULL);
+ checksum->set_bytesperchecksum(512);
+ block_op_resp.set_status(::hadoop::hdfs::Status::SUCCESS);
+
+ EXPECT_CALL(conn, Produce())
+ .WillOnce(Return(Produce(ToDelimitedString(&block_op_resp))))
+ .WillOnce(Return(ProducePacket(kChunkData, "", kOffset, 1, true)));
+
+ ExtendedBlockProto block;
+ string data(kLength, 0);
+ size_t transferred = 0;
+ Status stat;
+ ReadContent(&conn, nullptr, block, data.size(), kOffset,
+ buffer(const_cast<char *>(data.c_str()), data.size()), &stat,
+ &transferred);
+ ASSERT_TRUE(stat.ok());
+ ASSERT_EQ(kLength, transferred);
+ ASSERT_EQ(kChunkData.substr(kOffset, kLength), data);
+}
+
+TEST(RemoteBlockReaderTest, TestReadMultiplePacket) {
+ static const size_t kChunkSize = 1024;
+ static const string kChunkData(kChunkSize, 'a');
+
+ MockDNConnection conn;
+ BlockOpResponseProto block_op_resp;
+ block_op_resp.set_status(::hadoop::hdfs::Status::SUCCESS);
+
+ EXPECT_CALL(conn, Produce())
+ .WillOnce(Return(Produce(ToDelimitedString(&block_op_resp))))
+ .WillOnce(Return(ProducePacket(kChunkData, "", 0, 1, false)))
+ .WillOnce(Return(ProducePacket(kChunkData, "", kChunkSize, 2, true)));
+
+ ExtendedBlockProto block;
+ string data(kChunkSize, 0);
+ size_t transferred = 0;
+ Status stat;
+ mutable_buffers_1 buf = buffer(const_cast<char *>(data.c_str()), data.size());
+ auto reader = ReadContent(&conn, nullptr, block, data.size(), 0, buf, &stat,
+ &transferred);
+ ASSERT_TRUE(stat.ok());
+ ASSERT_EQ(kChunkSize, transferred);
+ ASSERT_EQ(kChunkData, data);
+
+ data.clear();
+ data.resize(kChunkSize);
+ transferred = 0;
+
+ reader->async_read_some(buf, [&data](const Status &stat, size_t transferred) {
+ ASSERT_TRUE(stat.ok());
+ ASSERT_EQ(kChunkSize, transferred);
+ ASSERT_EQ(kChunkData, data);
+ });
+}
+
+int main(int argc, char *argv[]) {
+ // The following line must be executed to initialize Google Mock
+ // (and Google Test) before running the tests.
+ ::testing::InitGoogleMock(&argc, argv);
+ return RUN_ALL_TESTS();
+}