Merge pull request #784 from ctiller/timeout
Introduce slowdown factor for unit test deadlines
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 57d176f..b58c356 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -24,27 +24,27 @@
If you are planning to work on any of the languages other than C and C++, you
will also need their appropriate development environments.
-If you want to work under Windows, we recommend you to use Visual Studio 2013.
+If you want to work under Windows, we recommend the use of Visual Studio 2013.
The [Community or Express editions](http://www.visualstudio.com/en-us/downloads/download-visual-studio-vs.aspx)
are free and suitable for developing with grpc. Note however that our test
environment and tools are available for Unix environments only at the moment.
## Testing your changes
-We provide a tool to help you run our suite of tests in various environments.
+We provide a tool to help run the suite of tests in various environments.
In order to run most of the available tests, one would need to run:
`./tools/run_tests/run_tests.py`
-If you want to run all the possible tests for all possible languages, do this:
+If you want to run all the possible tests for any of the languages {c, c++, node, php, python}, do this:
-`./tools/run_tests/run_tests.py -lall -call`
+`./tools/run_tests/run_tests.py -l <lang> -c all`
## Adding or removing source code
Each language uses its own build system to work. Currently, the root's Makefile
-and the Visual Studio project files are building the C and C++ source code only
-at the moment. In order to ease the maintenance of these files, we have a
+and the Visual Studio project files are building only the C and C++ source code.
+In order to ease the maintenance of these files, we have a
template system. Please do not contribute manual changes to any of the generated
files. Instead, modify the template files, or the build.json file, and
re-generate the project files using the following command:
diff --git a/INSTALL b/INSTALL
index 2f5f29c..50040d7 100644
--- a/INSTALL
+++ b/INSTALL
@@ -9,15 +9,16 @@
* If you are in a hurry *
*************************
-A typical unix installation won't require any more steps than running:
-
- $ make
- # make install
+ $ git clone https://github.com/grpc/grpc.git
+ $ cd grpc
+ $ git submodule update --init
+ $ make
+ $ sudo make install
You don't need anything else than GNU Make, gcc and autotools. Under a Debian
or Ubuntu system, this should boil down to the following packages:
- # apt-get install build-essential autoconf libtool
+ $ apt-get install build-essential autoconf libtool
Building the python wrapper requires the following:
diff --git a/Makefile b/Makefile
index e87f641..310f0dc 100644
--- a/Makefile
+++ b/Makefile
@@ -195,7 +195,7 @@
Q = @
endif
-VERSION = 0.8.0.0
+VERSION = 0.5.0.0
CPPFLAGS_NO_ARCH += $(addprefix -I, $(INCLUDES)) $(addprefix -D, $(DEFINES))
CPPFLAGS += $(CPPFLAGS_NO_ARCH) $(ARCH_FLAGS)
diff --git a/README.md b/README.md
index fc35934..eacb8c0 100644
--- a/README.md
+++ b/README.md
@@ -13,9 +13,9 @@
of shared C core library [src/core] (src/core).
* C++ source code: [src/cpp] (src/cpp)
- * Python source code: [src/python] (src/python)
* Ruby source code: [src/ruby] (src/ruby)
* NodeJS source code: [src/node] (src/node)
+ * Python source code: [src/python] (src/python)
* PHP source code: [src/php] (src/php)
* C# source code: [src/csharp] (src/csharp)
* Objective-C source code: [src/objective-c] (src/objective-c)
@@ -33,9 +33,9 @@
* shared C core library [src/core] (src/core) : Early adopter ready - Alpha.
* C++ Library: [src/cpp] (src/cpp) : Early adopter ready - Alpha.
- * Python Library: [src/python] (src/python) : Early adopter ready - Alpha.
* Ruby Library: [src/ruby] (src/ruby) : Early adopter ready - Alpha.
* NodeJS Library: [src/node] (src/node) : Early adopter ready - Alpha.
+ * Python Library: [src/python] (src/python) : Usable with limitations - Pre-Alpha.
* PHP Library: [src/php] (src/php) : Pre-Alpha.
* C# Library: [src/csharp] (src/csharp) : Pre-Alpha.
* Objective-C Library: [src/objective-c] (src/objective-c): Pre-Alpha.
diff --git a/build.json b/build.json
index 7a030ea..96437d0 100644
--- a/build.json
+++ b/build.json
@@ -3,7 +3,7 @@
"#": "The public version number of the library.",
"version": {
"major": 0,
- "minor": 8,
+ "minor": 5,
"micro": 0,
"build": 0
}
diff --git a/examples/pubsub/README b/examples/pubsub/README
index b55083a..faeb622 100644
--- a/examples/pubsub/README
+++ b/examples/pubsub/README
@@ -1,3 +1,6 @@
+Experimental example code, likely to change.
+Users should not attempt to run this code till this warning is removed.
+
C++ Client implementation for Cloud Pub/Sub service
(https://developers.google.com/apis-explorer/#p/pubsub/v1beta1/).
@@ -12,19 +15,7 @@
gcloud compute instances create instance-name
--image debian-7 --scopes https://www.googleapis.com/auth/cloud-platform
-Google TLS cert is required to run the client, which can be downloaded from
-Chrome browser.
-
-To run the client from GCE:
-make pubsub_client
-GRPC_DEFAULT_SSL_ROOTS_FILE_PATH="Google TLS cert" bins/opt/pubsub_client
- --project_id="your project id"
-
-A service account credential is required to run the client from other
-environments, which can be generated as a JSON key file from
-https://console.developers.google.com/project/. To run the client with a service
-account credential:
-GRPC_DEFAULT_SSL_ROOTS_FILE_PATH="Google TLS cert" bins/opt/pubsub_client
- --project_id="your project id"
- --service_account_key_file="absolute path to the JSON key file"
+To run the client:
+make pubsub_client
+bins/opt/pubsub_client --project_id="your project id"
diff --git a/src/compiler/python_generator.cc b/src/compiler/python_generator.cc
index cdd3d8a..a93b08c 100644
--- a/src/compiler/python_generator.cc
+++ b/src/compiler/python_generator.cc
@@ -33,9 +33,11 @@
#include <cassert>
#include <cctype>
+#include <cstring>
#include <map>
#include <ostream>
#include <sstream>
+#include <vector>
#include "src/compiler/python_generator.h"
#include <google/protobuf/io/printer.h>
@@ -43,14 +45,19 @@
#include <google/protobuf/descriptor.pb.h>
#include <google/protobuf/descriptor.h>
+using google::protobuf::Descriptor;
using google::protobuf::FileDescriptor;
using google::protobuf::ServiceDescriptor;
using google::protobuf::MethodDescriptor;
using google::protobuf::io::Printer;
using google::protobuf::io::StringOutputStream;
using std::initializer_list;
+using std::make_pair;
using std::map;
+using std::pair;
using std::string;
+using std::strlen;
+using std::vector;
namespace grpc_python_generator {
namespace {
@@ -99,62 +106,81 @@
// END FORMATTING BOILERPLATE //
////////////////////////////////
-void PrintService(const ServiceDescriptor* service,
- Printer* out) {
- string doc = "<fill me in later!>";
- map<string, string> dict = ListToDict({
- "Service", service->name(),
- "Documentation", doc,
- });
- out->Print(dict, "class $Service$Service(object):\n");
- {
- IndentScope raii_class_indent(out);
- out->Print(dict, "\"\"\"$Documentation$\"\"\"\n");
- out->Print("def __init__(self):\n");
- {
- IndentScope raii_method_indent(out);
- out->Print("pass\n");
- }
- }
-}
-
-void PrintServicer(const ServiceDescriptor* service,
+bool PrintServicer(const ServiceDescriptor* service,
Printer* out) {
string doc = "<fill me in later!>";
map<string, string> dict = ListToDict({
"Service", service->name(),
"Documentation", doc,
});
- out->Print(dict, "class $Service$Servicer(object):\n");
+ out->Print(dict, "class EarlyAdopter$Service$Servicer(object):\n");
{
IndentScope raii_class_indent(out);
out->Print(dict, "\"\"\"$Documentation$\"\"\"\n");
+ out->Print("__metaclass__ = abc.ABCMeta\n");
for (int i = 0; i < service->method_count(); ++i) {
auto meth = service->method(i);
- out->Print("def $Method$(self, arg):\n", "Method", meth->name());
+ string arg_name = meth->client_streaming() ?
+ "request_iterator" : "request";
+ out->Print("@abc.abstractmethod\n");
+ out->Print("def $Method$(self, $ArgName$):\n",
+ "Method", meth->name(), "ArgName", arg_name);
{
IndentScope raii_method_indent(out);
out->Print("raise NotImplementedError()\n");
}
}
}
+ return true;
}
-void PrintStub(const ServiceDescriptor* service,
+bool PrintServer(const ServiceDescriptor* service, Printer* out) {
+ string doc = "<fill me in later!>";
+ map<string, string> dict = ListToDict({
+ "Service", service->name(),
+ "Documentation", doc,
+ });
+ out->Print(dict, "class EarlyAdopter$Service$Server(object):\n");
+ {
+ IndentScope raii_class_indent(out);
+ out->Print(dict, "\"\"\"$Documentation$\"\"\"\n");
+ out->Print("__metaclass__ = abc.ABCMeta\n");
+ out->Print("@abc.abstractmethod\n");
+ out->Print("def start(self):\n");
+ {
+ IndentScope raii_method_indent(out);
+ out->Print("raise NotImplementedError()\n");
+ }
+
+ out->Print("@abc.abstractmethod\n");
+ out->Print("def stop(self):\n");
+ {
+ IndentScope raii_method_indent(out);
+ out->Print("raise NotImplementedError()\n");
+ }
+ }
+ return true;
+}
+
+bool PrintStub(const ServiceDescriptor* service,
Printer* out) {
string doc = "<fill me in later!>";
map<string, string> dict = ListToDict({
"Service", service->name(),
"Documentation", doc,
});
- out->Print(dict, "class $Service$Stub(object):\n");
+ out->Print(dict, "class EarlyAdopter$Service$Stub(object):\n");
{
IndentScope raii_class_indent(out);
out->Print(dict, "\"\"\"$Documentation$\"\"\"\n");
+ out->Print("__metaclass__ = abc.ABCMeta\n");
for (int i = 0; i < service->method_count(); ++i) {
const MethodDescriptor* meth = service->method(i);
- auto methdict = ListToDict({"Method", meth->name()});
- out->Print(methdict, "def $Method$(self, arg):\n");
+ string arg_name = meth->client_streaming() ?
+ "request_iterator" : "request";
+ auto methdict = ListToDict({"Method", meth->name(), "ArgName", arg_name});
+ out->Print("@abc.abstractmethod\n");
+ out->Print(methdict, "def $Method$(self, $ArgName$):\n");
{
IndentScope raii_method_indent(out);
out->Print("raise NotImplementedError()\n");
@@ -162,169 +188,190 @@
out->Print(methdict, "$Method$.async = None\n");
}
}
+ return true;
}
-void PrintStubImpl(const ServiceDescriptor* service,
- Printer* out) {
- map<string, string> dict = ListToDict({
- "Service", service->name(),
- });
- out->Print(dict, "class _$Service$Stub($Service$Stub):\n");
- {
- IndentScope raii_class_indent(out);
- out->Print("def __init__(self, face_stub, default_timeout):\n");
- {
- IndentScope raii_method_indent(out);
- out->Print("self._face_stub = face_stub\n"
- "self._default_timeout = default_timeout\n"
- "stub_self = self\n");
-
- for (int i = 0; i < service->method_count(); ++i) {
- const MethodDescriptor* meth = service->method(i);
- bool server_streaming = meth->server_streaming();
- bool client_streaming = meth->client_streaming();
- std::string blocking_call, future_call;
- if (server_streaming) {
- if (client_streaming) {
- blocking_call = "stub_self._face_stub.inline_stream_in_stream_out";
- future_call = blocking_call;
- } else {
- blocking_call = "stub_self._face_stub.inline_value_in_stream_out";
- future_call = blocking_call;
- }
- } else {
- if (client_streaming) {
- blocking_call = "stub_self._face_stub.blocking_stream_in_value_out";
- future_call = "stub_self._face_stub.future_stream_in_value_out";
- } else {
- blocking_call = "stub_self._face_stub.blocking_value_in_value_out";
- future_call = "stub_self._face_stub.future_value_in_value_out";
- }
- }
- // TODO(atash): use the solution described at
- // http://stackoverflow.com/a/2982 to bind 'async' attribute
- // functions to def'd functions instead of using callable attributes.
- auto methdict = ListToDict({
- "Method", meth->name(),
- "BlockingCall", blocking_call,
- "FutureCall", future_call
- });
- out->Print(methdict, "class $Method$(object):\n");
- {
- IndentScope raii_callable_indent(out);
- out->Print("def __call__(self, arg):\n");
- {
- IndentScope raii_callable_call_indent(out);
- out->Print(methdict,
- "return $BlockingCall$(\"$Method$\", arg, "
- "stub_self._default_timeout)\n");
- }
- out->Print("def async(self, arg):\n");
- {
- IndentScope raii_callable_async_indent(out);
- out->Print(methdict,
- "return $FutureCall$(\"$Method$\", arg, "
- "stub_self._default_timeout)\n");
- }
- }
- out->Print(methdict, "self.$Method$ = $Method$()\n");
- }
- }
+bool GetModuleAndMessagePath(const Descriptor* type,
+ pair<string, string>* out) {
+ const Descriptor* path_elem_type = type;
+ vector<const Descriptor*> message_path;
+ do {
+ message_path.push_back(path_elem_type);
+ path_elem_type = path_elem_type->containing_type();
+ } while (path_elem_type != nullptr);
+ string file_name = type->file()->name();
+ string module_name;
+ static const int proto_suffix_length = strlen(".proto");
+ if (!(file_name.size() > static_cast<size_t>(proto_suffix_length) &&
+ file_name.find_last_of(".proto") == file_name.size() - 1)) {
+ return false;
}
+ module_name = file_name.substr(
+ 0, file_name.size() - proto_suffix_length) + "_pb2";
+ string package = type->file()->package();
+ string module = (package.empty() ? "" : package + ".") +
+ module_name;
+ string message_type;
+ for (auto path_iter = message_path.rbegin();
+ path_iter != message_path.rend(); ++path_iter) {
+ message_type += (*path_iter)->name() + ".";
+ }
+ message_type.pop_back();
+ *out = make_pair(module, message_type);
+ return true;
}
-void PrintStubGenerators(const ServiceDescriptor* service, Printer* out) {
- map<string, string> dict = ListToDict({
- "Service", service->name(),
- });
- // Write out a generator of linked pairs of Server/Stub
- out->Print(dict, "def mock_$Service$(servicer, default_timeout):\n");
+bool PrintServerFactory(const ServiceDescriptor* service, Printer* out) {
+ out->Print("def early_adopter_create_$Service$_server(servicer, port, "
+ "root_certificates, key_chain_pairs):\n",
+ "Service", service->name());
{
- IndentScope raii_mock_indent(out);
- out->Print("value_in_value_out = {}\n"
- "value_in_stream_out = {}\n"
- "stream_in_value_out = {}\n"
- "stream_in_stream_out = {}\n");
+ IndentScope raii_create_server_indent(out);
+ map<string, pair<string, string>> method_to_module_and_message;
+ out->Print("method_implementations = {\n");
for (int i = 0; i < service->method_count(); ++i) {
+ IndentScope raii_implementations_indent(out);
const MethodDescriptor* meth = service->method(i);
- std::string super_interface, meth_dict;
- bool server_streaming = meth->server_streaming();
- bool client_streaming = meth->client_streaming();
- if (server_streaming) {
- if (client_streaming) {
- super_interface = "InlineStreamInStreamOutMethod";
- meth_dict = "stream_in_stream_out";
- } else {
- super_interface = "InlineValueInStreamOutMethod";
- meth_dict = "value_in_stream_out";
- }
- } else {
- if (client_streaming) {
- super_interface = "InlineStreamInValueOutMethod";
- meth_dict = "stream_in_value_out";
- } else {
- super_interface = "InlineValueInValueOutMethod";
- meth_dict = "value_in_value_out";
- }
+ string meth_type =
+ string(meth->client_streaming() ? "stream" : "unary") +
+ string(meth->server_streaming() ? "_stream" : "_unary") + "_inline";
+ out->Print("\"$Method$\": utilities.$Type$(servicer.$Method$),\n",
+ "Method", meth->name(),
+ "Type", meth_type);
+ // Maintain information on the input type of the service method for later
+ // use in constructing the service assembly's activated fore link.
+ const Descriptor* input_type = meth->input_type();
+ pair<string, string> module_and_message;
+ if (!GetModuleAndMessagePath(input_type, &module_and_message)) {
+ return false;
}
- map<string, string> methdict = ListToDict({
- "Method", meth->name(),
- "SuperInterface", super_interface,
- "MethodDict", meth_dict
- });
- out->Print(
- methdict, "class $Method$(_face_interfaces.$SuperInterface$):\n");
- {
- IndentScope raii_inline_class_indent(out);
- out->Print("def service(self, request, context):\n");
- {
- IndentScope raii_inline_class_fn_indent(out);
- out->Print(methdict, "return servicer.$Method$(request)\n");
- }
- }
- out->Print(methdict, "$MethodDict$['$Method$'] = $Method$()\n");
+ method_to_module_and_message.insert(
+ make_pair(meth->name(), module_and_message));
}
- out->Print(
- "face_linked_pair = _face_testing.server_and_stub(default_timeout,"
- "inline_value_in_value_out_methods=value_in_value_out,"
- "inline_value_in_stream_out_methods=value_in_stream_out,"
- "inline_stream_in_value_out_methods=stream_in_value_out,"
- "inline_stream_in_stream_out_methods=stream_in_stream_out)\n");
- out->Print("class LinkedPair(object):\n");
- {
- IndentScope raii_linked_pair(out);
- out->Print("def __init__(self, server, stub):\n");
- {
- IndentScope raii_linked_pair_init(out);
- out->Print("self.server = server\n"
- "self.stub = stub\n");
- }
+ out->Print("}\n");
+ // Ensure that we've imported all of the relevant messages.
+ for (auto& meth_vals : method_to_module_and_message) {
+ out->Print("import $Module$\n",
+ "Module", meth_vals.second.first);
}
- out->Print(
- dict,
- "stub = _$Service$Stub(face_linked_pair.stub, default_timeout)\n");
- out->Print("return LinkedPair(None, stub)\n");
+ out->Print("request_deserializers = {\n");
+ for (auto& meth_vals : method_to_module_and_message) {
+ IndentScope raii_serializers_indent(out);
+ string full_input_type_path = meth_vals.second.first + "." +
+ meth_vals.second.second;
+ out->Print("\"$Method$\": $Type$.FromString,\n",
+ "Method", meth_vals.first,
+ "Type", full_input_type_path);
+ }
+ out->Print("}\n");
+ out->Print("response_serializers = {\n");
+ for (auto& meth_vals : method_to_module_and_message) {
+ IndentScope raii_serializers_indent(out);
+ out->Print("\"$Method$\": lambda x: x.SerializeToString(),\n",
+ "Method", meth_vals.first);
+ }
+ out->Print("}\n");
+ out->Print("link = fore.activated_fore_link(port, request_deserializers, "
+ "response_serializers, root_certificates, key_chain_pairs)\n");
+ out->Print("return implementations.assemble_service("
+ "method_implementations, link)\n");
}
+ return true;
+}
+
+bool PrintStubFactory(const ServiceDescriptor* service, Printer* out) {
+ map<string, string> dict = ListToDict({
+ "Service", service->name(),
+ });
+ out->Print(dict, "def early_adopter_create_$Service$_stub(host, port):\n");
+ {
+ IndentScope raii_create_server_indent(out);
+ map<string, pair<string, string>> method_to_module_and_message;
+ out->Print("method_implementations = {\n");
+ for (int i = 0; i < service->method_count(); ++i) {
+ IndentScope raii_implementations_indent(out);
+ const MethodDescriptor* meth = service->method(i);
+ string meth_type =
+ string(meth->client_streaming() ? "stream" : "unary") +
+ string(meth->server_streaming() ? "_stream" : "_unary") + "_inline";
+ // TODO(atash): once the expected input to assemble_dynamic_inline_stub is
+ // cleaned up, change this to the expected argument's dictionary values.
+ out->Print("\"$Method$\": utilities.$Type$(None),\n",
+ "Method", meth->name(),
+ "Type", meth_type);
+ // Maintain information on the input type of the service method for later
+ // use in constructing the service assembly's activated fore link.
+ const Descriptor* output_type = meth->output_type();
+ pair<string, string> module_and_message;
+ if (!GetModuleAndMessagePath(output_type, &module_and_message)) {
+ return false;
+ }
+ method_to_module_and_message.insert(
+ make_pair(meth->name(), module_and_message));
+ }
+ out->Print("}\n");
+ // Ensure that we've imported all of the relevant messages.
+ for (auto& meth_vals : method_to_module_and_message) {
+ out->Print("import $Module$\n",
+ "Module", meth_vals.second.first);
+ }
+ out->Print("response_deserializers = {\n");
+ for (auto& meth_vals : method_to_module_and_message) {
+ IndentScope raii_serializers_indent(out);
+ string full_output_type_path = meth_vals.second.first + "." +
+ meth_vals.second.second;
+ out->Print("\"$Method$\": $Type$.FromString,\n",
+ "Method", meth_vals.first,
+ "Type", full_output_type_path);
+ }
+ out->Print("}\n");
+ out->Print("request_serializers = {\n");
+ for (auto& meth_vals : method_to_module_and_message) {
+ IndentScope raii_serializers_indent(out);
+ out->Print("\"$Method$\": lambda x: x.SerializeToString(),\n",
+ "Method", meth_vals.first);
+ }
+ out->Print("}\n");
+ out->Print("link = rear.activated_rear_link("
+ "host, port, request_serializers, response_deserializers)\n");
+ out->Print("return implementations.assemble_dynamic_inline_stub("
+ "method_implementations, link)\n");
+ }
+ return true;
+}
+
+bool PrintPreamble(const FileDescriptor* file, Printer* out) {
+ out->Print("import abc\n");
+ out->Print("from grpc._adapter import fore\n");
+ out->Print("from grpc._adapter import rear\n");
+ out->Print("from grpc.framework.assembly import implementations\n");
+ out->Print("from grpc.framework.assembly import utilities\n");
+ return true;
}
} // namespace
-string GetServices(const FileDescriptor* file) {
+pair<bool, string> GetServices(const FileDescriptor* file) {
string output;
- StringOutputStream output_stream(&output);
- Printer out(&output_stream, '$');
- out.Print("from grpc.framework.face import demonstration as _face_testing\n");
- out.Print("from grpc.framework.face import interfaces as _face_interfaces\n");
-
- for (int i = 0; i < file->service_count(); ++i) {
- auto service = file->service(i);
- PrintService(service, &out);
- PrintServicer(service, &out);
- PrintStub(service, &out);
- PrintStubImpl(service, &out);
- PrintStubGenerators(service, &out);
+ {
+ // Scope the output stream so it closes and finalizes output to the string.
+ StringOutputStream output_stream(&output);
+ Printer out(&output_stream, '$');
+ if (!PrintPreamble(file, &out)) {
+ return make_pair(false, "");
+ }
+ for (int i = 0; i < file->service_count(); ++i) {
+ auto service = file->service(i);
+ if (!(PrintServicer(service, &out) &&
+ PrintServer(service, &out) &&
+ PrintStub(service, &out) &&
+ PrintServerFactory(service, &out) &&
+ PrintStubFactory(service, &out))) {
+ return make_pair(false, "");
+ }
+ }
}
- return output;
+ return make_pair(true, std::move(output));
}
} // namespace grpc_python_generator
diff --git a/src/compiler/python_generator.h b/src/compiler/python_generator.h
index 673ef7b..773dfa3 100644
--- a/src/compiler/python_generator.h
+++ b/src/compiler/python_generator.h
@@ -35,6 +35,7 @@
#define __GRPC_COMPILER_PYTHON_GENERATOR_H__
#include <string>
+#include <utility>
namespace google {
namespace protobuf {
@@ -44,7 +45,7 @@
namespace grpc_python_generator {
-std::string GetServices(const google::protobuf::FileDescriptor* file);
+std::pair<bool, std::string> GetServices(const google::protobuf::FileDescriptor* file);
} // namespace grpc_python_generator
diff --git a/src/compiler/python_plugin.cc b/src/compiler/python_plugin.cc
index 05c6b09..ed1e049 100644
--- a/src/compiler/python_plugin.cc
+++ b/src/compiler/python_plugin.cc
@@ -33,6 +33,7 @@
// Generates a Python gRPC service interface out of Protobuf IDL.
+#include <cstring>
#include <memory>
#include <string>
@@ -50,6 +51,7 @@
using google::protobuf::io::CodedOutputStream;
using google::protobuf::io::ZeroCopyOutputStream;
using std::string;
+using std::strlen;
class PythonGrpcGenerator : public CodeGenerator {
public:
@@ -62,7 +64,7 @@
string* error) const override {
// Get output file name.
string file_name;
- static const int proto_suffix_length = 6; // length of ".proto"
+ static const int proto_suffix_length = strlen(".proto");
if (file->name().size() > static_cast<size_t>(proto_suffix_length) &&
file->name().find_last_of(".proto") == file->name().size() - 1) {
file_name = file->name().substr(
@@ -75,9 +77,15 @@
std::unique_ptr<ZeroCopyOutputStream> output(
context->OpenForInsert(file_name, "module_scope"));
CodedOutputStream coded_out(output.get());
- string code = grpc_python_generator::GetServices(file);
- coded_out.WriteRaw(code.data(), code.size());
- return true;
+ bool success = false;
+ string code = "";
+ tie(success, code) = grpc_python_generator::GetServices(file);
+ if (success) {
+ coded_out.WriteRaw(code.data(), code.size());
+ return true;
+ } else {
+ return false;
+ }
}
};
diff --git a/src/core/security/security_context.c b/src/core/security/security_context.c
index 60064dc..0dc37fa 100644
--- a/src/core/security/security_context.c
+++ b/src/core/security/security_context.c
@@ -338,6 +338,24 @@
return ssl_create_handshaker(c->handshaker_factory, 0, NULL, handshaker);
}
+static int ssl_host_matches_name(const tsi_peer *peer,
+ const char *peer_name) {
+ char *allocated_name = NULL;
+ int r;
+
+ if (strchr(peer_name, ':') != NULL) {
+ char *ignored_port;
+ gpr_split_host_port(peer_name, &allocated_name, &ignored_port);
+ gpr_free(ignored_port);
+ peer_name = allocated_name;
+ if (!peer_name) return 0;
+ }
+
+ r = tsi_ssl_peer_matches_name(peer, peer_name);
+ gpr_free(allocated_name);
+ return r;
+}
+
static grpc_security_status ssl_check_peer(const char *peer_name,
const tsi_peer *peer) {
/* Check the ALPN. */
@@ -359,10 +377,11 @@
/* Check the peer name if specified. */
if (peer_name != NULL &&
- !tsi_ssl_peer_matches_name(peer, peer_name)) {
+ !ssl_host_matches_name(peer, peer_name)) {
gpr_log(GPR_ERROR, "Peer name %s is not in peer certificate", peer_name);
return GRPC_SECURITY_ERROR;
}
+
return GRPC_SECURITY_OK;
}
@@ -398,7 +417,7 @@
grpc_ssl_channel_security_context *c =
(grpc_ssl_channel_security_context *)ctx;
- if (tsi_ssl_peer_matches_name(&c->peer, host)) return GRPC_SECURITY_OK;
+ if (ssl_host_matches_name(&c->peer, host)) return GRPC_SECURITY_OK;
/* If the target name was overridden, then the original target_name was
'checked' transitively during the previous peer check at the end of the
diff --git a/src/core/tsi/ssl_transport_security.c b/src/core/tsi/ssl_transport_security.c
index 9ca8e6d..8446cc4 100644
--- a/src/core/tsi/ssl_transport_security.c
+++ b/src/core/tsi/ssl_transport_security.c
@@ -1094,8 +1094,9 @@
return 0;
}
name_subdomain = strchr(name, '.');
+ if (name_subdomain == NULL) return 0;
name_subdomain_length = strlen(name_subdomain);
- if (name_subdomain == NULL || name_subdomain_length < 2) return 0;
+ if (name_subdomain_length < 2) return 0;
name_subdomain++; /* Starts after the dot. */
name_subdomain_length--;
entry += 2; /* Remove *. */
diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj
index 135ce26..183c442 100644
--- a/src/csharp/Grpc.Core/Grpc.Core.csproj
+++ b/src/csharp/Grpc.Core/Grpc.Core.csproj
@@ -1,4 +1,4 @@
-<?xml version="1.0" encoding="utf-8"?>
+<?xml version="1.0" encoding="utf-8"?>
<Project DefaultTargets="Build" ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
<PropertyGroup>
<Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
@@ -33,6 +33,7 @@
<Reference Include="System" />
</ItemGroup>
<ItemGroup>
+ <Compile Include="Internal\GrpcLog.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
<Compile Include="RpcException.cs" />
<Compile Include="Calls.cs" />
@@ -65,9 +66,17 @@
<Compile Include="Utils\BenchmarkUtil.cs" />
<Compile Include="Utils\ExceptionHelper.cs" />
</ItemGroup>
+ <Choose>
+ <!-- Under Windows, automatically copy the C core library to output dir.
+ Under Monodevelop it's not supported so it has no effect. -->
+ <When Condition=" '$(Platform)' == 'AnyCPU' ">
+ <ItemGroup>
+ <Content Include="..\..\..\vsprojects\vs2013\Debug\grpc_csharp_ext.dll">
+ <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+ </Content>
+ </ItemGroup>
+ </When>
+ <Otherwise/>
+ </Choose>
<Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
- <ItemGroup>
- <Folder Include="Internal\" />
- <Folder Include="Utils\" />
- </ItemGroup>
</Project>
\ No newline at end of file
diff --git a/src/csharp/Grpc.Core/GrpcEnvironment.cs b/src/csharp/Grpc.Core/GrpcEnvironment.cs
index 0e3a0a5..d3a8da4 100644
--- a/src/csharp/Grpc.Core/GrpcEnvironment.cs
+++ b/src/csharp/Grpc.Core/GrpcEnvironment.cs
@@ -107,6 +107,7 @@
/// </summary>
private GrpcEnvironment()
{
+ GrpcLog.RedirectNativeLogs(Console.Error);
grpcsharp_init();
threadPool = new GrpcThreadPool(THREAD_POOL_SIZE);
threadPool.Start();
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
index dadc9ab..6f37b05 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
@@ -181,6 +181,7 @@
{
started = true;
halfcloseRequested = true;
+ halfclosed = true; // halfclose not confirmed yet, but it will be once finishedHandler is called.
this.readObserver = readObserver;
@@ -544,6 +545,8 @@
}
observer = readObserver;
status = finishedStatus;
+
+ ReleaseResourcesIfPossible();
}
// TODO: wrap deserialization...
diff --git a/src/csharp/Grpc.Core/Internal/GrpcLog.cs b/src/csharp/Grpc.Core/Internal/GrpcLog.cs
new file mode 100644
index 0000000..98768d0
--- /dev/null
+++ b/src/csharp/Grpc.Core/Internal/GrpcLog.cs
@@ -0,0 +1,94 @@
+#region Copyright notice and license
+
+// Copyright 2015, Google Inc.
+// All rights reserved.
+//
+// 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.
+
+#endregion
+
+using System;
+using System.Collections.Concurrent;
+using System.Diagnostics;
+using System.IO;
+using System.Runtime.InteropServices;
+using System.Threading;
+
+namespace Grpc.Core.Internal
+{
+ internal delegate void GprLogDelegate(IntPtr fileStringPtr, Int32 line, UInt64 threadId, IntPtr severityStringPtr, IntPtr msgPtr);
+
+ /// <summary>
+ /// Logs from gRPC C core library can get lost if your application is not a console app.
+ /// This class allows redirection of logs to arbitrary destination.
+ /// </summary>
+ internal static class GrpcLog
+ {
+ static object staticLock = new object();
+ static GprLogDelegate writeCallback;
+ static TextWriter dest;
+
+ [DllImport("grpc_csharp_ext.dll")]
+ static extern void grpcsharp_redirect_log(GprLogDelegate callback);
+
+ /// <summary>
+ /// Sets text writer as destination for logs from native gRPC C core library.
+ /// Only first invocation has effect.
+ /// </summary>
+ /// <param name="textWriter"></param>
+ public static void RedirectNativeLogs(TextWriter textWriter)
+ {
+ lock (staticLock)
+ {
+ if (writeCallback == null)
+ {
+ writeCallback = new GprLogDelegate(HandleWrite);
+ dest = textWriter;
+ grpcsharp_redirect_log(writeCallback);
+ }
+ }
+ }
+
+ private static void HandleWrite(IntPtr fileStringPtr, Int32 line, UInt64 threadId, IntPtr severityStringPtr, IntPtr msgPtr)
+ {
+ try
+ {
+ // TODO: DateTime format used here is different than in C core.
+ dest.WriteLine(string.Format("{0}{1} {2} {3}:{4}: {5}",
+ Marshal.PtrToStringAnsi(severityStringPtr), DateTime.Now,
+ threadId,
+ Marshal.PtrToStringAnsi(fileStringPtr),
+ line,
+ Marshal.PtrToStringAnsi(msgPtr)));
+ }
+ catch (Exception e)
+ {
+ Console.WriteLine("Caught exception in native callback " + e);
+ }
+ }
+ }
+}
diff --git a/src/csharp/Grpc.Examples/MathServiceImpl.cs b/src/csharp/Grpc.Examples/MathServiceImpl.cs
index 462fab4..76a08ce 100644
--- a/src/csharp/Grpc.Examples/MathServiceImpl.cs
+++ b/src/csharp/Grpc.Examples/MathServiceImpl.cs
@@ -127,8 +127,7 @@
public void OnCompleted()
{
- Task.Factory.StartNew(() =>
- responseObserver.OnCompleted());
+ responseObserver.OnCompleted();
}
public void OnError(Exception error)
@@ -138,13 +137,7 @@
public void OnNext(DivArgs value)
{
- // TODO: currently we need this indirection because
- // responseObserver waits for write to finish, this
- // callback is called from grpc threadpool which
- // currently only has one thread.
- // Same story for OnCompleted().
- Task.Factory.StartNew(() =>
- responseObserver.OnNext(DivInternal(value)));
+ responseObserver.OnNext(DivInternal(value));
}
}
}
diff --git a/src/csharp/Grpc.IntegrationTesting/Client.cs b/src/csharp/Grpc.IntegrationTesting/Client.cs
index 0c70744..fa1c7cd 100644
--- a/src/csharp/Grpc.IntegrationTesting/Client.cs
+++ b/src/csharp/Grpc.IntegrationTesting/Client.cs
@@ -138,7 +138,7 @@
}
}
- private void RunEmptyUnary(TestServiceGrpc.ITestServiceClient client)
+ public static void RunEmptyUnary(TestServiceGrpc.ITestServiceClient client)
{
Console.WriteLine("running empty_unary");
var response = client.EmptyCall(Empty.DefaultInstance);
@@ -146,7 +146,7 @@
Console.WriteLine("Passed!");
}
- private void RunLargeUnary(TestServiceGrpc.ITestServiceClient client)
+ public static void RunLargeUnary(TestServiceGrpc.ITestServiceClient client)
{
Console.WriteLine("running large_unary");
var request = SimpleRequest.CreateBuilder()
@@ -162,7 +162,7 @@
Console.WriteLine("Passed!");
}
- private void RunClientStreaming(TestServiceGrpc.ITestServiceClient client)
+ public static void RunClientStreaming(TestServiceGrpc.ITestServiceClient client)
{
Console.WriteLine("running client_streaming");
@@ -181,7 +181,7 @@
Console.WriteLine("Passed!");
}
- private void RunServerStreaming(TestServiceGrpc.ITestServiceClient client)
+ public static void RunServerStreaming(TestServiceGrpc.ITestServiceClient client)
{
Console.WriteLine("running server_streaming");
@@ -206,7 +206,7 @@
Console.WriteLine("Passed!");
}
- private void RunPingPong(TestServiceGrpc.ITestServiceClient client)
+ public static void RunPingPong(TestServiceGrpc.ITestServiceClient client)
{
Console.WriteLine("running ping_pong");
@@ -235,7 +235,7 @@
inputs.OnNext(StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE)
- .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(2635))
+ .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(2653))
.SetPayload(CreateZerosPayload(1828)).Build());
response = recorder.Queue.Take();
@@ -252,13 +252,15 @@
Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type);
Assert.AreEqual(58979, response.Payload.Body.Length);
+ inputs.OnCompleted();
+
recorder.Finished.Wait();
Assert.AreEqual(0, recorder.Queue.Count);
Console.WriteLine("Passed!");
}
- private void RunEmptyStream(TestServiceGrpc.ITestServiceClient client)
+ public static void RunEmptyStream(TestServiceGrpc.ITestServiceClient client)
{
Console.WriteLine("running empty_stream");
@@ -273,13 +275,13 @@
}
// This is not an official interop test, but it's useful.
- private void RunBenchmarkEmptyUnary(TestServiceGrpc.ITestServiceClient client)
+ public static void RunBenchmarkEmptyUnary(TestServiceGrpc.ITestServiceClient client)
{
BenchmarkUtil.RunBenchmark(10000, 10000,
() => { client.EmptyCall(Empty.DefaultInstance);});
}
- private Payload CreateZerosPayload(int size) {
+ private static Payload CreateZerosPayload(int size) {
return Payload.CreateBuilder().SetBody(ByteString.CopyFrom(new byte[size])).Build();
}
diff --git a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj
index 9b46a64..e66f708 100644
--- a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj
+++ b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj
@@ -47,6 +47,8 @@
<Compile Include="TestServiceGrpc.cs" />
<Compile Include="Empty.cs" />
<Compile Include="Messages.cs" />
+ <Compile Include="InteropClientServerTest.cs" />
+ <Compile Include="TestServiceImpl.cs" />
</ItemGroup>
<Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
<ItemGroup>
diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs
new file mode 100644
index 0000000..87d25b0
--- /dev/null
+++ b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs
@@ -0,0 +1,119 @@
+#region Copyright notice and license
+
+// Copyright 2015, Google Inc.
+// All rights reserved.
+//
+// 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.
+
+#endregion
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Grpc.Core;
+using Grpc.Core.Utils;
+using NUnit.Framework;
+using grpc.testing;
+
+namespace Grpc.IntegrationTesting
+{
+ /// <summary>
+ /// Runs interop tests in-process.
+ /// </summary>
+ public class InteropClientServerTest
+ {
+ string host = "localhost";
+ Server server;
+ Channel channel;
+ TestServiceGrpc.ITestServiceClient client;
+
+ [TestFixtureSetUp]
+ public void Init()
+ {
+ GrpcEnvironment.Initialize();
+
+ server = new Server();
+ server.AddServiceDefinition(TestServiceGrpc.BindService(new TestServiceImpl()));
+ int port = server.AddPort(host + ":0");
+ server.Start();
+ channel = new Channel(host + ":" + port);
+ client = TestServiceGrpc.NewStub(channel);
+ }
+
+ [TestFixtureTearDown]
+ public void Cleanup()
+ {
+ channel.Dispose();
+
+ server.ShutdownAsync().Wait();
+ GrpcEnvironment.Shutdown();
+ }
+
+ [Test]
+ public void EmptyUnary()
+ {
+ Client.RunEmptyUnary(client);
+ }
+
+ [Test]
+ public void LargeUnary()
+ {
+ Client.RunEmptyUnary(client);
+ }
+
+ [Test]
+ public void ClientStreaming()
+ {
+ Client.RunClientStreaming(client);
+ }
+
+ [Test]
+ public void ServerStreaming()
+ {
+ Client.RunServerStreaming(client);
+ }
+
+ [Test]
+ public void PingPong()
+ {
+ Client.RunPingPong(client);
+ }
+
+ [Test]
+ public void EmptyStream()
+ {
+ Client.RunEmptyStream(client);
+ }
+
+ // TODO: add cancel_after_begin
+
+ // TODO: add cancel_after_first_response
+
+ }
+}
+
diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
new file mode 100644
index 0000000..176843b
--- /dev/null
+++ b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
@@ -0,0 +1,140 @@
+#region Copyright notice and license
+
+// Copyright 2015, Google Inc.
+// All rights reserved.
+//
+// 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.
+
+#endregion
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Google.ProtocolBuffers;
+using Grpc.Core.Utils;
+
+namespace grpc.testing
+{
+ /// <summary>
+ /// Implementation of TestService server
+ /// </summary>
+ public class TestServiceImpl : TestServiceGrpc.ITestService
+ {
+ public void EmptyCall(Empty request, IObserver<Empty> responseObserver)
+ {
+ responseObserver.OnNext(Empty.DefaultInstance);
+ responseObserver.OnCompleted();
+ }
+
+ public void UnaryCall(SimpleRequest request, IObserver<SimpleResponse> responseObserver)
+ {
+ var response = SimpleResponse.CreateBuilder()
+ .SetPayload(CreateZerosPayload(request.ResponseSize)).Build();
+ //TODO: check we support ReponseType
+ responseObserver.OnNext(response);
+ responseObserver.OnCompleted();
+ }
+
+ public void StreamingOutputCall(StreamingOutputCallRequest request, IObserver<StreamingOutputCallResponse> responseObserver)
+ {
+ foreach(var responseParam in request.ResponseParametersList)
+ {
+ var response = StreamingOutputCallResponse.CreateBuilder()
+ .SetPayload(CreateZerosPayload(responseParam.Size)).Build();
+ responseObserver.OnNext(response);
+ }
+ responseObserver.OnCompleted();
+ }
+
+ public IObserver<StreamingInputCallRequest> StreamingInputCall(IObserver<StreamingInputCallResponse> responseObserver)
+ {
+ var recorder = new RecordingObserver<StreamingInputCallRequest>();
+ Task.Run(() => {
+ int sum = 0;
+ foreach(var req in recorder.ToList().Result)
+ {
+ sum += req.Payload.Body.Length;
+ }
+ var response = StreamingInputCallResponse.CreateBuilder()
+ .SetAggregatedPayloadSize(sum).Build();
+ responseObserver.OnNext(response);
+ responseObserver.OnCompleted();
+ });
+ return recorder;
+ }
+
+ public IObserver<StreamingOutputCallRequest> FullDuplexCall(IObserver<StreamingOutputCallResponse> responseObserver)
+ {
+ return new FullDuplexObserver(responseObserver);
+ }
+
+ public IObserver<StreamingOutputCallRequest> HalfDuplexCall(IObserver<StreamingOutputCallResponse> responseObserver)
+ {
+ throw new NotImplementedException();
+ }
+
+ private class FullDuplexObserver : IObserver<StreamingOutputCallRequest> {
+
+ readonly IObserver<StreamingOutputCallResponse> responseObserver;
+
+ public FullDuplexObserver(IObserver<StreamingOutputCallResponse> responseObserver)
+ {
+ this.responseObserver = responseObserver;
+ }
+
+ public void OnCompleted()
+ {
+ responseObserver.OnCompleted();
+ }
+
+ public void OnError(Exception error)
+ {
+ throw new NotImplementedException();
+ }
+
+ public void OnNext(StreamingOutputCallRequest value)
+ {
+ // TODO: this is not in order!!!
+ //Task.Factory.StartNew(() => {
+
+ foreach(var responseParam in value.ResponseParametersList)
+ {
+ var response = StreamingOutputCallResponse.CreateBuilder()
+ .SetPayload(CreateZerosPayload(responseParam.Size)).Build();
+ responseObserver.OnNext(response);
+ }
+ //});
+ }
+ }
+
+ private static Payload CreateZerosPayload(int size) {
+ return Payload.CreateBuilder().SetBody(ByteString.CopyFrom(new byte[size])).Build();
+ }
+ }
+}
+
diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c
index 18e0431..8f5a414 100644
--- a/src/csharp/ext/grpc_csharp_ext.c
+++ b/src/csharp/ext/grpc_csharp_ext.c
@@ -35,9 +35,10 @@
#include <grpc/support/port_platform.h>
#include <grpc/support/alloc.h>
-#include <grpc/grpc.h>
#include <grpc/support/log.h>
#include <grpc/support/slice.h>
+#include <grpc/support/thd.h>
+#include <grpc/grpc.h>
#include <string.h>
@@ -345,14 +346,19 @@
/* Synchronous unary call */
GPR_EXPORT void GPR_CALLTYPE
-grpcsharp_call_blocking_unary(grpc_call *call, grpc_completion_queue *dedicated_cq, callback_funcptr callback,
- const char *send_buffer, size_t send_buffer_len) {
- GPR_ASSERT(grpcsharp_call_start_unary(call, callback, send_buffer, send_buffer_len) == GRPC_CALL_OK);
+grpcsharp_call_blocking_unary(grpc_call *call,
+ grpc_completion_queue *dedicated_cq,
+ callback_funcptr callback,
+ const char *send_buffer, size_t send_buffer_len) {
+ GPR_ASSERT(grpcsharp_call_start_unary(call, callback, send_buffer,
+ send_buffer_len) == GRPC_CALL_OK);
/* TODO: we would like to use pluck, but we don't know the tag */
- GPR_ASSERT(grpcsharp_completion_queue_next_with_callback(dedicated_cq) == GRPC_OP_COMPLETE);
+ GPR_ASSERT(grpcsharp_completion_queue_next_with_callback(dedicated_cq) ==
+ GRPC_OP_COMPLETE);
grpc_completion_queue_shutdown(dedicated_cq);
- GPR_ASSERT(grpcsharp_completion_queue_next_with_callback(dedicated_cq) == GRPC_QUEUE_SHUTDOWN);
+ GPR_ASSERT(grpcsharp_completion_queue_next_with_callback(dedicated_cq) ==
+ GRPC_QUEUE_SHUTDOWN);
}
GPR_EXPORT grpc_call_error GPR_CALLTYPE
@@ -579,6 +585,25 @@
&(ctx->server_rpc_new.request_metadata), cq, ctx);
}
+/* Logging */
+
+typedef void(GPR_CALLTYPE *grpcsharp_log_func)(const char *file, gpr_int32 line,
+ gpr_uint64 thd_id,
+ const char *severity_string,
+ const char *msg);
+static grpcsharp_log_func log_func = NULL;
+
+/* Redirects gpr_log to log_func callback */
+static void grpcsharp_log_handler(gpr_log_func_args *args) {
+ log_func(args->file, args->line, gpr_thd_currentid(),
+ gpr_log_severity_string(args->severity), args->message);
+}
+
+GPR_EXPORT void GPR_CALLTYPE grpcsharp_redirect_log(grpcsharp_log_func func) {
+ GPR_ASSERT(func);
+ log_func = func;
+ gpr_set_log_function(grpcsharp_log_handler);
+}
/* For testing */
GPR_EXPORT void GPR_CALLTYPE
@@ -587,7 +612,4 @@
}
/* For testing */
-GPR_EXPORT void *GPR_CALLTYPE
-grpcsharp_test_nop(void *ptr) {
- return ptr;
-}
+GPR_EXPORT void *GPR_CALLTYPE grpcsharp_test_nop(void *ptr) { return ptr; }
diff --git a/src/node/README.md b/src/node/README.md
index 8880213..5b3de6b 100644
--- a/src/node/README.md
+++ b/src/node/README.md
@@ -4,6 +4,10 @@
Alpha : Ready for early adopters
+## Prerequisites
+
+This requires `node` to be installed. If you instead have the `nodejs` executable on Debian, you should install the [`nodejs-legacy`](https://packages.debian.org/sid/nodejs-legacy) package.
+
## Installation
First, clone this repository (NPM package coming soon). Then follow the instructions in the `INSTALL` file in the root of the repository to install the C core library that this package depends on.
diff --git a/src/node/binding.gyp b/src/node/binding.gyp
index fb4c779..5c34be2 100644
--- a/src/node/binding.gyp
+++ b/src/node/binding.gyp
@@ -7,7 +7,7 @@
"targets" : [
{
'include_dirs': [
- "<!(nodejs -e \"require('nan')\")"
+ "<!(node -e \"require('nan')\")"
],
'cflags': [
'-std=c++11',
diff --git a/src/node/ext/channel.cc b/src/node/ext/channel.cc
index 6c7a89e..bc9461d 100644
--- a/src/node/ext/channel.cc
+++ b/src/node/ext/channel.cc
@@ -103,11 +103,15 @@
grpc_channel *wrapped_channel;
// Owned by the Channel object
NanUtf8String *host = new NanUtf8String(args[0]);
+ NanUtf8String *host_override = NULL;
if (args[1]->IsUndefined()) {
wrapped_channel = grpc_channel_create(**host, NULL);
} else if (args[1]->IsObject()) {
grpc_credentials *creds = NULL;
Handle<Object> args_hash(args[1]->ToObject()->Clone());
+ if (args_hash->HasOwnProperty(NanNew(GRPC_SSL_TARGET_NAME_OVERRIDE_ARG))) {
+ host_override = new NanUtf8String(args_hash->Get(NanNew(GRPC_SSL_TARGET_NAME_OVERRIDE_ARG)));
+ }
if (args_hash->HasOwnProperty(NanNew("credentials"))) {
Handle<Value> creds_value = args_hash->Get(NanNew("credentials"));
if (!Credentials::HasInstance(creds_value)) {
@@ -155,7 +159,12 @@
} else {
return NanThrowTypeError("Channel expects a string and an object");
}
- Channel *channel = new Channel(wrapped_channel, host);
+ Channel *channel;
+ if (host_override == NULL) {
+ channel = new Channel(wrapped_channel, host);
+ } else {
+ channel = new Channel(wrapped_channel, host_override);
+ }
channel->Wrap(args.This());
NanReturnValue(args.This());
} else {
diff --git a/src/node/package.json b/src/node/package.json
index 1c44b10..e9995e7 100644
--- a/src/node/package.json
+++ b/src/node/package.json
@@ -1,10 +1,10 @@
{
"name": "grpc",
- "version": "0.2.0",
+ "version": "0.5.0",
"description": "gRPC Library for Node",
"scripts": {
- "lint": "nodejs ./node_modules/jshint/bin/jshint src test examples interop index.js",
- "test": "nodejs ./node_modules/mocha/bin/mocha && npm run-script lint"
+ "lint": "node ./node_modules/jshint/bin/jshint src test examples interop index.js",
+ "test": "node ./node_modules/mocha/bin/mocha && npm run-script lint"
},
"dependencies": {
"bindings": "^1.2.1",
diff --git a/src/php/ext/grpc/credentials.c b/src/php/ext/grpc/credentials.c
index f25e042..6d8f59f 100644
--- a/src/php/ext/grpc/credentials.c
+++ b/src/php/ext/grpc/credentials.c
@@ -94,7 +94,7 @@
* @return Credentials The new default credentials object
*/
PHP_METHOD(Credentials, createDefault) {
- grpc_credentials *creds = grpc_default_credentials_create();
+ grpc_credentials *creds = grpc_google_default_credentials_create();
zval *creds_object = grpc_php_wrap_credentials(creds);
RETURN_DESTROY_ZVAL(creds_object);
}
diff --git a/src/php/ext/grpc/event.c b/src/php/ext/grpc/event.c
index 8d39845..452c4b8 100644
--- a/src/php/ext/grpc/event.c
+++ b/src/php/ext/grpc/event.c
@@ -90,10 +90,6 @@
add_property_stringl(event_object, "data", read_string, read_len, true);
}
break;
- case GRPC_INVOKE_ACCEPTED:
- add_property_long(event_object, "data",
- (long)event->data.invoke_accepted);
- break;
case GRPC_WRITE_ACCEPTED:
add_property_long(event_object, "data", (long)event->data.write_accepted);
break;
diff --git a/src/python/README.md b/src/python/README.md
index be2f2be..0ead86b 100755
--- a/src/python/README.md
+++ b/src/python/README.md
@@ -1,9 +1,14 @@
-GRPC Python
+gRPC Python
=========
-The Python facility of GRPC.
+The Python facility of gRPC.
+Status
+-------
+
+Usable with limitations, Pre-Alpha
+
Prerequisites
-----------------------
@@ -13,8 +18,8 @@
Building from source
----------------------
-- Build the GRPC core
-E.g, from the root of the grpc [git repo](https://github.com/google/grpc)
+- Build the gRPC core from the root of the
+ [gRPC git repo](https://github.com/grpc/grpc)
```
$ make shared_c static_c
```
@@ -28,7 +33,7 @@
Testing
-----------------------
-- Use run_python.sh to run GRPC as it was installed into the virtual environment
+- Use run_python.sh to run gRPC as it was installed into the virtual environment
```
$ tools/run_tests/run_python.sh
```
diff --git a/src/python/interop/interop/methods.py b/src/python/interop/interop/methods.py
index 26c1869..6d59900 100644
--- a/src/python/interop/interop/methods.py
+++ b/src/python/interop/interop/methods.py
@@ -34,47 +34,47 @@
from interop import empty_pb2
from interop import messages_pb2
-def _empty_call(request):
+def _empty_call(request, unused_context):
return empty_pb2.Empty()
-_CLIENT_EMPTY_CALL = utilities.unary_unary_client_rpc_method(
+_CLIENT_EMPTY_CALL = utilities.unary_unary_invocation_description(
empty_pb2.Empty.SerializeToString, empty_pb2.Empty.FromString)
-_SERVER_EMPTY_CALL = utilities.unary_unary_server_rpc_method(
+_SERVER_EMPTY_CALL = utilities.unary_unary_service_description(
_empty_call, empty_pb2.Empty.FromString,
empty_pb2.Empty.SerializeToString)
-def _unary_call(request):
+def _unary_call(request, unused_context):
return messages_pb2.SimpleResponse(
payload=messages_pb2.Payload(
type=messages_pb2.COMPRESSABLE,
body=b'\x00' * request.response_size))
-_CLIENT_UNARY_CALL = utilities.unary_unary_client_rpc_method(
+_CLIENT_UNARY_CALL = utilities.unary_unary_invocation_description(
messages_pb2.SimpleRequest.SerializeToString,
messages_pb2.SimpleResponse.FromString)
-_SERVER_UNARY_CALL = utilities.unary_unary_server_rpc_method(
+_SERVER_UNARY_CALL = utilities.unary_unary_service_description(
_unary_call, messages_pb2.SimpleRequest.FromString,
messages_pb2.SimpleResponse.SerializeToString)
-def _streaming_output_call(request):
+def _streaming_output_call(request, unused_context):
for response_parameters in request.response_parameters:
yield messages_pb2.StreamingOutputCallResponse(
payload=messages_pb2.Payload(
type=request.response_type,
body=b'\x00' * response_parameters.size))
-_CLIENT_STREAMING_OUTPUT_CALL = utilities.unary_stream_client_rpc_method(
+_CLIENT_STREAMING_OUTPUT_CALL = utilities.unary_stream_invocation_description(
messages_pb2.StreamingOutputCallRequest.SerializeToString,
messages_pb2.StreamingOutputCallResponse.FromString)
-_SERVER_STREAMING_OUTPUT_CALL = utilities.unary_stream_server_rpc_method(
+_SERVER_STREAMING_OUTPUT_CALL = utilities.unary_stream_service_description(
_streaming_output_call,
messages_pb2.StreamingOutputCallRequest.FromString,
messages_pb2.StreamingOutputCallResponse.SerializeToString)
-def _streaming_input_call(request_iterator):
+def _streaming_input_call(request_iterator, unused_context):
aggregate_size = 0
for request in request_iterator:
if request.payload and request.payload.body:
@@ -82,35 +82,35 @@
return messages_pb2.StreamingInputCallResponse(
aggregated_payload_size=aggregate_size)
-_CLIENT_STREAMING_INPUT_CALL = utilities.stream_unary_client_rpc_method(
+_CLIENT_STREAMING_INPUT_CALL = utilities.stream_unary_invocation_description(
messages_pb2.StreamingInputCallRequest.SerializeToString,
messages_pb2.StreamingInputCallResponse.FromString)
-_SERVER_STREAMING_INPUT_CALL = utilities.stream_unary_server_rpc_method(
+_SERVER_STREAMING_INPUT_CALL = utilities.stream_unary_service_description(
_streaming_input_call,
messages_pb2.StreamingInputCallRequest.FromString,
messages_pb2.StreamingInputCallResponse.SerializeToString)
-def _full_duplex_call(request_iterator):
+def _full_duplex_call(request_iterator, unused_context):
for request in request_iterator:
yield messages_pb2.StreamingOutputCallResponse(
payload=messages_pb2.Payload(
type=request.payload.type,
body=b'\x00' * request.response_parameters[0].size))
-_CLIENT_FULL_DUPLEX_CALL = utilities.stream_stream_client_rpc_method(
+_CLIENT_FULL_DUPLEX_CALL = utilities.stream_stream_invocation_description(
messages_pb2.StreamingOutputCallRequest.SerializeToString,
messages_pb2.StreamingOutputCallResponse.FromString)
-_SERVER_FULL_DUPLEX_CALL = utilities.stream_stream_server_rpc_method(
+_SERVER_FULL_DUPLEX_CALL = utilities.stream_stream_service_description(
_full_duplex_call,
messages_pb2.StreamingOutputCallRequest.FromString,
messages_pb2.StreamingOutputCallResponse.SerializeToString)
# NOTE(nathaniel): Apparently this is the same as the full-duplex call?
-_CLIENT_HALF_DUPLEX_CALL = utilities.stream_stream_client_rpc_method(
+_CLIENT_HALF_DUPLEX_CALL = utilities.stream_stream_invocation_description(
messages_pb2.StreamingOutputCallRequest.SerializeToString,
messages_pb2.StreamingOutputCallResponse.FromString)
-_SERVER_HALF_DUPLEX_CALL = utilities.stream_stream_server_rpc_method(
+_SERVER_HALF_DUPLEX_CALL = utilities.stream_stream_service_description(
_full_duplex_call,
messages_pb2.StreamingOutputCallRequest.FromString,
messages_pb2.StreamingOutputCallResponse.SerializeToString)
diff --git a/src/python/src/grpc/_adapter/_c.c b/src/python/src/grpc/_adapter/_c.c
index 55b9d05..f096a55 100644
--- a/src/python/src/grpc/_adapter/_c.c
+++ b/src/python/src/grpc/_adapter/_c.c
@@ -38,6 +38,7 @@
#include "grpc/_adapter/_channel.h"
#include "grpc/_adapter/_call.h"
#include "grpc/_adapter/_server.h"
+#include "grpc/_adapter/_client_credentials.h"
#include "grpc/_adapter/_server_credentials.h"
static PyObject *init(PyObject *self) {
@@ -76,6 +77,9 @@
if (pygrpc_add_server(module) == -1) {
return;
}
+ if (pygrpc_add_client_credentials(module) == -1) {
+ return;
+ }
if (pygrpc_add_server_credentials(module) == -1) {
return;
}
diff --git a/src/python/src/grpc/_adapter/_c_test.py b/src/python/src/grpc/_adapter/_c_test.py
index d421692..44aff44 100644
--- a/src/python/src/grpc/_adapter/_c_test.py
+++ b/src/python/src/grpc/_adapter/_c_test.py
@@ -136,6 +136,29 @@
_c.shut_down()
+ def test_client_credentials(self):
+ root_certificates = b'Trust starts here. Really.'
+ private_key = b'This is a really bad private key, yo.'
+ certificate_chain = b'Trust me! Do I not look trustworty?'
+
+ _c.init()
+
+ client_credentials = _c.ClientCredentials(
+ None, None, None)
+ self.assertIsNotNone(client_credentials)
+ client_credentials = _c.ClientCredentials(
+ root_certificates, None, None)
+ self.assertIsNotNone(client_credentials)
+ client_credentials = _c.ClientCredentials(
+ None, private_key, certificate_chain)
+ self.assertIsNotNone(client_credentials)
+ client_credentials = _c.ClientCredentials(
+ root_certificates, private_key, certificate_chain)
+ self.assertIsNotNone(client_credentials)
+ del client_credentials
+
+ _c.shut_down()
+
def test_server_credentials(self):
root_certificates = b'Trust starts here. Really.'
first_private_key = b'This is a really bad private key, yo.'
diff --git a/src/python/src/grpc/_adapter/_call.c b/src/python/src/grpc/_adapter/_call.c
index 325d3d5..dca2e49 100644
--- a/src/python/src/grpc/_adapter/_call.c
+++ b/src/python/src/grpc/_adapter/_call.c
@@ -161,7 +161,7 @@
}
static const PyObject *pygrpc_call_premetadata(Call *self) {
- /* TODO(b/18702680): Actually support metadata. */
+ /* TODO(nathaniel): Metadata support. */
return pygrpc_translate_call_error(
grpc_call_server_end_initial_metadata_old(self->c_call, 0));
}
diff --git a/src/python/src/grpc/_adapter/_client_credentials.c b/src/python/src/grpc/_adapter/_client_credentials.c
new file mode 100644
index 0000000..b970c86
--- /dev/null
+++ b/src/python/src/grpc/_adapter/_client_credentials.c
@@ -0,0 +1,120 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * 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.
+ *
+ */
+
+#include "grpc/_adapter/_client_credentials.h"
+
+#include <Python.h>
+#include <grpc/grpc_security.h>
+#include <grpc/support/alloc.h>
+
+static int pygrpc_client_credentials_init(ClientCredentials *self,
+ PyObject *args, PyObject *kwds) {
+ char *root_certificates;
+ grpc_ssl_pem_key_cert_pair key_certificate_pair;
+ static char *kwlist[] = {"root_certificates", "private_key",
+ "certificate_chain", NULL};
+
+ if (!PyArg_ParseTupleAndKeywords(args, kwds, "zzz:ClientCredentials", kwlist,
+ &root_certificates,
+ &key_certificate_pair.private_key,
+ &key_certificate_pair.cert_chain)) {
+ return -1;
+ }
+
+ if (key_certificate_pair.private_key != NULL && key_certificate_pair.cert_chain != NULL) {
+ self->c_client_credentials =
+ grpc_ssl_credentials_create(root_certificates, &key_certificate_pair);
+ } else {
+ self->c_client_credentials =
+ grpc_ssl_credentials_create(root_certificates, NULL);
+ }
+}
+
+static void pygrpc_client_credentials_dealloc(ClientCredentials *self) {
+ if (self->c_client_credentials != NULL) {
+ grpc_credentials_release(self->c_client_credentials);
+ }
+ self->ob_type->tp_free((PyObject *)self);
+}
+
+PyTypeObject pygrpc_ClientCredentialsType = {
+ PyVarObject_HEAD_INIT(NULL, 0)
+ "_grpc.ClientCredencials", /*tp_name*/
+ sizeof(ClientCredentials), /*tp_basicsize*/
+ 0, /*tp_itemsize*/
+ (destructor)pygrpc_client_credentials_dealloc, /*tp_dealloc*/
+ 0, /*tp_print*/
+ 0, /*tp_getattr*/
+ 0, /*tp_setattr*/
+ 0, /*tp_compare*/
+ 0, /*tp_repr*/
+ 0, /*tp_as_number*/
+ 0, /*tp_as_sequence*/
+ 0, /*tp_as_mapping*/
+ 0, /*tp_hash */
+ 0, /*tp_call*/
+ 0, /*tp_str*/
+ 0, /*tp_getattro*/
+ 0, /*tp_setattro*/
+ 0, /*tp_as_buffer*/
+ Py_TPFLAGS_DEFAULT, /*tp_flags*/
+ "Wrapping of grpc_credentials.", /* tp_doc */
+ 0, /* tp_traverse */
+ 0, /* tp_clear */
+ 0, /* tp_richcompare */
+ 0, /* tp_weaklistoffset */
+ 0, /* tp_iter */
+ 0, /* tp_iternext */
+ 0, /* tp_methods */
+ 0, /* tp_members */
+ 0, /* tp_getset */
+ 0, /* tp_base */
+ 0, /* tp_dict */
+ 0, /* tp_descr_get */
+ 0, /* tp_descr_set */
+ 0, /* tp_dictoffset */
+ (initproc)pygrpc_client_credentials_init, /* tp_init */
+ 0, /* tp_alloc */
+ PyType_GenericNew, /* tp_new */
+};
+
+int pygrpc_add_client_credentials(PyObject *module) {
+ if (PyType_Ready(&pygrpc_ClientCredentialsType) < 0) {
+ return -1;
+ }
+ if (PyModule_AddObject(module, "ClientCredentials",
+ (PyObject *)&pygrpc_ClientCredentialsType) == -1) {
+ return -1;
+ }
+ return 0;
+}
diff --git a/src/python/src/grpc/_adapter/_client_credentials.h b/src/python/src/grpc/_adapter/_client_credentials.h
new file mode 100644
index 0000000..664dc80
--- /dev/null
+++ b/src/python/src/grpc/_adapter/_client_credentials.h
@@ -0,0 +1,48 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * 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.
+ *
+ */
+
+#ifndef _ADAPTER__CLIENT_CREDENTIALS_H_
+#define _ADAPTER__CLIENT_CREDENTIALS_H_
+
+#include <Python.h>
+#include <grpc/grpc_security.h>
+
+typedef struct {
+ PyObject_HEAD grpc_credentials *c_client_credentials;
+} ClientCredentials;
+
+PyTypeObject pygrpc_ClientCredentialsType;
+
+int pygrpc_add_client_credentials(PyObject *module);
+
+#endif /* _ADAPTER__CLIENT_CREDENTIALS_H_ */
diff --git a/src/python/src/grpc/_adapter/_low.py b/src/python/src/grpc/_adapter/_low.py
index 2ef2eb8..a24baae 100644
--- a/src/python/src/grpc/_adapter/_low.py
+++ b/src/python/src/grpc/_adapter/_low.py
@@ -52,5 +52,6 @@
Channel = _c.Channel
CompletionQueue = _c.CompletionQueue
Server = _c.Server
+ClientCredentials = _c.ClientCredentials
ServerCredentials = _c.ServerCredentials
# pylint: enable=invalid-name
diff --git a/src/python/src/grpc/early_adopter/_assembly_utilities.py b/src/python/src/grpc/early_adopter/_assembly_utilities.py
new file mode 100644
index 0000000..facfc2b
--- /dev/null
+++ b/src/python/src/grpc/early_adopter/_assembly_utilities.py
@@ -0,0 +1,168 @@
+# Copyright 2015, Google Inc.
+# All rights reserved.
+#
+# 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.
+
+import abc
+import collections
+
+# assembly_interfaces is referenced from specification in this module.
+from grpc.framework.assembly import interfaces as assembly_interfaces # pylint: disable=unused-import
+from grpc.framework.assembly import utilities as assembly_utilities
+from grpc.early_adopter import _reexport
+from grpc.early_adopter import interfaces
+
+
+# TODO(issue 726): Kill the "implementations" attribute of this in favor
+# of the same-information-less-bogusly-represented "cardinalities".
+class InvocationBreakdown(object):
+ """An intermediate representation of invocation-side views of RPC methods.
+
+ Attributes:
+ cardinalities: A dictionary from RPC method name to interfaces.Cardinality
+ value.
+ implementations: A dictionary from RPC method name to
+ assembly_interfaces.MethodImplementation describing the method.
+ request_serializers: A dictionary from RPC method name to callable
+ behavior to be used serializing request values for the RPC.
+ response_deserializers: A dictionary from RPC method name to callable
+ behavior to be used deserializing response values for the RPC.
+ """
+ __metaclass__ = abc.ABCMeta
+
+
+class _EasyInvocationBreakdown(
+ InvocationBreakdown,
+ collections.namedtuple(
+ '_EasyInvocationBreakdown',
+ ('cardinalities', 'implementations', 'request_serializers',
+ 'response_deserializers'))):
+ pass
+
+
+class ServiceBreakdown(object):
+ """An intermediate representation of service-side views of RPC methods.
+
+ Attributes:
+ implementations: A dictionary from RPC method name
+ assembly_interfaces.MethodImplementation implementing the RPC method.
+ request_deserializers: A dictionary from RPC method name to callable
+ behavior to be used deserializing request values for the RPC.
+ response_serializers: A dictionary from RPC method name to callable
+ behavior to be used serializing response values for the RPC.
+ """
+ __metaclass__ = abc.ABCMeta
+
+
+class _EasyServiceBreakdown(
+ ServiceBreakdown,
+ collections.namedtuple(
+ '_EasyServiceBreakdown',
+ ('implementations', 'request_deserializers', 'response_serializers'))):
+ pass
+
+
+def break_down_invocation(method_descriptions):
+ """Derives an InvocationBreakdown from several RPC method descriptions.
+
+ Args:
+ method_descriptions: A dictionary from RPC method name to
+ interfaces.RpcMethodInvocationDescription describing the RPCs.
+
+ Returns:
+ An InvocationBreakdown corresponding to the given method descriptions.
+ """
+ cardinalities = {}
+ implementations = {}
+ request_serializers = {}
+ response_deserializers = {}
+ for name, method_description in method_descriptions.iteritems():
+ cardinality = method_description.cardinality()
+ cardinalities[name] = cardinality
+ if cardinality is interfaces.Cardinality.UNARY_UNARY:
+ implementations[name] = assembly_utilities.unary_unary_inline(None)
+ elif cardinality is interfaces.Cardinality.UNARY_STREAM:
+ implementations[name] = assembly_utilities.unary_stream_inline(None)
+ elif cardinality is interfaces.Cardinality.STREAM_UNARY:
+ implementations[name] = assembly_utilities.stream_unary_inline(None)
+ elif cardinality is interfaces.Cardinality.STREAM_STREAM:
+ implementations[name] = assembly_utilities.stream_stream_inline(None)
+ request_serializers[name] = method_description.serialize_request
+ response_deserializers[name] = method_description.deserialize_response
+ return _EasyInvocationBreakdown(
+ cardinalities, implementations, request_serializers,
+ response_deserializers)
+
+
+def break_down_service(method_descriptions):
+ """Derives a ServiceBreakdown from several RPC method descriptions.
+
+ Args:
+ method_descriptions: A dictionary from RPC method name to
+ interfaces.RpcMethodServiceDescription describing the RPCs.
+
+ Returns:
+ A ServiceBreakdown corresponding to the given method descriptions.
+ """
+ implementations = {}
+ request_deserializers = {}
+ response_serializers = {}
+ for name, method_description in method_descriptions.iteritems():
+ cardinality = method_description.cardinality()
+ if cardinality is interfaces.Cardinality.UNARY_UNARY:
+ def service(
+ request, face_rpc_context,
+ service_behavior=method_description.service_unary_unary):
+ return service_behavior(
+ request, _reexport.rpc_context(face_rpc_context))
+ implementations[name] = assembly_utilities.unary_unary_inline(service)
+ elif cardinality is interfaces.Cardinality.UNARY_STREAM:
+ def service(
+ request, face_rpc_context,
+ service_behavior=method_description.service_unary_stream):
+ return service_behavior(
+ request, _reexport.rpc_context(face_rpc_context))
+ implementations[name] = assembly_utilities.unary_stream_inline(service)
+ elif cardinality is interfaces.Cardinality.STREAM_UNARY:
+ def service(
+ request_iterator, face_rpc_context,
+ service_behavior=method_description.service_stream_unary):
+ return service_behavior(
+ request_iterator, _reexport.rpc_context(face_rpc_context))
+ implementations[name] = assembly_utilities.stream_unary_inline(service)
+ elif cardinality is interfaces.Cardinality.STREAM_STREAM:
+ def service(
+ request_iterator, face_rpc_context,
+ service_behavior=method_description.service_stream_stream):
+ return service_behavior(
+ request_iterator, _reexport.rpc_context(face_rpc_context))
+ implementations[name] = assembly_utilities.stream_stream_inline(service)
+ request_deserializers[name] = method_description.deserialize_request
+ response_serializers[name] = method_description.serialize_response
+
+ return _EasyServiceBreakdown(
+ implementations, request_deserializers, response_serializers)
diff --git a/src/python/src/grpc/early_adopter/_face_utilities.py b/src/python/src/grpc/early_adopter/_face_utilities.py
deleted file mode 100644
index 3e37b08..0000000
--- a/src/python/src/grpc/early_adopter/_face_utilities.py
+++ /dev/null
@@ -1,178 +0,0 @@
-# Copyright 2015, Google Inc.
-# All rights reserved.
-#
-# 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.
-
-import abc
-import collections
-
-from grpc.framework.face import interfaces as face_interfaces
-
-from grpc.early_adopter import interfaces
-
-
-class _InlineUnaryUnaryMethod(face_interfaces.InlineValueInValueOutMethod):
-
- def __init__(self, unary_unary_server_rpc_method):
- self._method = unary_unary_server_rpc_method
-
- def service(self, request, context):
- """See face_interfaces.InlineValueInValueOutMethod.service for spec."""
- return self._method.service_unary_unary(request)
-
-
-class _InlineUnaryStreamMethod(face_interfaces.InlineValueInStreamOutMethod):
-
- def __init__(self, unary_stream_server_rpc_method):
- self._method = unary_stream_server_rpc_method
-
- def service(self, request, context):
- """See face_interfaces.InlineValueInStreamOutMethod.service for spec."""
- return self._method.service_unary_stream(request)
-
-
-class _InlineStreamUnaryMethod(face_interfaces.InlineStreamInValueOutMethod):
-
- def __init__(self, stream_unary_server_rpc_method):
- self._method = stream_unary_server_rpc_method
-
- def service(self, request_iterator, context):
- """See face_interfaces.InlineStreamInValueOutMethod.service for spec."""
- return self._method.service_stream_unary(request_iterator)
-
-
-class _InlineStreamStreamMethod(face_interfaces.InlineStreamInStreamOutMethod):
-
- def __init__(self, stream_stream_server_rpc_method):
- self._method = stream_stream_server_rpc_method
-
- def service(self, request_iterator, context):
- """See face_interfaces.InlineStreamInStreamOutMethod.service for spec."""
- return self._method.service_stream_stream(request_iterator)
-
-
-class ClientBreakdown(object):
- """An intermediate representation of invocation-side views of RPC methods.
-
- Attributes:
- request_serializers: A dictionary from RPC method name to callable
- behavior to be used serializing request values for the RPC.
- response_deserializers: A dictionary from RPC method name to callable
- behavior to be used deserializing response values for the RPC.
- """
- __metaclass__ = abc.ABCMeta
-
-
-class _EasyClientBreakdown(
- ClientBreakdown,
- collections.namedtuple(
- '_EasyClientBreakdown',
- ('request_serializers', 'response_deserializers'))):
- pass
-
-
-class ServerBreakdown(object):
- """An intermediate representation of implementations of RPC methods.
-
- Attributes:
- unary_unary_methods: A dictionary from RPC method name to callable
- behavior implementing the RPC method for unary-unary RPC methods.
- unary_stream_methods: A dictionary from RPC method name to callable
- behavior implementing the RPC method for unary-stream RPC methods.
- stream_unary_methods: A dictionary from RPC method name to callable
- behavior implementing the RPC method for stream-unary RPC methods.
- stream_stream_methods: A dictionary from RPC method name to callable
- behavior implementing the RPC method for stream-stream RPC methods.
- request_deserializers: A dictionary from RPC method name to callable
- behavior to be used deserializing request values for the RPC.
- response_serializers: A dictionary from RPC method name to callable
- behavior to be used serializing response values for the RPC.
- """
- __metaclass__ = abc.ABCMeta
-
-
-
-class _EasyServerBreakdown(
- ServerBreakdown,
- collections.namedtuple(
- '_EasyServerBreakdown',
- ('unary_unary_methods', 'unary_stream_methods', 'stream_unary_methods',
- 'stream_stream_methods', 'request_deserializers',
- 'response_serializers'))):
- pass
-
-
-def client_break_down(methods):
- """Derives a ClientBreakdown from several interfaces.ClientRpcMethods.
-
- Args:
- methods: A dictionary from RPC mthod name to
- interfaces.ClientRpcMethod object describing the RPCs.
-
- Returns:
- A ClientBreakdown corresponding to the given methods.
- """
- request_serializers = {}
- response_deserializers = {}
- for name, method in methods.iteritems():
- request_serializers[name] = method.serialize_request
- response_deserializers[name] = method.deserialize_response
- return _EasyClientBreakdown(request_serializers, response_deserializers)
-
-
-def server_break_down(methods):
- """Derives a ServerBreakdown from several interfaces.ServerRpcMethods.
-
- Args:
- methods: A dictionary from RPC mthod name to
- interfaces.ServerRpcMethod object describing the RPCs.
-
- Returns:
- A ServerBreakdown corresponding to the given methods.
- """
- unary_unary = {}
- unary_stream = {}
- stream_unary = {}
- stream_stream = {}
- request_deserializers = {}
- response_serializers = {}
- for name, method in methods.iteritems():
- cardinality = method.cardinality()
- if cardinality is interfaces.Cardinality.UNARY_UNARY:
- unary_unary[name] = _InlineUnaryUnaryMethod(method)
- elif cardinality is interfaces.Cardinality.UNARY_STREAM:
- unary_stream[name] = _InlineUnaryStreamMethod(method)
- elif cardinality is interfaces.Cardinality.STREAM_UNARY:
- stream_unary[name] = _InlineStreamUnaryMethod(method)
- elif cardinality is interfaces.Cardinality.STREAM_STREAM:
- stream_stream[name] = _InlineStreamStreamMethod(method)
- request_deserializers[name] = method.deserialize_request
- response_serializers[name] = method.serialize_response
-
- return _EasyServerBreakdown(
- unary_unary, unary_stream, stream_unary, stream_stream,
- request_deserializers, response_serializers)
diff --git a/src/python/src/grpc/early_adopter/_reexport.py b/src/python/src/grpc/early_adopter/_reexport.py
new file mode 100644
index 0000000..35855bc
--- /dev/null
+++ b/src/python/src/grpc/early_adopter/_reexport.py
@@ -0,0 +1,207 @@
+# Copyright 2015, Google Inc.
+# All rights reserved.
+#
+# 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.
+
+import abc
+import collections
+
+from grpc.framework.face import exceptions as face_exceptions
+from grpc.framework.face import interfaces as face_interfaces
+from grpc.framework.foundation import future
+from grpc.early_adopter import exceptions
+from grpc.early_adopter import interfaces
+
+_ABORTION_REEXPORT = {
+ face_interfaces.Abortion.CANCELLED: interfaces.Abortion.CANCELLED,
+ face_interfaces.Abortion.EXPIRED: interfaces.Abortion.EXPIRED,
+ face_interfaces.Abortion.NETWORK_FAILURE:
+ interfaces.Abortion.NETWORK_FAILURE,
+ face_interfaces.Abortion.SERVICED_FAILURE:
+ interfaces.Abortion.SERVICED_FAILURE,
+ face_interfaces.Abortion.SERVICER_FAILURE:
+ interfaces.Abortion.SERVICER_FAILURE,
+}
+
+
+class _RpcError(exceptions.RpcError):
+ pass
+
+
+def _reexport_error(face_rpc_error):
+ if isinstance(face_rpc_error, face_exceptions.CancellationError):
+ return exceptions.CancellationError()
+ elif isinstance(face_rpc_error, face_exceptions.ExpirationError):
+ return exceptions.ExpirationError()
+ else:
+ return _RpcError()
+
+
+def _as_face_abortion_callback(abortion_callback):
+ def face_abortion_callback(face_abortion):
+ abortion_callback(_ABORTION_REEXPORT[face_abortion])
+ return face_abortion_callback
+
+
+class _ReexportedFuture(future.Future):
+
+ def __init__(self, face_future):
+ self._face_future = face_future
+
+ def cancel(self):
+ return self._face_future.cancel()
+
+ def cancelled(self):
+ return self._face_future.cancelled()
+
+ def running(self):
+ return self._face_future.running()
+
+ def done(self):
+ return self._face_future.done()
+
+ def result(self, timeout=None):
+ try:
+ return self._face_future.result(timeout=timeout)
+ except face_exceptions.RpcError as e:
+ raise _reexport_error(e)
+
+ def exception(self, timeout=None):
+ face_error = self._face_future.exception(timeout=timeout)
+ return None if face_error is None else _reexport_error(face_error)
+
+ def traceback(self, timeout=None):
+ return self._face_future.traceback(timeout=timeout)
+
+ def add_done_callback(self, fn):
+ self._face_future.add_done_callback(lambda unused_face_future: fn(self))
+
+
+def _call_reexporting_errors(behavior, *args, **kwargs):
+ try:
+ return behavior(*args, **kwargs)
+ except face_exceptions.RpcError as e:
+ raise _reexport_error(e)
+
+
+def _reexported_future(face_future):
+ return _ReexportedFuture(face_future)
+
+
+class _CancellableIterator(interfaces.CancellableIterator):
+
+ def __init__(self, face_cancellable_iterator):
+ self._face_cancellable_iterator = face_cancellable_iterator
+
+ def __iter__(self):
+ return self
+
+ def next(self):
+ return _call_reexporting_errors(self._face_cancellable_iterator.next)
+
+ def cancel(self):
+ self._face_cancellable_iterator.cancel()
+
+
+class _RpcContext(interfaces.RpcContext):
+
+ def __init__(self, face_rpc_context):
+ self._face_rpc_context = face_rpc_context
+
+ def is_active(self):
+ return self._face_rpc_context.is_active()
+
+ def time_remaining(self):
+ return self._face_rpc_context.time_remaining()
+
+ def add_abortion_callback(self, abortion_callback):
+ self._face_rpc_context.add_abortion_callback(
+ _as_face_abortion_callback(abortion_callback))
+
+
+class _UnaryUnarySyncAsync(interfaces.UnaryUnarySyncAsync):
+
+ def __init__(self, face_unary_unary_sync_async):
+ self._underlying = face_unary_unary_sync_async
+
+ def __call__(self, request, timeout):
+ return _call_reexporting_errors(
+ self._underlying, request, timeout)
+
+ def async(self, request, timeout):
+ return _ReexportedFuture(self._underlying.async(request, timeout))
+
+
+class _StreamUnarySyncAsync(interfaces.StreamUnarySyncAsync):
+
+ def __init__(self, face_stream_unary_sync_async):
+ self._underlying = face_stream_unary_sync_async
+
+ def __call__(self, request_iterator, timeout):
+ return _call_reexporting_errors(
+ self._underlying, request_iterator, timeout)
+
+ def async(self, request_iterator, timeout):
+ return _ReexportedFuture(self._underlying.async(request_iterator, timeout))
+
+
+class _Stub(interfaces.Stub):
+
+ def __init__(self, assembly_stub, cardinalities):
+ self._assembly_stub = assembly_stub
+ self._cardinalities = cardinalities
+
+ def __enter__(self):
+ self._assembly_stub.__enter__()
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self._assembly_stub.__exit__(exc_type, exc_val, exc_tb)
+ return False
+
+ def __getattr__(self, attr):
+ underlying_attr = self._assembly_stub.__getattr__(attr)
+ cardinality = self._cardinalities.get(attr)
+ if cardinality is interfaces.Cardinality.UNARY_UNARY:
+ return _UnaryUnarySyncAsync(underlying_attr)
+ elif cardinality is interfaces.Cardinality.UNARY_STREAM:
+ return lambda request, timeout: _CancellableIterator(
+ underlying_attr(request, timeout))
+ elif cardinality is interfaces.Cardinality.STREAM_UNARY:
+ return _StreamUnarySyncAsync(underlying_attr)
+ elif cardinality is interfaces.Cardinality.STREAM_STREAM:
+ return lambda request_iterator, timeout: _CancellableIterator(
+ underlying_attr(request_iterator, timeout))
+ else:
+ raise AttributeError(attr)
+
+def rpc_context(face_rpc_context):
+ return _RpcContext(face_rpc_context)
+
+
+def stub(assembly_stub, cardinalities):
+ return _Stub(assembly_stub, cardinalities)
diff --git a/src/python/src/grpc/early_adopter/exceptions.py b/src/python/src/grpc/early_adopter/exceptions.py
new file mode 100644
index 0000000..5234d3b
--- /dev/null
+++ b/src/python/src/grpc/early_adopter/exceptions.py
@@ -0,0 +1,48 @@
+# Copyright 2015, Google Inc.
+# All rights reserved.
+#
+# 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.
+
+"""Exceptions raised by GRPC.
+
+Only GRPC should instantiate and raise these exceptions.
+"""
+
+import abc
+
+
+class RpcError(Exception):
+ """Common super type for all exceptions raised by GRPC."""
+ __metaclass__ = abc.ABCMeta
+
+
+class CancellationError(RpcError):
+ """Indicates that an RPC has been cancelled."""
+
+
+class ExpirationError(RpcError):
+ """Indicates that an RPC has expired ("timed out")."""
diff --git a/src/python/src/grpc/early_adopter/implementations.py b/src/python/src/grpc/early_adopter/implementations.py
index 1d76d0f..241ed7d 100644
--- a/src/python/src/grpc/early_adopter/implementations.py
+++ b/src/python/src/grpc/early_adopter/implementations.py
@@ -31,15 +31,12 @@
import threading
-from grpc._adapter import fore
-from grpc.framework.base.packets import implementations as _tickets_implementations
-from grpc.framework.face import implementations as _face_implementations
-from grpc.framework.foundation import logging_pool
-from grpc.early_adopter import _face_utilities
+from grpc._adapter import fore as _fore
+from grpc._adapter import rear as _rear
+from grpc.early_adopter import _assembly_utilities
+from grpc.early_adopter import _reexport
from grpc.early_adopter import interfaces
-
-_MEGA_TIMEOUT = 60 * 60 * 24
-_THREAD_POOL_SIZE = 80
+from grpc.framework.assembly import implementations as _assembly_implementations
class _Server(interfaces.Server):
@@ -48,63 +45,120 @@
self._lock = threading.Lock()
self._breakdown = breakdown
self._port = port
- self._private_key = private_key
- self._certificate_chain = certificate_chain
+ if private_key is None or certificate_chain is None:
+ self._key_chain_pairs = ()
+ else:
+ self._key_chain_pairs = ((private_key, certificate_chain),)
- self._pool = None
self._fore_link = None
- self._back = None
+ self._server = None
- def start(self):
- """See interfaces.Server.start for specification."""
+ def _start(self):
with self._lock:
- if self._pool is None:
- self._pool = logging_pool.pool(_THREAD_POOL_SIZE)
- servicer = _face_implementations.servicer(
- self._pool,
- inline_value_in_value_out_methods=self._breakdown.unary_unary_methods,
- inline_value_in_stream_out_methods=self._breakdown.unary_stream_methods,
- inline_stream_in_value_out_methods=self._breakdown.stream_unary_methods,
- inline_stream_in_stream_out_methods=self._breakdown.stream_stream_methods)
- self._fore_link = fore.ForeLink(
- self._pool, self._breakdown.request_deserializers,
- self._breakdown.response_serializers, None,
- ((self._private_key, self._certificate_chain),), port=self._port)
- self._fore_link.start()
- port = self._fore_link.port()
- self._back = _tickets_implementations.back(
- servicer, self._pool, self._pool, self._pool, _MEGA_TIMEOUT,
- _MEGA_TIMEOUT)
- self._fore_link.join_rear_link(self._back)
- self._back.join_fore_link(self._fore_link)
- return port
+ if self._server is None:
+ self._fore_link = _fore.activated_fore_link(
+ self._port, self._breakdown.request_deserializers,
+ self._breakdown.response_serializers, None, self._key_chain_pairs)
+
+ self._server = _assembly_implementations.assemble_service(
+ self._breakdown.implementations, self._fore_link)
+ self._server.start()
else:
raise ValueError('Server currently running!')
- def stop(self):
- """See interfaces.Server.stop for specification."""
+ def _stop(self):
with self._lock:
- if self._pool is None:
+ if self._server is None:
raise ValueError('Server not running!')
else:
- self._fore_link.stop()
- self._pool.shutdown(wait=True)
- self._pool = None
+ self._server.stop()
+ self._server = None
+ self._fore_link = None
+
+ def __enter__(self):
+ self._start()
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self._stop()
+ return False
+
+ def start(self):
+ self._start()
+
+ def stop(self):
+ self._stop()
+
+ def port(self):
+ with self._lock:
+ return self._fore_link.port()
+
+def _build_stub(
+ methods, host, port, root_certificates, private_key, certificate_chain):
+ breakdown = _assembly_utilities.break_down_invocation(methods)
+ # TODO(nathaniel): pass security values.
+ activated_rear_link = _rear.activated_rear_link(
+ host, port, breakdown.request_serializers,
+ breakdown.response_deserializers)
+ assembly_stub = _assembly_implementations.assemble_dynamic_inline_stub(
+ breakdown.implementations, activated_rear_link)
+ return _reexport.stub(assembly_stub, breakdown.cardinalities)
def _build_server(methods, port, private_key, certificate_chain):
- breakdown = _face_utilities.server_break_down(methods)
+ breakdown = _assembly_utilities.break_down_service(methods)
return _Server(breakdown, port, private_key, certificate_chain)
+def insecure_stub(methods, host, port):
+ """Constructs an insecure interfaces.Stub.
+
+ Args:
+ methods: A dictionary from RPC method name to
+ interfaces.RpcMethodInvocationDescription describing the RPCs to be
+ supported by the created stub.
+ host: The host to which to connect for RPC service.
+ port: The port to which to connect for RPC service.
+
+ Returns:
+ An interfaces.Stub affording RPC invocation.
+ """
+ return _build_stub(methods, host, port, None, None, None)
+
+
+def secure_stub(
+ methods, host, port, root_certificates, private_key, certificate_chain):
+ """Constructs an insecure interfaces.Stub.
+
+ Args:
+ methods: A dictionary from RPC method name to
+ interfaces.RpcMethodInvocationDescription describing the RPCs to be
+ supported by the created stub.
+ host: The host to which to connect for RPC service.
+ port: The port to which to connect for RPC service.
+ root_certificates: The PEM-encoded root certificates or None to ask for
+ them to be retrieved from a default location.
+ private_key: The PEM-encoded private key to use or None if no private key
+ should be used.
+ certificate_chain: The PEM-encoded certificate chain to use or None if no
+ certificate chain should be used.
+
+ Returns:
+ An interfaces.Stub affording RPC invocation.
+ """
+ return _build_stub(
+ methods, host, port, root_certificates, private_key, certificate_chain)
+
+
def insecure_server(methods, port):
"""Constructs an insecure interfaces.Server.
Args:
methods: A dictionary from RPC method name to
- interfaces.ServerRpcMethod object describing the RPCs to
+ interfaces.RpcMethodServiceDescription describing the RPCs to
be serviced by the created server.
- port: The port on which to serve.
+ port: The desired port on which to serve or zero to ask for a port to
+ be automatically selected.
Returns:
An interfaces.Server that will run with no security and
@@ -118,9 +172,10 @@
Args:
methods: A dictionary from RPC method name to
- interfaces.ServerRpcMethod object describing the RPCs to
+ interfaces.RpcMethodServiceDescription describing the RPCs to
be serviced by the created server.
- port: The port on which to serve.
+ port: The port on which to serve or zero to ask for a port to be
+ automatically selected.
private_key: A pem-encoded private key.
certificate_chain: A pem-encoded certificate chain.
diff --git a/src/python/src/grpc/early_adopter/implementations_test.py b/src/python/src/grpc/early_adopter/implementations_test.py
new file mode 100644
index 0000000..9ef06c3
--- /dev/null
+++ b/src/python/src/grpc/early_adopter/implementations_test.py
@@ -0,0 +1,176 @@
+# Copyright 2015, Google Inc.
+# All rights reserved.
+#
+# 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.
+
+# TODO(nathaniel): Expand this test coverage.
+
+"""Test of the GRPC-backed ForeLink and RearLink."""
+
+import unittest
+
+from grpc.early_adopter import implementations
+from grpc.early_adopter import utilities
+from grpc._junkdrawer import math_pb2
+
+DIV = 'Div'
+DIV_MANY = 'DivMany'
+FIB = 'Fib'
+SUM = 'Sum'
+
+def _fibbonacci(limit):
+ left, right = 0, 1
+ for _ in xrange(limit):
+ yield left
+ left, right = right, left + right
+
+
+def _div(request, unused_context):
+ return math_pb2.DivReply(
+ quotient=request.dividend / request.divisor,
+ remainder=request.dividend % request.divisor)
+
+
+def _div_many(request_iterator, unused_context):
+ for request in request_iterator:
+ yield math_pb2.DivReply(
+ quotient=request.dividend / request.divisor,
+ remainder=request.dividend % request.divisor)
+
+
+def _fib(request, unused_context):
+ for number in _fibbonacci(request.limit):
+ yield math_pb2.Num(num=number)
+
+
+def _sum(request_iterator, unused_context):
+ accumulation = 0
+ for request in request_iterator:
+ accumulation += request.num
+ return math_pb2.Num(num=accumulation)
+
+
+_INVOCATION_DESCRIPTIONS = {
+ DIV: utilities.unary_unary_invocation_description(
+ math_pb2.DivArgs.SerializeToString, math_pb2.DivReply.FromString),
+ DIV_MANY: utilities.stream_stream_invocation_description(
+ math_pb2.DivArgs.SerializeToString, math_pb2.DivReply.FromString),
+ FIB: utilities.unary_stream_invocation_description(
+ math_pb2.FibArgs.SerializeToString, math_pb2.Num.FromString),
+ SUM: utilities.stream_unary_invocation_description(
+ math_pb2.Num.SerializeToString, math_pb2.Num.FromString),
+}
+
+_SERVICE_DESCRIPTIONS = {
+ DIV: utilities.unary_unary_service_description(
+ _div, math_pb2.DivArgs.FromString,
+ math_pb2.DivReply.SerializeToString),
+ DIV_MANY: utilities.stream_stream_service_description(
+ _div_many, math_pb2.DivArgs.FromString,
+ math_pb2.DivReply.SerializeToString),
+ FIB: utilities.unary_stream_service_description(
+ _fib, math_pb2.FibArgs.FromString, math_pb2.Num.SerializeToString),
+ SUM: utilities.stream_unary_service_description(
+ _sum, math_pb2.Num.FromString, math_pb2.Num.SerializeToString),
+}
+
+_TIMEOUT = 3
+
+
+class EarlyAdopterImplementationsTest(unittest.TestCase):
+
+ def setUp(self):
+ self.server = implementations.insecure_server(_SERVICE_DESCRIPTIONS, 0)
+ self.server.start()
+ port = self.server.port()
+ self.stub = implementations.insecure_stub(_INVOCATION_DESCRIPTIONS, 'localhost', port)
+
+ def tearDown(self):
+ self.server.stop()
+
+ def testUpAndDown(self):
+ with self.stub:
+ pass
+
+ def testUnaryUnary(self):
+ divisor = 59
+ dividend = 973
+ expected_quotient = dividend / divisor
+ expected_remainder = dividend % divisor
+
+ with self.stub:
+ response = self.stub.Div(
+ math_pb2.DivArgs(divisor=divisor, dividend=dividend), _TIMEOUT)
+ self.assertEqual(expected_quotient, response.quotient)
+ self.assertEqual(expected_remainder, response.remainder)
+
+ def testUnaryStream(self):
+ stream_length = 43
+
+ with self.stub:
+ response_iterator = self.stub.Fib(
+ math_pb2.FibArgs(limit=stream_length), _TIMEOUT)
+ numbers = tuple(response.num for response in response_iterator)
+ for early, middle, later in zip(numbers, numbers[:1], numbers[:2]):
+ self.assertEqual(early + middle, later)
+ self.assertEqual(stream_length, len(numbers))
+
+ def testStreamUnary(self):
+ stream_length = 127
+
+ with self.stub:
+ response_future = self.stub.Sum.async(
+ (math_pb2.Num(num=index) for index in range(stream_length)),
+ _TIMEOUT)
+ self.assertEqual(
+ (stream_length * (stream_length - 1)) / 2,
+ response_future.result().num)
+
+ def testStreamStream(self):
+ stream_length = 179
+ divisor_offset = 71
+ dividend_offset = 1763
+
+ with self.stub:
+ response_iterator = self.stub.DivMany(
+ (math_pb2.DivArgs(
+ divisor=divisor_offset + index,
+ dividend=dividend_offset + index)
+ for index in range(stream_length)),
+ _TIMEOUT)
+ for index, response in enumerate(response_iterator):
+ self.assertEqual(
+ (dividend_offset + index) / (divisor_offset + index),
+ response.quotient)
+ self.assertEqual(
+ (dividend_offset + index) % (divisor_offset + index),
+ response.remainder)
+ self.assertEqual(stream_length, index + 1)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/src/python/src/grpc/early_adopter/interfaces.py b/src/python/src/grpc/early_adopter/interfaces.py
index 0ec371f..b733873 100644
--- a/src/python/src/grpc/early_adopter/interfaces.py
+++ b/src/python/src/grpc/early_adopter/interfaces.py
@@ -32,6 +32,11 @@
import abc
import enum
+# exceptions is referenced from specification in this module.
+from grpc.early_adopter import exceptions # pylint: disable=unused-import
+from grpc.framework.foundation import activated
+from grpc.framework.foundation import future
+
@enum.unique
class Cardinality(enum.Enum):
@@ -43,24 +48,166 @@
STREAM_STREAM = 'request-streaming/response-streaming'
-class RpcMethod(object):
- """A type for the common aspects of RPC method specifications."""
+@enum.unique
+class Abortion(enum.Enum):
+ """Categories of RPC abortion."""
+
+ CANCELLED = 'cancelled'
+ EXPIRED = 'expired'
+ NETWORK_FAILURE = 'network failure'
+ SERVICED_FAILURE = 'serviced failure'
+ SERVICER_FAILURE = 'servicer failure'
+
+
+class CancellableIterator(object):
+ """Implements the Iterator protocol and affords a cancel method."""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
- def cardinality(self):
- """Identifies the cardinality of this RpcMethod.
+ def __iter__(self):
+ """Returns the self object in accordance with the Iterator protocol."""
+ raise NotImplementedError()
+ @abc.abstractmethod
+ def next(self):
+ """Returns a value or raises StopIteration per the Iterator protocol."""
+ raise NotImplementedError()
+
+ @abc.abstractmethod
+ def cancel(self):
+ """Requests cancellation of whatever computation underlies this iterator."""
+ raise NotImplementedError()
+
+
+class RpcContext(object):
+ """Provides RPC-related information and action."""
+ __metaclass__ = abc.ABCMeta
+
+ @abc.abstractmethod
+ def is_active(self):
+ """Describes whether the RPC is active or has terminated."""
+ raise NotImplementedError()
+
+ @abc.abstractmethod
+ def time_remaining(self):
+ """Describes the length of allowed time remaining for the RPC.
Returns:
- A Cardinality value identifying whether or not this
- RpcMethod is request-unary or request-streaming and
- whether or not it is response-unary or
- response-streaming.
+ A nonnegative float indicating the length of allowed time in seconds
+ remaining for the RPC to complete before it is considered to have timed
+ out.
+ """
+ raise NotImplementedError()
+
+ @abc.abstractmethod
+ def add_abortion_callback(self, abortion_callback):
+ """Registers a callback to be called if the RPC is aborted.
+ Args:
+ abortion_callback: A callable to be called and passed an Abortion value
+ in the event of RPC abortion.
"""
raise NotImplementedError()
-class ClientRpcMethod(RpcMethod):
+class UnaryUnarySyncAsync(object):
+ """Affords invoking a unary-unary RPC synchronously or asynchronously.
+ Values implementing this interface are directly callable and present an
+ "async" method. Both calls take a request value and a numeric timeout.
+ Direct invocation of a value of this type invokes its associated RPC and
+ blocks until the RPC's response is available. Calling the "async" method
+ of a value of this type invokes its associated RPC and immediately returns a
+ future.Future bound to the asynchronous execution of the RPC.
+ """
+ __metaclass__ = abc.ABCMeta
+
+ @abc.abstractmethod
+ def __call__(self, request, timeout):
+ """Synchronously invokes the underlying RPC.
+ Args:
+ request: The request value for the RPC.
+ timeout: A duration of time in seconds to allow for the RPC.
+ Returns:
+ The response value for the RPC.
+ Raises:
+ exceptions.RpcError: Indicating that the RPC was aborted.
+ """
+ raise NotImplementedError()
+
+ @abc.abstractmethod
+ def async(self, request, timeout):
+ """Asynchronously invokes the underlying RPC.
+ Args:
+ request: The request value for the RPC.
+ timeout: A duration of time in seconds to allow for the RPC.
+ Returns:
+ A future.Future representing the RPC. In the event of RPC completion, the
+ returned Future's result value will be the response value of the RPC.
+ In the event of RPC abortion, the returned Future's exception value
+ will be an exceptions.RpcError.
+ """
+ raise NotImplementedError()
+
+
+class StreamUnarySyncAsync(object):
+ """Affords invoking a stream-unary RPC synchronously or asynchronously.
+ Values implementing this interface are directly callable and present an
+ "async" method. Both calls take an iterator of request values and a numeric
+ timeout. Direct invocation of a value of this type invokes its associated RPC
+ and blocks until the RPC's response is available. Calling the "async" method
+ of a value of this type invokes its associated RPC and immediately returns a
+ future.Future bound to the asynchronous execution of the RPC.
+ """
+ __metaclass__ = abc.ABCMeta
+
+ @abc.abstractmethod
+ def __call__(self, request_iterator, timeout):
+ """Synchronously invokes the underlying RPC.
+
+ Args:
+ request_iterator: An iterator that yields request values for the RPC.
+ timeout: A duration of time in seconds to allow for the RPC.
+
+ Returns:
+ The response value for the RPC.
+
+ Raises:
+ exceptions.RpcError: Indicating that the RPC was aborted.
+ """
+ raise NotImplementedError()
+
+ @abc.abstractmethod
+ def async(self, request_iterator, timeout):
+ """Asynchronously invokes the underlying RPC.
+
+ Args:
+ request_iterator: An iterator that yields request values for the RPC.
+ timeout: A duration of time in seconds to allow for the RPC.
+
+ Returns:
+ A future.Future representing the RPC. In the event of RPC completion, the
+ returned Future's result value will be the response value of the RPC.
+ In the event of RPC abortion, the returned Future's exception value
+ will be an exceptions.RpcError.
+ """
+ raise NotImplementedError()
+
+
+class RpcMethodDescription(object):
+ """A type for the common aspects of RPC method descriptions."""
+ __metaclass__ = abc.ABCMeta
+
+ @abc.abstractmethod
+ def cardinality(self):
+ """Identifies the cardinality of this RpcMethodDescription.
+
+ Returns:
+ A Cardinality value identifying whether or not this
+ RpcMethodDescription is request-unary or request-streaming and
+ whether or not it is response-unary or response-streaming.
+ """
+ raise NotImplementedError()
+
+
+class RpcMethodInvocationDescription(RpcMethodDescription):
"""Invocation-side description of an RPC method."""
__metaclass__ = abc.ABCMeta
@@ -69,7 +216,8 @@
"""Serializes a request value.
Args:
- request: A request value appropriate for this RpcMethod.
+ request: A request value appropriate for the RPC method described by this
+ RpcMethodInvocationDescription.
Returns:
The serialization of the given request value as a
@@ -82,9 +230,9 @@
"""Deserializes a response value.
Args:
- serialized_response: A bytestring that is the
- serialization of a response value appropriate for this
- RpcMethod.
+ serialized_response: A bytestring that is the serialization of a response
+ value appropriate for the RPC method described by this
+ RpcMethodInvocationDescription.
Returns:
A response value corresponding to the given bytestring.
@@ -92,7 +240,7 @@
raise NotImplementedError()
-class ServerRpcMethod(RpcMethod):
+class RpcMethodServiceDescription(RpcMethodDescription):
"""Service-side description of an RPC method."""
__metaclass__ = abc.ABCMeta
@@ -101,9 +249,9 @@
"""Deserializes a request value.
Args:
- serialized_request: A bytestring that is the
- serialization of a request value appropriate for this
- RpcMethod.
+ serialized_request: A bytestring that is the serialization of a request
+ value appropriate for the RPC method described by this
+ RpcMethodServiceDescription.
Returns:
A request value corresponding to the given bytestring.
@@ -115,7 +263,8 @@
"""Serializes a response value.
Args:
- response: A response value appropriate for this RpcMethod.
+ response: A response value appropriate for the RPC method described by
+ this RpcMethodServiceDescription.
Returns:
The serialization of the given response value as a
@@ -124,80 +273,116 @@
raise NotImplementedError()
@abc.abstractmethod
- def service_unary_unary(self, request):
+ def service_unary_unary(self, request, context):
"""Carries out this RPC.
This method may only be called if the cardinality of this
- RpcMethod is Cardinality.UNARY_UNARY.
+ RpcMethodServiceDescription is Cardinality.UNARY_UNARY.
Args:
- request: A request value appropriate for this RpcMethod.
+ request: A request value appropriate for the RPC method described by this
+ RpcMethodServiceDescription.
+ context: An RpcContext object for the RPC.
Returns:
- A response value appropriate for this RpcMethod.
+ A response value appropriate for the RPC method described by this
+ RpcMethodServiceDescription.
"""
raise NotImplementedError()
@abc.abstractmethod
- def service_unary_stream(self, request):
+ def service_unary_stream(self, request, context):
"""Carries out this RPC.
This method may only be called if the cardinality of this
- RpcMethod is Cardinality.UNARY_STREAM.
+ RpcMethodServiceDescription is Cardinality.UNARY_STREAM.
Args:
- request: A request value appropriate for this RpcMethod.
+ request: A request value appropriate for the RPC method described by this
+ RpcMethodServiceDescription.
+ context: An RpcContext object for the RPC.
Yields:
- Zero or more response values appropriate for this
- RpcMethod.
+ Zero or more response values appropriate for the RPC method described by
+ this RpcMethodServiceDescription.
"""
raise NotImplementedError()
@abc.abstractmethod
- def service_stream_unary(self, request_iterator):
+ def service_stream_unary(self, request_iterator, context):
"""Carries out this RPC.
This method may only be called if the cardinality of this
- RpcMethod is Cardinality.STREAM_UNARY.
+ RpcMethodServiceDescription is Cardinality.STREAM_UNARY.
Args:
- request_iterator: An iterator of request values
- appropriate for this RpcMethod.
+ request_iterator: An iterator of request values appropriate for the RPC
+ method described by this RpcMethodServiceDescription.
+ context: An RpcContext object for the RPC.
Returns:
- A response value appropriate for this RpcMethod.
+ A response value appropriate for the RPC method described by this
+ RpcMethodServiceDescription.
"""
raise NotImplementedError()
@abc.abstractmethod
- def service_stream_stream(self, request_iterator):
+ def service_stream_stream(self, request_iterator, context):
"""Carries out this RPC.
This method may only be called if the cardinality of this
- RpcMethod is Cardinality.STREAM_STREAM.
+ RpcMethodServiceDescription is Cardinality.STREAM_STREAM.
Args:
- request_iterator: An iterator of request values
- appropriate for this RpcMethod.
+ request_iterator: An iterator of request values appropriate for the RPC
+ method described by this RpcMethodServiceDescription.
+ context: An RpcContext object for the RPC.
Yields:
- Zero or more response values appropraite for this
- RpcMethod.
+ Zero or more response values appropriate for the RPC method described by
+ this RpcMethodServiceDescription.
"""
raise NotImplementedError()
-class Server(object):
+class Stub(object):
+ """A stub with callable RPC method names for attributes.
+
+ Instances of this type are context managers and only afford RPC invocation
+ when used in context.
+
+ Instances of this type, when used in context, respond to attribute access
+ as follows: if the requested attribute is the name of a unary-unary RPC
+ method, the value of the attribute will be a UnaryUnarySyncAsync with which
+ to invoke the RPC method. If the requested attribute is the name of a
+ unary-stream RPC method, the value of the attribute will be a callable taking
+ a request object and a timeout parameter and returning a CancellableIterator
+ that yields the response values of the RPC. If the requested attribute is the
+ name of a stream-unary RPC method, the value of the attribute will be a
+ StreamUnarySyncAsync with which to invoke the RPC method. If the requested
+ attribute is the name of a stream-stream RPC method, the value of the
+ attribute will be a callable taking an iterator of request objects and a
+ timeout and returning a CancellableIterator that yields the response values
+ of the RPC.
+
+ In all cases indication of abortion is indicated by raising of
+ exceptions.RpcError, exceptions.CancellationError,
+ and exceptions.ExpirationError.
+ """
+ __metaclass__ = abc.ABCMeta
+
+
+class Server(activated.Activated):
"""A GRPC Server."""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
- def start(self):
- """Instructs this server to commence service of RPCs."""
- raise NotImplementedError()
+ def port(self):
+ """Reports the port on which the server is serving.
- @abc.abstractmethod
- def stop(self):
- """Instructs this server to halt service of RPCs."""
+ This method may only be called while the server is activated.
+
+ Returns:
+ The port on which the server is serving.
+ """
raise NotImplementedError()
diff --git a/src/python/src/grpc/early_adopter/utilities.py b/src/python/src/grpc/early_adopter/utilities.py
index 9277d3f..da8ef82 100644
--- a/src/python/src/grpc/early_adopter/utilities.py
+++ b/src/python/src/grpc/early_adopter/utilities.py
@@ -32,7 +32,9 @@
from grpc.early_adopter import interfaces
-class _RpcMethod(interfaces.ClientRpcMethod, interfaces.ServerRpcMethod):
+class _RpcMethodDescription(
+ interfaces.RpcMethodInvocationDescription,
+ interfaces.RpcMethodServiceDescription):
def __init__(
self, cardinality, unary_unary, unary_stream, stream_unary,
@@ -49,44 +51,45 @@
self._response_deserializer = response_deserializer
def cardinality(self):
- """See interfaces.RpcMethod.cardinality for specification."""
+ """See interfaces.RpcMethodDescription.cardinality for specification."""
return self._cardinality
def serialize_request(self, request):
- """See interfaces.RpcMethod.serialize_request for specification."""
+ """See interfaces.RpcMethodInvocationDescription.serialize_request."""
return self._request_serializer(request)
def deserialize_request(self, serialized_request):
- """See interfaces.RpcMethod.deserialize_request for specification."""
+ """See interfaces.RpcMethodServiceDescription.deserialize_request."""
return self._request_deserializer(serialized_request)
def serialize_response(self, response):
- """See interfaces.RpcMethod.serialize_response for specification."""
+ """See interfaces.RpcMethodServiceDescription.serialize_response."""
return self._response_serializer(response)
def deserialize_response(self, serialized_response):
- """See interfaces.RpcMethod.deserialize_response for specification."""
+ """See interfaces.RpcMethodInvocationDescription.deserialize_response."""
return self._response_deserializer(serialized_response)
- def service_unary_unary(self, request):
- """See interfaces.RpcMethod.service_unary_unary for specification."""
- return self._unary_unary(request)
+ def service_unary_unary(self, request, context):
+ """See interfaces.RpcMethodServiceDescription.service_unary_unary."""
+ return self._unary_unary(request, context)
- def service_unary_stream(self, request):
- """See interfaces.RpcMethod.service_unary_stream for specification."""
- return self._unary_stream(request)
+ def service_unary_stream(self, request, context):
+ """See interfaces.RpcMethodServiceDescription.service_unary_stream."""
+ return self._unary_stream(request, context)
- def service_stream_unary(self, request_iterator):
- """See interfaces.RpcMethod.service_stream_unary for specification."""
- return self._stream_unary(request_iterator)
+ def service_stream_unary(self, request_iterator, context):
+ """See interfaces.RpcMethodServiceDescription.service_stream_unary."""
+ return self._stream_unary(request_iterator, context)
- def service_stream_stream(self, request_iterator):
- """See interfaces.RpcMethod.service_stream_stream for specification."""
- return self._stream_stream(request_iterator)
+ def service_stream_stream(self, request_iterator, context):
+ """See interfaces.RpcMethodServiceDescription.service_stream_stream."""
+ return self._stream_stream(request_iterator, context)
-def unary_unary_client_rpc_method(request_serializer, response_deserializer):
- """Constructs an interfaces.ClientRpcMethod for a unary-unary RPC method.
+def unary_unary_invocation_description(
+ request_serializer, response_deserializer):
+ """Creates an interfaces.RpcMethodInvocationDescription for an RPC method.
Args:
request_serializer: A callable that when called on a request
@@ -96,17 +99,17 @@
that bytestring.
Returns:
- An interfaces.ClientRpcMethod constructed from the given
- arguments representing a unary-request/unary-response RPC
- method.
+ An interfaces.RpcMethodInvocationDescription constructed from the given
+ arguments representing a unary-request/unary-response RPC method.
"""
- return _RpcMethod(
+ return _RpcMethodDescription(
interfaces.Cardinality.UNARY_UNARY, None, None, None, None,
request_serializer, None, None, response_deserializer)
-def unary_stream_client_rpc_method(request_serializer, response_deserializer):
- """Constructs an interfaces.ClientRpcMethod for a unary-stream RPC method.
+def unary_stream_invocation_description(
+ request_serializer, response_deserializer):
+ """Creates an interfaces.RpcMethodInvocationDescription for an RPC method.
Args:
request_serializer: A callable that when called on a request
@@ -116,17 +119,17 @@
that bytestring.
Returns:
- An interfaces.ClientRpcMethod constructed from the given
- arguments representing a unary-request/streaming-response
- RPC method.
+ An interfaces.RpcMethodInvocationDescription constructed from the given
+ arguments representing a unary-request/streaming-response RPC method.
"""
- return _RpcMethod(
+ return _RpcMethodDescription(
interfaces.Cardinality.UNARY_STREAM, None, None, None, None,
request_serializer, None, None, response_deserializer)
-def stream_unary_client_rpc_method(request_serializer, response_deserializer):
- """Constructs an interfaces.ClientRpcMethod for a stream-unary RPC method.
+def stream_unary_invocation_description(
+ request_serializer, response_deserializer):
+ """Creates an interfaces.RpcMethodInvocationDescription for an RPC method.
Args:
request_serializer: A callable that when called on a request
@@ -136,17 +139,17 @@
that bytestring.
Returns:
- An interfaces.ClientRpcMethod constructed from the given
- arguments representing a streaming-request/unary-response
- RPC method.
+ An interfaces.RpcMethodInvocationDescription constructed from the given
+ arguments representing a streaming-request/unary-response RPC method.
"""
- return _RpcMethod(
+ return _RpcMethodDescription(
interfaces.Cardinality.STREAM_UNARY, None, None, None, None,
request_serializer, None, None, response_deserializer)
-def stream_stream_client_rpc_method(request_serializer, response_deserializer):
- """Constructs an interfaces.ClientRpcMethod for a stream-stream RPC method.
+def stream_stream_invocation_description(
+ request_serializer, response_deserializer):
+ """Creates an interfaces.RpcMethodInvocationDescription for an RPC method.
Args:
request_serializer: A callable that when called on a request
@@ -156,23 +159,23 @@
that bytestring.
Returns:
- An interfaces.ClientRpcMethod constructed from the given
- arguments representing a
- streaming-request/streaming-response RPC method.
+ An interfaces.RpcMethodInvocationDescription constructed from the given
+ arguments representing a streaming-request/streaming-response RPC
+ method.
"""
- return _RpcMethod(
+ return _RpcMethodDescription(
interfaces.Cardinality.STREAM_STREAM, None, None, None, None,
request_serializer, None, None, response_deserializer)
-def unary_unary_server_rpc_method(
+def unary_unary_service_description(
behavior, request_deserializer, response_serializer):
- """Constructs an interfaces.ServerRpcMethod for the given behavior.
+ """Creates an interfaces.RpcMethodServiceDescription for the given behavior.
Args:
behavior: A callable that implements a unary-unary RPC
- method that accepts a single request and returns a single
- response.
+ method that accepts a single request and an interfaces.RpcContext and
+ returns a single response.
request_deserializer: A callable that when called on a
bytestring returns the request value corresponding to that
bytestring.
@@ -181,72 +184,22 @@
that value.
Returns:
- An interfaces.ServerRpcMethod constructed from the given
+ An interfaces.RpcMethodServiceDescription constructed from the given
arguments representing a unary-request/unary-response RPC
method.
"""
- return _RpcMethod(
+ return _RpcMethodDescription(
interfaces.Cardinality.UNARY_UNARY, behavior, None, None, None,
None, request_deserializer, response_serializer, None)
-def unary_stream_server_rpc_method(
+def unary_stream_service_description(
behavior, request_deserializer, response_serializer):
- """Constructs an interfaces.ServerRpcMethod for the given behavior.
+ """Creates an interfaces.RpcMethodServiceDescription for the given behavior.
Args:
behavior: A callable that implements a unary-stream RPC
- method that accepts a single request and returns an
- iterator of zero or more responses.
- request_deserializer: A callable that when called on a
- bytestring returns the request value corresponding to that
- bytestring.
- response_serializer: A callable that when called on a
- response value returns the bytestring corresponding to
- that value.
-
- Returns:
- An interfaces.ServerRpcMethod constructed from the given
- arguments representing a unary-request/streaming-response
- RPC method.
- """
- return _RpcMethod(
- interfaces.Cardinality.UNARY_STREAM, None, behavior, None, None,
- None, request_deserializer, response_serializer, None)
-
-
-def stream_unary_server_rpc_method(
- behavior, request_deserializer, response_serializer):
- """Constructs an interfaces.ServerRpcMethod for the given behavior.
-
- Args:
- behavior: A callable that implements a stream-unary RPC
- method that accepts an iterator of zero or more requests
- and returns a single response.
- request_deserializer: A callable that when called on a
- bytestring returns the request value corresponding to that
- bytestring.
- response_serializer: A callable that when called on a
- response value returns the bytestring corresponding to
- that value.
-
- Returns:
- An interfaces.ServerRpcMethod constructed from the given
- arguments representing a streaming-request/unary-response
- RPC method.
- """
- return _RpcMethod(
- interfaces.Cardinality.STREAM_UNARY, None, None, behavior, None,
- None, request_deserializer, response_serializer, None)
-
-
-def stream_stream_server_rpc_method(
- behavior, request_deserializer, response_serializer):
- """Constructs an interfaces.ServerRpcMethod for the given behavior.
-
- Args:
- behavior: A callable that implements a stream-stream RPC
- method that accepts an iterator of zero or more requests
+ method that accepts a single request and an interfaces.RpcContext
and returns an iterator of zero or more responses.
request_deserializer: A callable that when called on a
bytestring returns the request value corresponding to that
@@ -256,10 +209,61 @@
that value.
Returns:
- An interfaces.ServerRpcMethod constructed from the given
+ An interfaces.RpcMethodServiceDescription constructed from the given
+ arguments representing a unary-request/streaming-response
+ RPC method.
+ """
+ return _RpcMethodDescription(
+ interfaces.Cardinality.UNARY_STREAM, None, behavior, None, None,
+ None, request_deserializer, response_serializer, None)
+
+
+def stream_unary_service_description(
+ behavior, request_deserializer, response_serializer):
+ """Creates an interfaces.RpcMethodServiceDescription for the given behavior.
+
+ Args:
+ behavior: A callable that implements a stream-unary RPC
+ method that accepts an iterator of zero or more requests
+ and an interfaces.RpcContext and returns a single response.
+ request_deserializer: A callable that when called on a
+ bytestring returns the request value corresponding to that
+ bytestring.
+ response_serializer: A callable that when called on a
+ response value returns the bytestring corresponding to
+ that value.
+
+ Returns:
+ An interfaces.RpcMethodServiceDescription constructed from the given
+ arguments representing a streaming-request/unary-response
+ RPC method.
+ """
+ return _RpcMethodDescription(
+ interfaces.Cardinality.STREAM_UNARY, None, None, behavior, None,
+ None, request_deserializer, response_serializer, None)
+
+
+def stream_stream_service_description(
+ behavior, request_deserializer, response_serializer):
+ """Creates an interfaces.RpcMethodServiceDescription for the given behavior.
+
+ Args:
+ behavior: A callable that implements a stream-stream RPC
+ method that accepts an iterator of zero or more requests
+ and an interfaces.RpcContext and returns an iterator of
+ zero or more responses.
+ request_deserializer: A callable that when called on a
+ bytestring returns the request value corresponding to that
+ bytestring.
+ response_serializer: A callable that when called on a
+ response value returns the bytestring corresponding to
+ that value.
+
+ Returns:
+ An interfaces.RpcMethodServiceDescription constructed from the given
arguments representing a
streaming-request/streaming-response RPC method.
"""
- return _RpcMethod(
+ return _RpcMethodDescription(
interfaces.Cardinality.STREAM_STREAM, None, None, None, behavior,
None, request_deserializer, response_serializer, None)
diff --git a/src/python/src/setup.py b/src/python/src/setup.py
index e3f13fa..26121dc 100644
--- a/src/python/src/setup.py
+++ b/src/python/src/setup.py
@@ -38,6 +38,7 @@
'grpc/_adapter/_completion_queue.c',
'grpc/_adapter/_error.c',
'grpc/_adapter/_server.c',
+ 'grpc/_adapter/_client_credentials.c',
'grpc/_adapter/_server_credentials.c',
)
@@ -80,6 +81,6 @@
}
_core.setup(
- name='grpc-2015', version='0.0.1',
+ name='grpc-2015', version='0.4.0',
ext_modules=[_EXTENSION_MODULE], packages=_PACKAGES,
package_dir=_PACKAGE_DIRECTORIES)
diff --git a/src/ruby/bin/apis/pubsub_demo.rb b/src/ruby/bin/apis/pubsub_demo.rb
index 6656a56..9bb324f 100755
--- a/src/ruby/bin/apis/pubsub_demo.rb
+++ b/src/ruby/bin/apis/pubsub_demo.rb
@@ -31,10 +31,9 @@
# pubsub_demo demos accesses the Google PubSub API via its gRPC interface
#
-# TODO: update the Usage once the usable auth gem is available
-# $ SSL_CERT_FILE=<path/to/ssl/certs> \
+# $ GOOGLE_APPLICATION_CREDENTIALS=<path_to_service_account_key_file> \
+# SSL_CERT_FILE=<path/to/ssl/certs> \
# path/to/pubsub_demo.rb \
-# --service_account_key_file=<path_to_service_account> \
# [--action=<chosen_demo_action> ]
#
# There are options related to the chosen action, see #parse_args below.
@@ -49,6 +48,7 @@
require 'optparse'
require 'grpc'
+require 'googleauth'
require 'google/protobuf'
require 'google/protobuf/empty'
@@ -59,7 +59,9 @@
def load_prod_cert
fail 'could not find a production cert' if ENV['SSL_CERT_FILE'].nil?
p "loading prod certs from #{ENV['SSL_CERT_FILE']}"
- File.open(ENV['SSL_CERT_FILE']).read
+ File.open(ENV['SSL_CERT_FILE']) do |f|
+ return f.read
+ end
end
# creates a SSL Credentials from the production certificates.
@@ -68,14 +70,9 @@
end
# Builds the metadata authentication update proc.
-#
-# TODO: replace this once the ruby usable auth repo is available.
def auth_proc(opts)
- if GRPC::Auth::GCECredentials.on_gce?
- return GRPC::Auth::GCECredentials.new.updater_proc
- end
- fd = StringIO.new(File.read(opts.oauth_key_file))
- GRPC::Auth::ServiceAccountCredentials.new(opts.oauth_scope, fd).updater_proc
+ auth_creds = Google::Auth.get_application_default(opts.oauth_scope)
+ return auth_creds.updater_proc
end
# Creates a stub for accessing the publisher service.
@@ -216,14 +213,14 @@
end
# Args is used to hold the command line info.
-Args = Struct.new(:host, :oauth_scope, :oauth_key_file, :port, :action,
- :project_id, :topic_name, :sub_name)
+Args = Struct.new(:host, :oauth_scope, :port, :action, :project_id, :topic_name,
+ :sub_name)
# validates the the command line options, returning them as an Arg.
def parse_args
args = Args.new('pubsub-staging.googleapis.com',
'https://www.googleapis.com/auth/pubsub',
- nil, 443, 'list_some_topics', 'stoked-keyword-656')
+ 443, 'list_some_topics', 'stoked-keyword-656')
OptionParser.new do |opts|
opts.on('--oauth_scope scope',
'Scope for OAuth tokens') { |v| args['oauth_scope'] = v }
@@ -233,10 +230,6 @@
opts.on('--server_port SERVER_PORT', 'server port') do |v|
args.port = v
end
- opts.on('--service_account_key_file PATH',
- 'Path to the service account json key file') do |v|
- args.oauth_key_file = v
- end
# instance_methods(false) gives only the methods defined in that class.
scenes = NamedActions.instance_methods(false).map { |t| t.to_s }
@@ -257,15 +250,11 @@
end
def _check_args(args)
- %w(host port action).each do |a|
+ %w(host port action oauth_scope).each do |a|
if args[a].nil?
raise OptionParser::MissingArgument.new("please specify --#{a}")
end
end
- if args['oauth_key_file'].nil? || args['oauth_scope'].nil?
- fail(OptionParser::MissingArgument,
- 'please specify both of --service_account_key_file and --oauth_scope')
- end
args
end
diff --git a/src/ruby/bin/interop/interop_client.rb b/src/ruby/bin/interop/interop_client.rb
index 380ceb1..b0b24b9 100755
--- a/src/ruby/bin/interop/interop_client.rb
+++ b/src/ruby/bin/interop/interop_client.rb
@@ -48,6 +48,7 @@
require 'minitest/assertions'
require 'grpc'
+require 'googleauth'
require 'google/protobuf'
require 'test/cpp/interop/test_services'
@@ -56,7 +57,7 @@
require 'signet/ssl_config'
-include GRPC::Auth
+AUTH_ENV = Google::Auth::ServiceAccountCredentials::ENV_VAR
# loads the certificates used to access the test server securely.
def load_test_certs
@@ -101,22 +102,14 @@
}
# Add service account creds if specified
- if %w(all service_account_creds).include?(opts.test_case)
+ wants_creds = %w(all compute_engine_creds service_account_creds)
+ if wants_creds.include?(opts.test_case)
unless opts.oauth_scope.nil?
- fd = StringIO.new(File.read(opts.oauth_key_file))
- logger.info("loading oauth certs from #{opts.oauth_key_file}")
- auth_creds = ServiceAccountCredentials.new(opts.oauth_scope, fd)
+ auth_creds = Google::Auth.get_application_default(opts.oauth_scope)
stub_opts[:update_metadata] = auth_creds.updater_proc
end
end
- # Add compute engine creds if specified
- if %w(all compute_engine_creds).include?(opts.test_case)
- unless opts.oauth_scope.nil?
- stub_opts[:update_metadata] = GCECredentials.new.update_proc
- end
- end
-
logger.info("... connecting securely to #{address}")
Grpc::Testing::TestService::Stub.new(address, **stub_opts)
else
@@ -193,11 +186,11 @@
def service_account_creds
# ignore this test if the oauth options are not set
- if @args.oauth_scope.nil? || @args.oauth_key_file.nil?
+ if @args.oauth_scope.nil?
p 'NOT RUN: service_account_creds; no service_account settings'
return
end
- json_key = File.read(@args.oauth_key_file)
+ json_key = File.read(ENV[AUTH_ENV])
wanted_email = MultiJson.load(json_key)['client_email']
resp = perform_large_unary(fill_username: true,
fill_oauth_scope: true)
@@ -285,7 +278,7 @@
# Args is used to hold the command line info.
Args = Struct.new(:default_service_account, :host, :host_override,
- :oauth_scope, :oauth_key_file, :port, :secure, :test_case,
+ :oauth_scope, :port, :secure, :test_case,
:use_test_ca)
# validates the the command line options, returning them as a Hash.
@@ -302,10 +295,6 @@
'email address of the default service account') do |v|
args['default_service_account'] = v
end
- opts.on('--service_account_key_file PATH',
- 'Path to the service account json key file') do |v|
- args['oauth_key_file'] = v
- end
opts.on('--server_host_override HOST_OVERRIDE',
'override host via a HTTP header') do |v|
args['host_override'] = v
@@ -333,10 +322,6 @@
fail(OptionParser::MissingArgument, "please specify --#{arg}")
end
end
- if args['oauth_key_file'].nil? ^ args['oauth_scope'].nil?
- fail(OptionParser::MissingArgument,
- 'please specify both of --service_account_key_file and --oauth_scope')
- end
args
end
diff --git a/src/ruby/grpc.gemspec b/src/ruby/grpc.gemspec
index bc59c23..25a3ff5 100755
--- a/src/ruby/grpc.gemspec
+++ b/src/ruby/grpc.gemspec
@@ -22,6 +22,7 @@
s.add_dependency 'faraday', '~> 0.9'
s.add_dependency 'google-protobuf', '~> 3.0.0alpha.1.1'
+ s.add_dependency 'googleauth', '~> 0.1'
s.add_dependency 'logging', '~> 1.8'
s.add_dependency 'jwt', '~> 1.2.1'
s.add_dependency 'minitest', '~> 5.4' # reqd for interop tests
diff --git a/src/ruby/lib/grpc.rb b/src/ruby/lib/grpc.rb
index a2a609f..dd02ef7 100644
--- a/src/ruby/lib/grpc.rb
+++ b/src/ruby/lib/grpc.rb
@@ -27,8 +27,6 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-require 'grpc/auth/compute_engine.rb'
-require 'grpc/auth/service_account.rb'
require 'grpc/errors'
require 'grpc/grpc'
require 'grpc/logconfig'
diff --git a/src/ruby/lib/grpc/auth/compute_engine.rb b/src/ruby/lib/grpc/auth/compute_engine.rb
deleted file mode 100644
index 5cb1e1a..0000000
--- a/src/ruby/lib/grpc/auth/compute_engine.rb
+++ /dev/null
@@ -1,67 +0,0 @@
-# Copyright 2015, Google Inc.
-# All rights reserved.
-#
-# 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.
-
-require 'faraday'
-require 'grpc/auth/signet'
-
-module GRPC
- # Module Auth provides classes that provide Google-specific authentication
- # used to access Google gRPC services.
- module Auth
- # Extends Signet::OAuth2::Client so that the auth token is obtained from
- # the GCE metadata server.
- class GCECredentials < Signet::OAuth2::Client
- COMPUTE_AUTH_TOKEN_URI = 'http://metadata/computeMetadata/v1/'\
- 'instance/service-accounts/default/token'
- COMPUTE_CHECK_URI = 'http://metadata.google.internal'
-
- # Detect if this appear to be a GCE instance, by checking if metadata
- # is available
- def self.on_gce?(options = {})
- c = options[:connection] || Faraday.default_connection
- resp = c.get(COMPUTE_CHECK_URI)
- return false unless resp.status == 200
- return false unless resp.headers.key?('Metadata-Flavor')
- return resp.headers['Metadata-Flavor'] == 'Google'
- rescue Faraday::ConnectionFailed
- return false
- end
-
- # Overrides the super class method to change how access tokens are
- # fetched.
- def fetch_access_token(options = {})
- c = options[:connection] || Faraday.default_connection
- c.headers = { 'Metadata-Flavor' => 'Google' }
- resp = c.get(COMPUTE_AUTH_TOKEN_URI)
- Signet::OAuth2.parse_credentials(resp.body,
- resp.headers['content-type'])
- end
- end
- end
-end
diff --git a/src/ruby/lib/grpc/auth/service_account.rb b/src/ruby/lib/grpc/auth/service_account.rb
deleted file mode 100644
index 14b81a9..0000000
--- a/src/ruby/lib/grpc/auth/service_account.rb
+++ /dev/null
@@ -1,66 +0,0 @@
-# Copyright 2015, Google Inc.
-# All rights reserved.
-#
-# 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.
-
-require 'grpc/auth/signet'
-require 'multi_json'
-require 'openssl'
-
-# Reads the private key and client email fields from service account JSON key.
-def read_json_key(json_key_io)
- json_key = MultiJson.load(json_key_io.read)
- fail 'missing client_email' unless json_key.key?('client_email')
- fail 'missing private_key' unless json_key.key?('private_key')
- [json_key['private_key'], json_key['client_email']]
-end
-
-module GRPC
- # Module Auth provides classes that provide Google-specific authentication
- # used to access Google gRPC services.
- module Auth
- # Authenticates requests using Google's Service Account credentials.
- # (cf https://developers.google.com/accounts/docs/OAuth2ServiceAccount)
- class ServiceAccountCredentials < Signet::OAuth2::Client
- TOKEN_CRED_URI = 'https://www.googleapis.com/oauth2/v3/token'
- AUDIENCE = TOKEN_CRED_URI
-
- # Initializes a ServiceAccountCredentials.
- #
- # @param scope [string|array] the scope(s) to access
- # @param json_key_io [IO] an IO from which the JSON key can be read
- def initialize(scope, json_key_io)
- private_key, client_email = read_json_key(json_key_io)
- super(token_credential_uri: TOKEN_CRED_URI,
- audience: AUDIENCE,
- scope: scope,
- issuer: client_email,
- signing_key: OpenSSL::PKey::RSA.new(private_key))
- end
- end
- end
-end
diff --git a/src/ruby/lib/grpc/auth/signet.rb b/src/ruby/lib/grpc/auth/signet.rb
deleted file mode 100644
index a8bce12..0000000
--- a/src/ruby/lib/grpc/auth/signet.rb
+++ /dev/null
@@ -1,67 +0,0 @@
-# Copyright 2015, Google Inc.
-# All rights reserved.
-#
-# 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.
-
-require 'signet/oauth_2/client'
-
-module Signet
- # Signet::OAuth2 supports OAuth2 authentication.
- module OAuth2
- AUTH_METADATA_KEY = :Authorization
- # Signet::OAuth2::Client creates an OAuth2 client
- #
- # Here client is re-opened to add the #apply and #apply! methods which
- # update a hash map with the fetched authentication token
- #
- # Eventually, this change may be merged into signet itself, or some other
- # package that provides Google-specific auth via signet, and this extension
- # will be unnecessary.
- class Client
- # Updates a_hash updated with the authentication token
- def apply!(a_hash, opts = {})
- # fetch the access token there is currently not one, or if the client
- # has expired
- fetch_access_token!(opts) if access_token.nil? || expired?
- a_hash[AUTH_METADATA_KEY] = "Bearer #{access_token}"
- end
-
- # Returns a clone of a_hash updated with the authentication token
- def apply(a_hash, opts = {})
- a_copy = a_hash.clone
- apply!(a_copy, opts)
- a_copy
- end
-
- # Returns a reference to the #apply method, suitable for passing as
- # a closure
- def updater_proc
- lambda(&method(:apply))
- end
- end
- end
-end
diff --git a/src/ruby/lib/grpc/generic/client_stub.rb b/src/ruby/lib/grpc/generic/client_stub.rb
index 7fc0d83..f234984 100644
--- a/src/ruby/lib/grpc/generic/client_stub.rb
+++ b/src/ruby/lib/grpc/generic/client_stub.rb
@@ -400,7 +400,12 @@
# @param deadline [TimeConst]
def new_active_call(ch, marshal, unmarshal, deadline = nil)
absolute_deadline = Core::TimeConsts.from_relative_time(deadline)
- call = @ch.create_call(ch, @host, absolute_deadline)
+ # It should be OK to to pass the hostname:port to create_call, but at
+ # the moment this fails a security check. This will be corrected.
+ #
+ # TODO: # remove this after create_call is updated
+ host = @host.split(':')[0]
+ call = @ch.create_call(ch, host, absolute_deadline)
ActiveCall.new(call, @queue, marshal, unmarshal, absolute_deadline,
started: false)
end
diff --git a/src/ruby/lib/grpc/version.rb b/src/ruby/lib/grpc/version.rb
index d4eb0ed..513a537 100644
--- a/src/ruby/lib/grpc/version.rb
+++ b/src/ruby/lib/grpc/version.rb
@@ -29,5 +29,5 @@
# GRPC contains the General RPC module.
module GRPC
- VERSION = '0.0.1'
+ VERSION = '0.5.0'
end
diff --git a/src/ruby/spec/auth/apply_auth_examples.rb b/src/ruby/spec/auth/apply_auth_examples.rb
deleted file mode 100644
index 09b3930..0000000
--- a/src/ruby/spec/auth/apply_auth_examples.rb
+++ /dev/null
@@ -1,163 +0,0 @@
-# Copyright 2015, Google Inc.
-# All rights reserved.
-#
-# 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.
-
-spec_dir = File.expand_path(File.join(File.dirname(__FILE__)))
-$LOAD_PATH.unshift(spec_dir)
-$LOAD_PATH.uniq!
-
-require 'faraday'
-require 'spec_helper'
-
-def build_json_response(payload)
- [200,
- { 'Content-Type' => 'application/json; charset=utf-8' },
- MultiJson.dump(payload)]
-end
-
-WANTED_AUTH_KEY = :Authorization
-
-shared_examples 'apply/apply! are OK' do
- # tests that use these examples need to define
- #
- # @client which should be an auth client
- #
- # @make_auth_stubs, which should stub out the expected http behaviour of the
- # auth client
- describe '#fetch_access_token' do
- it 'should set access_token to the fetched value' do
- token = '1/abcdef1234567890'
- stubs = make_auth_stubs with_access_token: token
- c = Faraday.new do |b|
- b.adapter(:test, stubs)
- end
-
- @client.fetch_access_token!(connection: c)
- expect(@client.access_token).to eq(token)
- stubs.verify_stubbed_calls
- end
- end
-
- describe '#apply!' do
- it 'should update the target hash with fetched access token' do
- token = '1/abcdef1234567890'
- stubs = make_auth_stubs with_access_token: token
- c = Faraday.new do |b|
- b.adapter(:test, stubs)
- end
-
- md = { foo: 'bar' }
- @client.apply!(md, connection: c)
- want = { :foo => 'bar', WANTED_AUTH_KEY => "Bearer #{token}" }
- expect(md).to eq(want)
- stubs.verify_stubbed_calls
- end
- end
-
- describe 'updater_proc' do
- it 'should provide a proc that updates a hash with the access token' do
- token = '1/abcdef1234567890'
- stubs = make_auth_stubs with_access_token: token
- c = Faraday.new do |b|
- b.adapter(:test, stubs)
- end
-
- md = { foo: 'bar' }
- the_proc = @client.updater_proc
- got = the_proc.call(md, connection: c)
- want = { :foo => 'bar', WANTED_AUTH_KEY => "Bearer #{token}" }
- expect(got).to eq(want)
- stubs.verify_stubbed_calls
- end
- end
-
- describe '#apply' do
- it 'should not update the original hash with the access token' do
- token = '1/abcdef1234567890'
- stubs = make_auth_stubs with_access_token: token
- c = Faraday.new do |b|
- b.adapter(:test, stubs)
- end
-
- md = { foo: 'bar' }
- @client.apply(md, connection: c)
- want = { foo: 'bar' }
- expect(md).to eq(want)
- stubs.verify_stubbed_calls
- end
-
- it 'should add the token to the returned hash' do
- token = '1/abcdef1234567890'
- stubs = make_auth_stubs with_access_token: token
- c = Faraday.new do |b|
- b.adapter(:test, stubs)
- end
-
- md = { foo: 'bar' }
- got = @client.apply(md, connection: c)
- want = { :foo => 'bar', WANTED_AUTH_KEY => "Bearer #{token}" }
- expect(got).to eq(want)
- stubs.verify_stubbed_calls
- end
-
- it 'should not fetch a new token if the current is not expired' do
- token = '1/abcdef1234567890'
- stubs = make_auth_stubs with_access_token: token
- c = Faraday.new do |b|
- b.adapter(:test, stubs)
- end
-
- n = 5 # arbitrary
- n.times do |_t|
- md = { foo: 'bar' }
- got = @client.apply(md, connection: c)
- want = { :foo => 'bar', WANTED_AUTH_KEY => "Bearer #{token}" }
- expect(got).to eq(want)
- end
- stubs.verify_stubbed_calls
- end
-
- it 'should fetch a new token if the current one is expired' do
- token_1 = '1/abcdef1234567890'
- token_2 = '2/abcdef1234567890'
-
- [token_1, token_2].each do |t|
- stubs = make_auth_stubs with_access_token: t
- c = Faraday.new do |b|
- b.adapter(:test, stubs)
- end
- md = { foo: 'bar' }
- got = @client.apply(md, connection: c)
- want = { :foo => 'bar', WANTED_AUTH_KEY => "Bearer #{t}" }
- expect(got).to eq(want)
- stubs.verify_stubbed_calls
- @client.expires_at -= 3601 # default is to expire in 1hr
- end
- end
- end
-end
diff --git a/src/ruby/spec/auth/compute_engine_spec.rb b/src/ruby/spec/auth/compute_engine_spec.rb
deleted file mode 100644
index c43214d..0000000
--- a/src/ruby/spec/auth/compute_engine_spec.rb
+++ /dev/null
@@ -1,108 +0,0 @@
-# Copyright 2015, Google Inc.
-# All rights reserved.
-#
-# 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.
-
-spec_dir = File.expand_path(File.join(File.dirname(__FILE__)))
-$LOAD_PATH.unshift(spec_dir)
-$LOAD_PATH.uniq!
-
-require 'apply_auth_examples'
-require 'faraday'
-require 'grpc/auth/compute_engine'
-require 'spec_helper'
-
-describe GRPC::Auth::GCECredentials do
- MD_URI = '/computeMetadata/v1/instance/service-accounts/default/token'
- GCECredentials = GRPC::Auth::GCECredentials
-
- before(:example) do
- @client = GCECredentials.new
- end
-
- def make_auth_stubs(with_access_token: '')
- Faraday::Adapter::Test::Stubs.new do |stub|
- stub.get(MD_URI) do |env|
- headers = env[:request_headers]
- expect(headers['Metadata-Flavor']).to eq('Google')
- build_json_response(
- 'access_token' => with_access_token,
- 'token_type' => 'Bearer',
- 'expires_in' => 3600)
- end
- end
- end
-
- it_behaves_like 'apply/apply! are OK'
-
- describe '#on_gce?' do
- it 'should be true when Metadata-Flavor is Google' do
- stubs = Faraday::Adapter::Test::Stubs.new do |stub|
- stub.get('/') do |_env|
- [200,
- { 'Metadata-Flavor' => 'Google' },
- '']
- end
- end
- c = Faraday.new do |b|
- b.adapter(:test, stubs)
- end
- expect(GCECredentials.on_gce?(connection: c)).to eq(true)
- stubs.verify_stubbed_calls
- end
-
- it 'should be false when Metadata-Flavor is not Google' do
- stubs = Faraday::Adapter::Test::Stubs.new do |stub|
- stub.get('/') do |_env|
- [200,
- { 'Metadata-Flavor' => 'NotGoogle' },
- '']
- end
- end
- c = Faraday.new do |b|
- b.adapter(:test, stubs)
- end
- expect(GCECredentials.on_gce?(connection: c)).to eq(false)
- stubs.verify_stubbed_calls
- end
-
- it 'should be false if the response is not 200' do
- stubs = Faraday::Adapter::Test::Stubs.new do |stub|
- stub.get('/') do |_env|
- [404,
- { 'Metadata-Flavor' => 'Google' },
- '']
- end
- end
- c = Faraday.new do |b|
- b.adapter(:test, stubs)
- end
- expect(GCECredentials.on_gce?(connection: c)).to eq(false)
- stubs.verify_stubbed_calls
- end
- end
-end
diff --git a/src/ruby/spec/auth/service_account_spec.rb b/src/ruby/spec/auth/service_account_spec.rb
deleted file mode 100644
index 2f14a1a..0000000
--- a/src/ruby/spec/auth/service_account_spec.rb
+++ /dev/null
@@ -1,75 +0,0 @@
-# Copyright 2015, Google Inc.
-# All rights reserved.
-#
-# 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.
-
-spec_dir = File.expand_path(File.join(File.dirname(__FILE__)))
-$LOAD_PATH.unshift(spec_dir)
-$LOAD_PATH.uniq!
-
-require 'apply_auth_examples'
-require 'grpc/auth/service_account'
-require 'jwt'
-require 'multi_json'
-require 'openssl'
-require 'spec_helper'
-
-describe GRPC::Auth::ServiceAccountCredentials do
- before(:example) do
- @key = OpenSSL::PKey::RSA.new(2048)
- cred_json = {
- private_key_id: 'a_private_key_id',
- private_key: @key.to_pem,
- client_email: 'app@developer.gserviceaccount.com',
- client_id: 'app.apps.googleusercontent.com',
- type: 'service_account'
- }
- cred_json_text = MultiJson.dump(cred_json)
- @client = GRPC::Auth::ServiceAccountCredentials.new(
- 'https://www.googleapis.com/auth/userinfo.profile',
- StringIO.new(cred_json_text))
- end
-
- def make_auth_stubs(with_access_token: '')
- Faraday::Adapter::Test::Stubs.new do |stub|
- stub.post('/oauth2/v3/token') do |env|
- params = Addressable::URI.form_unencode(env[:body])
- _claim, _header = JWT.decode(params.assoc('assertion').last,
- @key.public_key)
- want = ['grant_type', 'urn:ietf:params:oauth:grant-type:jwt-bearer']
- expect(params.assoc('grant_type')).to eq(want)
- build_json_response(
- 'access_token' => with_access_token,
- 'token_type' => 'Bearer',
- 'expires_in' => 3600
- )
- end
- end
- end
-
- it_behaves_like 'apply/apply! are OK'
-end
diff --git a/src/ruby/spec/auth/signet_spec.rb b/src/ruby/spec/auth/signet_spec.rb
deleted file mode 100644
index 1712edf..0000000
--- a/src/ruby/spec/auth/signet_spec.rb
+++ /dev/null
@@ -1,70 +0,0 @@
-# Copyright 2015, Google Inc.
-# All rights reserved.
-#
-# 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.
-
-spec_dir = File.expand_path(File.join(File.dirname(__FILE__)))
-$LOAD_PATH.unshift(spec_dir)
-$LOAD_PATH.uniq!
-
-require 'apply_auth_examples'
-require 'grpc/auth/signet'
-require 'jwt'
-require 'openssl'
-require 'spec_helper'
-
-describe Signet::OAuth2::Client do
- before(:example) do
- @key = OpenSSL::PKey::RSA.new(2048)
- @client = Signet::OAuth2::Client.new(
- token_credential_uri: 'https://accounts.google.com/o/oauth2/token',
- scope: 'https://www.googleapis.com/auth/userinfo.profile',
- issuer: 'app@example.com',
- audience: 'https://accounts.google.com/o/oauth2/token',
- signing_key: @key
- )
- end
-
- def make_auth_stubs(with_access_token: '')
- Faraday::Adapter::Test::Stubs.new do |stub|
- stub.post('/o/oauth2/token') do |env|
- params = Addressable::URI.form_unencode(env[:body])
- _claim, _header = JWT.decode(params.assoc('assertion').last,
- @key.public_key)
- want = ['grant_type', 'urn:ietf:params:oauth:grant-type:jwt-bearer']
- expect(params.assoc('grant_type')).to eq(want)
- build_json_response(
- 'access_token' => with_access_token,
- 'token_type' => 'Bearer',
- 'expires_in' => 3600
- )
- end
- end
- end
-
- it_behaves_like 'apply/apply! are OK'
-end
diff --git a/src/ruby/spec/credentials_spec.rb b/src/ruby/spec/credentials_spec.rb
index 001fecd..fc97d11 100644
--- a/src/ruby/spec/credentials_spec.rb
+++ b/src/ruby/spec/credentials_spec.rb
@@ -68,10 +68,4 @@
expect { cred1.compose(cred2) }.to_not raise_error
end
end
-
- describe 'Credentials#default' do
- it 'is not implemented yet' do
- expect { Credentials.default }.to raise_error RuntimeError
- end
- end
end
diff --git a/test/compiler/python_plugin_test.py b/test/compiler/python_plugin_test.py
index b0c9ec6..3919de1 100644
--- a/test/compiler/python_plugin_test.py
+++ b/test/compiler/python_plugin_test.py
@@ -40,8 +40,24 @@
from grpc.framework.face import exceptions
from grpc.framework.foundation import future
+# Identifiers of entities we expect to find in the generated module.
+SERVICER_IDENTIFIER = 'EarlyAdopterTestServiceServicer'
+SERVER_IDENTIFIER = 'EarlyAdopterTestServiceServer'
+STUB_IDENTIFIER = 'EarlyAdopterTestServiceStub'
+SERVER_FACTORY_IDENTIFIER = 'early_adopter_create_TestService_server'
+STUB_FACTORY_IDENTIFIER = 'early_adopter_create_TestService_stub'
+
+# Timeouts and delays.
+SHORT_TIMEOUT = 0.1
+NORMAL_TIMEOUT = 1
+LONG_TIMEOUT = 2
+DOES_NOT_MATTER_DELAY = 0
+NO_DELAY = 0
+LONG_DELAY = 1
+
# Assigned in __main__.
_build_mode = None
+_port = None
class _ServicerMethods(object):
@@ -71,14 +87,14 @@
while self._paused:
time.sleep(0)
- def UnaryCall(self, request):
+ def UnaryCall(self, request, context):
response = self.test_pb2.SimpleResponse()
response.payload.payload_type = self.test_pb2.COMPRESSABLE
response.payload.payload_compressable = 'a' * request.response_size
self._control()
return response
- def StreamingOutputCall(self, request):
+ def StreamingOutputCall(self, request, context):
for parameter in request.response_parameters:
response = self.test_pb2.StreamingOutputCallResponse()
response.payload.payload_type = self.test_pb2.COMPRESSABLE
@@ -86,7 +102,7 @@
self._control()
yield response
- def StreamingInputCall(self, request_iter):
+ def StreamingInputCall(self, request_iter, context):
response = self.test_pb2.StreamingInputCallResponse()
aggregated_payload_size = 0
for request in request_iter:
@@ -95,7 +111,7 @@
self._control()
return response
- def FullDuplexCall(self, request_iter):
+ def FullDuplexCall(self, request_iter, context):
for request in request_iter:
for parameter in request.response_parameters:
response = self.test_pb2.StreamingOutputCallResponse()
@@ -104,7 +120,7 @@
self._control()
yield response
- def HalfDuplexCall(self, request_iter):
+ def HalfDuplexCall(self, request_iter, context):
responses = []
for request in request_iter:
for parameter in request.response_parameters:
@@ -117,7 +133,7 @@
yield response
-def CreateService(test_pb2, delay=0, timeout=1):
+def _CreateService(test_pb2, delay):
"""Provides a servicer backend and a stub.
The servicer is just the implementation
@@ -136,28 +152,30 @@
A two-tuple (servicer, stub), where the servicer is the back-end of the
service bound to the stub.
"""
- class Servicer(test_pb2.TestServiceServicer):
-
- def UnaryCall(self, request):
- return servicer_methods.UnaryCall(request)
-
- def StreamingOutputCall(self, request):
- return servicer_methods.StreamingOutputCall(request)
-
- def StreamingInputCall(self, request_iter):
- return servicer_methods.StreamingInputCall(request_iter)
-
- def FullDuplexCall(self, request_iter):
- return servicer_methods.FullDuplexCall(request_iter)
-
- def HalfDuplexCall(self, request_iter):
- return servicer_methods.HalfDuplexCall(request_iter)
-
servicer_methods = _ServicerMethods(test_pb2, delay)
+
+ class Servicer(getattr(test_pb2, SERVICER_IDENTIFIER)):
+
+ def UnaryCall(self, request, context):
+ return servicer_methods.UnaryCall(request, context)
+
+ def StreamingOutputCall(self, request, context):
+ return servicer_methods.StreamingOutputCall(request, context)
+
+ def StreamingInputCall(self, request_iter, context):
+ return servicer_methods.StreamingInputCall(request_iter, context)
+
+ def FullDuplexCall(self, request_iter, context):
+ return servicer_methods.FullDuplexCall(request_iter, context)
+
+ def HalfDuplexCall(self, request_iter, context):
+ return servicer_methods.HalfDuplexCall(request_iter, context)
+
servicer = Servicer()
- linked_pair = test_pb2.mock_TestService(servicer, timeout)
- stub = linked_pair.stub
- return servicer_methods, stub
+ server = getattr(test_pb2, SERVER_FACTORY_IDENTIFIER)(servicer, _port,
+ None, None)
+ stub = getattr(test_pb2, STUB_FACTORY_IDENTIFIER)('localhost', _port)
+ return servicer_methods, stub, server
def StreamingInputRequest(test_pb2):
@@ -198,19 +216,20 @@
def setUp(self):
protoc_command = '../../bins/%s/protobuf/protoc' % _build_mode
protoc_plugin_filename = '../../bins/%s/grpc_python_plugin' % _build_mode
- test_proto_filename = '../cpp/interop/test.proto'
+ test_proto_filename = './test.proto'
if not os.path.isfile(protoc_command):
# Assume that if we haven't built protoc that it's on the system.
protoc_command = 'protoc'
- # ensure that the output directory exists
- outdir = '../../gens/test/compiler/python/'
+ # Ensure that the output directory exists.
+ outdir = '../../gens/test/compiler/python'
try:
os.makedirs(outdir)
except OSError as exception:
if exception.errno != errno.EEXIST:
raise
+ # Invoke protoc with the plugin.
cmd = [
protoc_command,
'--plugin=protoc-gen-python-grpc=%s' % protoc_plugin_filename,
@@ -222,215 +241,231 @@
subprocess.call(' '.join(cmd), shell=True)
sys.path.append(outdir)
- self.delay = 1 # seconds
- self.timeout = 2 # seconds
+ # TODO(atash): Figure out which of theses tests is hanging flakily with small
+ # probability.
def testImportAttributes(self):
- # check that we can access the members
+ # check that we can access the generated module and its members.
import test_pb2 # pylint: disable=g-import-not-at-top
- self.assertIsNotNone(getattr(test_pb2, 'TestServiceServicer', None))
- self.assertIsNotNone(getattr(test_pb2, 'TestServiceService', None))
- self.assertIsNotNone(getattr(test_pb2, 'TestServiceStub', None))
+ self.assertIsNotNone(getattr(test_pb2, SERVICER_IDENTIFIER, None))
+ self.assertIsNotNone(getattr(test_pb2, SERVER_IDENTIFIER, None))
+ self.assertIsNotNone(getattr(test_pb2, STUB_IDENTIFIER, None))
+ self.assertIsNotNone(getattr(test_pb2, SERVER_FACTORY_IDENTIFIER, None))
+ self.assertIsNotNone(getattr(test_pb2, STUB_FACTORY_IDENTIFIER, None))
+
+ def testUpDown(self):
+ import test_pb2
+ servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
+ request = test_pb2.SimpleRequest(response_size=13)
+ with server, stub:
+ pass
def testUnaryCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2)
+ servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
request = test_pb2.SimpleRequest(response_size=13)
- response = stub.UnaryCall(request)
- expected_response = servicer.UnaryCall(request)
+ with server, stub:
+ response = stub.UnaryCall(request, NORMAL_TIMEOUT)
+ expected_response = servicer.UnaryCall(request, None)
self.assertEqual(expected_response, response)
def testUnaryCallAsync(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(
- test_pb2, delay=self.delay, timeout=self.timeout)
+ servicer, stub, server = _CreateService(test_pb2, LONG_DELAY)
request = test_pb2.SimpleRequest(response_size=13)
- # TODO(atash): consider using the 'profile' module? Does it even work here?
- start_time = time.clock()
- response_future = stub.UnaryCall.async(request)
- self.assertGreater(self.delay, time.clock() - start_time)
- response = response_future.result()
- expected_response = servicer.UnaryCall(request)
+ with server, stub:
+ start_time = time.clock()
+ response_future = stub.UnaryCall.async(request, LONG_TIMEOUT)
+ # Check that we didn't block on the asynchronous call.
+ self.assertGreater(LONG_DELAY, time.clock() - start_time)
+ response = response_future.result()
+ expected_response = servicer.UnaryCall(request, None)
self.assertEqual(expected_response, response)
def testUnaryCallAsyncExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top
# set the timeout super low...
- servicer, stub = CreateService(test_pb2, delay=1, timeout=0.1)
+ servicer, stub, server = _CreateService(test_pb2,
+ delay=DOES_NOT_MATTER_DELAY)
request = test_pb2.SimpleRequest(response_size=13)
- with servicer.pause():
- response_future = stub.UnaryCall.async(request)
- with self.assertRaises(exceptions.ExpirationError):
- response_future.result()
+ with server, stub:
+ with servicer.pause():
+ response_future = stub.UnaryCall.async(request, SHORT_TIMEOUT)
+ with self.assertRaises(exceptions.ExpirationError):
+ response_future.result()
def testUnaryCallAsyncCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2)
+ servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
request = test_pb2.SimpleRequest(response_size=13)
- with servicer.pause():
- response_future = stub.UnaryCall.async(request)
- response_future.cancel()
- self.assertTrue(response_future.cancelled())
+ with server, stub:
+ with servicer.pause():
+ response_future = stub.UnaryCall.async(request, 1)
+ response_future.cancel()
+ self.assertTrue(response_future.cancelled())
def testUnaryCallAsyncFailed(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2)
+ servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
request = test_pb2.SimpleRequest(response_size=13)
- with servicer.fail():
- response_future = stub.UnaryCall.async(request)
- self.assertIsNotNone(response_future.exception())
+ with server, stub:
+ with servicer.fail():
+ response_future = stub.UnaryCall.async(request, NORMAL_TIMEOUT)
+ self.assertIsNotNone(response_future.exception())
def testStreamingOutputCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2)
+ servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
request = StreamingOutputRequest(test_pb2)
- responses = stub.StreamingOutputCall(request)
- expected_responses = servicer.StreamingOutputCall(request)
- for check in itertools.izip_longest(expected_responses, responses):
- expected_response, response = check
- self.assertEqual(expected_response, response)
+ with server, stub:
+ responses = stub.StreamingOutputCall(request, NORMAL_TIMEOUT)
+ expected_responses = servicer.StreamingOutputCall(request, None)
+ for check in itertools.izip_longest(expected_responses, responses):
+ expected_response, response = check
+ self.assertEqual(expected_response, response)
- def testStreamingOutputCallAsync(self):
+ def testStreamingOutputCallExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2, timeout=self.timeout)
+ servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
request = StreamingOutputRequest(test_pb2)
- responses = stub.StreamingOutputCall.async(request)
- expected_responses = servicer.StreamingOutputCall(request)
- for check in itertools.izip_longest(expected_responses, responses):
- expected_response, response = check
- self.assertEqual(expected_response, response)
+ with server, stub:
+ with servicer.pause():
+ responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT)
+ with self.assertRaises(exceptions.ExpirationError):
+ list(responses)
- def testStreamingOutputCallAsyncExpired(self):
+ def testStreamingOutputCallCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2, timeout=0.1)
+ unused_servicer, stub, server = _CreateService(test_pb2,
+ DOES_NOT_MATTER_DELAY)
request = StreamingOutputRequest(test_pb2)
- with servicer.pause():
- responses = stub.StreamingOutputCall.async(request)
- with self.assertRaises(exceptions.ExpirationError):
- list(responses)
-
- def testStreamingOutputCallAsyncCancelled(self):
- import test_pb2 # pylint: disable=g-import-not-at-top
- _, stub = CreateService(test_pb2, timeout=0.1)
- request = StreamingOutputRequest(test_pb2)
- responses = stub.StreamingOutputCall.async(request)
- next(responses)
- responses.cancel()
- with self.assertRaises(future.CancelledError):
+ with server, stub:
+ responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT)
next(responses)
-
- def testStreamingOutputCallAsyncFailed(self):
- import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2, timeout=0.1)
- request = StreamingOutputRequest(test_pb2)
- with servicer.fail():
- responses = stub.StreamingOutputCall.async(request)
- self.assertIsNotNone(responses)
- with self.assertRaises(exceptions.ServicerError):
+ responses.cancel()
+ with self.assertRaises(future.CancelledError):
next(responses)
+ @unittest.skip('TODO(atash,nathaniel): figure out why this times out '
+ 'instead of raising the proper error.')
+ def testStreamingOutputCallFailed(self):
+ import test_pb2 # pylint: disable=g-import-not-at-top
+ servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
+ request = StreamingOutputRequest(test_pb2)
+ with server, stub:
+ with servicer.fail():
+ responses = stub.StreamingOutputCall(request, 1)
+ self.assertIsNotNone(responses)
+ with self.assertRaises(exceptions.ServicerError):
+ next(responses)
+
def testStreamingInputCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2)
- response = stub.StreamingInputCall(StreamingInputRequest(test_pb2))
+ servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
+ with server, stub:
+ response = stub.StreamingInputCall(StreamingInputRequest(test_pb2),
+ NORMAL_TIMEOUT)
expected_response = servicer.StreamingInputCall(
- StreamingInputRequest(test_pb2))
+ StreamingInputRequest(test_pb2), None)
self.assertEqual(expected_response, response)
def testStreamingInputCallAsync(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(
- test_pb2, delay=self.delay, timeout=self.timeout)
- start_time = time.clock()
- response_future = stub.StreamingInputCall.async(
- StreamingInputRequest(test_pb2))
- self.assertGreater(self.delay, time.clock() - start_time)
- response = response_future.result()
+ servicer, stub, server = _CreateService(
+ test_pb2, LONG_DELAY)
+ with server, stub:
+ start_time = time.clock()
+ response_future = stub.StreamingInputCall.async(
+ StreamingInputRequest(test_pb2), LONG_TIMEOUT)
+ self.assertGreater(LONG_DELAY, time.clock() - start_time)
+ response = response_future.result()
expected_response = servicer.StreamingInputCall(
- StreamingInputRequest(test_pb2))
+ StreamingInputRequest(test_pb2), None)
self.assertEqual(expected_response, response)
def testStreamingInputCallAsyncExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top
# set the timeout super low...
- servicer, stub = CreateService(test_pb2, delay=1, timeout=0.1)
- with servicer.pause():
- response_future = stub.StreamingInputCall.async(
- StreamingInputRequest(test_pb2))
- with self.assertRaises(exceptions.ExpirationError):
- response_future.result()
- self.assertIsInstance(
- response_future.exception(), exceptions.ExpirationError)
+ servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
+ with server, stub:
+ with servicer.pause():
+ response_future = stub.StreamingInputCall.async(
+ StreamingInputRequest(test_pb2), SHORT_TIMEOUT)
+ with self.assertRaises(exceptions.ExpirationError):
+ response_future.result()
+ self.assertIsInstance(
+ response_future.exception(), exceptions.ExpirationError)
def testStreamingInputCallAsyncCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2)
- with servicer.pause():
- response_future = stub.StreamingInputCall.async(
- StreamingInputRequest(test_pb2))
- response_future.cancel()
- self.assertTrue(response_future.cancelled())
- with self.assertRaises(future.CancelledError):
- response_future.result()
+ servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
+ with server, stub:
+ with servicer.pause():
+ response_future = stub.StreamingInputCall.async(
+ StreamingInputRequest(test_pb2), NORMAL_TIMEOUT)
+ response_future.cancel()
+ self.assertTrue(response_future.cancelled())
+ with self.assertRaises(future.CancelledError):
+ response_future.result()
def testStreamingInputCallAsyncFailed(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2)
- with servicer.fail():
- response_future = stub.StreamingInputCall.async(
- StreamingInputRequest(test_pb2))
- self.assertIsNotNone(response_future.exception())
+ servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
+ with server, stub:
+ with servicer.fail():
+ response_future = stub.StreamingInputCall.async(
+ StreamingInputRequest(test_pb2), SHORT_TIMEOUT)
+ self.assertIsNotNone(response_future.exception())
def testFullDuplexCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2)
- responses = stub.FullDuplexCall(FullDuplexRequest(test_pb2))
- expected_responses = servicer.FullDuplexCall(FullDuplexRequest(test_pb2))
- for check in itertools.izip_longest(expected_responses, responses):
- expected_response, response = check
- self.assertEqual(expected_response, response)
+ servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
+ with server, stub:
+ responses = stub.FullDuplexCall(FullDuplexRequest(test_pb2),
+ NORMAL_TIMEOUT)
+ expected_responses = servicer.FullDuplexCall(FullDuplexRequest(test_pb2),
+ None)
+ for check in itertools.izip_longest(expected_responses, responses):
+ expected_response, response = check
+ self.assertEqual(expected_response, response)
- def testFullDuplexCallAsync(self):
+ def testFullDuplexCallExpired(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2, timeout=self.timeout)
- responses = stub.FullDuplexCall.async(FullDuplexRequest(test_pb2))
- expected_responses = servicer.FullDuplexCall(FullDuplexRequest(test_pb2))
- for check in itertools.izip_longest(expected_responses, responses):
- expected_response, response = check
- self.assertEqual(expected_response, response)
-
- def testFullDuplexCallAsyncExpired(self):
- import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2, timeout=0.1)
+ servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
request = FullDuplexRequest(test_pb2)
- with servicer.pause():
- responses = stub.FullDuplexCall.async(request)
- with self.assertRaises(exceptions.ExpirationError):
- list(responses)
+ with server, stub:
+ with servicer.pause():
+ responses = stub.FullDuplexCall(request, SHORT_TIMEOUT)
+ with self.assertRaises(exceptions.ExpirationError):
+ list(responses)
- def testFullDuplexCallAsyncCancelled(self):
+ def testFullDuplexCallCancelled(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- _, stub = CreateService(test_pb2, timeout=0.1)
- request = FullDuplexRequest(test_pb2)
- responses = stub.FullDuplexCall.async(request)
- next(responses)
- responses.cancel()
- with self.assertRaises(future.CancelledError):
+ unused_servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
+ with server, stub:
+ request = FullDuplexRequest(test_pb2)
+ responses = stub.FullDuplexCall(request, NORMAL_TIMEOUT)
next(responses)
-
- def testFullDuplexCallAsyncFailed(self):
- import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2, timeout=0.1)
- request = FullDuplexRequest(test_pb2)
- with servicer.fail():
- responses = stub.FullDuplexCall.async(request)
- self.assertIsNotNone(responses)
- with self.assertRaises(exceptions.ServicerError):
+ responses.cancel()
+ with self.assertRaises(future.CancelledError):
next(responses)
+ @unittest.skip('TODO(atash,nathaniel): figure out why this hangs forever '
+ 'and fix.')
+ def testFullDuplexCallFailed(self):
+ import test_pb2 # pylint: disable=g-import-not-at-top
+ servicer, stub, server = _CreateService(test_pb2, DOES_NOT_MATTER_DELAY)
+ request = FullDuplexRequest(test_pb2)
+ with server, stub:
+ with servicer.fail():
+ responses = stub.FullDuplexCall(request, NORMAL_TIMEOUT)
+ self.assertIsNotNone(responses)
+ with self.assertRaises(exceptions.ServicerError):
+ next(responses)
+
def testHalfDuplexCall(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- servicer, stub = CreateService(test_pb2)
+ servicer, stub, server = _CreateService(test_pb2, NO_DELAY)
def HalfDuplexRequest():
request = test_pb2.StreamingOutputCallRequest()
request.response_parameters.add(size=1, interval_us=0)
@@ -439,15 +474,16 @@
request.response_parameters.add(size=2, interval_us=0)
request.response_parameters.add(size=3, interval_us=0)
yield request
- responses = stub.HalfDuplexCall(HalfDuplexRequest())
- expected_responses = servicer.HalfDuplexCall(HalfDuplexRequest())
- for check in itertools.izip_longest(expected_responses, responses):
- expected_response, response = check
- self.assertEqual(expected_response, response)
+ with server, stub:
+ responses = stub.HalfDuplexCall(HalfDuplexRequest(), NORMAL_TIMEOUT)
+ expected_responses = servicer.HalfDuplexCall(HalfDuplexRequest(), None)
+ for check in itertools.izip_longest(expected_responses, responses):
+ expected_response, response = check
+ self.assertEqual(expected_response, response)
- def testHalfDuplexCallAsyncWedged(self):
+ def testHalfDuplexCallWedged(self):
import test_pb2 # pylint: disable=g-import-not-at-top
- _, stub = CreateService(test_pb2, timeout=1)
+ _, stub, server = _CreateService(test_pb2, NO_DELAY)
wait_flag = [False]
@contextlib.contextmanager
def wait(): # pylint: disable=invalid-name
@@ -461,20 +497,25 @@
yield request
while wait_flag[0]:
time.sleep(0.1)
- with wait():
- responses = stub.HalfDuplexCall.async(HalfDuplexRequest())
- # half-duplex waits for the client to send all info
- with self.assertRaises(exceptions.ExpirationError):
- next(responses)
+ with server, stub:
+ with wait():
+ responses = stub.HalfDuplexCall(HalfDuplexRequest(), NORMAL_TIMEOUT)
+ # half-duplex waits for the client to send all info
+ with self.assertRaises(exceptions.ExpirationError):
+ next(responses)
if __name__ == '__main__':
os.chdir(os.path.dirname(sys.argv[0]))
- parser = argparse.ArgumentParser(description='Run Python compiler plugin test.')
- parser.add_argument('--build_mode', dest='build_mode', type=str, default='dbg',
- help='The build mode of the targets to test, e.g. '
- '"dbg", "opt", "asan", etc.')
+ parser = argparse.ArgumentParser(
+ description='Run Python compiler plugin test.')
+ parser.add_argument(
+ '--build_mode', dest='build_mode', type=str, default='dbg',
+ help='The build mode of the targets to test, e.g. "dbg", "opt", "asan", '
+ 'etc.')
+ parser.add_argument('--port', dest='port', type=int, default=0)
args, remainder = parser.parse_known_args()
_build_mode = args.build_mode
+ _port = args.port
sys.argv[1:] = remainder
unittest.main()
diff --git a/test/compiler/test.proto b/test/compiler/test.proto
index ed7c6a7..1714de7 100644
--- a/test/compiler/test.proto
+++ b/test/compiler/test.proto
@@ -32,7 +32,8 @@
// This file is duplicated around the code base. See GitHub issue #526.
syntax = "proto2";
-package grpc.testing;
+// TODO(atash): Investigate this statement's utility.
+// package grpc.testing;
enum PayloadType {
// Compressable text format.
diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c
index e7183cc..66b76dc 100644
--- a/test/core/end2end/dualstack_socket_test.c
+++ b/test/core/end2end/dualstack_socket_test.c
@@ -75,6 +75,10 @@
gpr_timespec deadline;
int got_port;
+ if (port == 0) {
+ port = grpc_pick_unused_port_or_die();
+ }
+
gpr_join_host_port(&server_hostport, server_host, port);
/* Create server. */
@@ -179,7 +183,6 @@
int main(int argc, char **argv) {
int do_ipv6 = 1;
- int fixed_port;
grpc_test_init(argc, argv);
grpc_init();
@@ -189,32 +192,28 @@
do_ipv6 = 0;
}
- for (fixed_port = 0; fixed_port <= 1; fixed_port++) {
- int port = fixed_port ? grpc_pick_unused_port_or_die() : 0;
-
/* For coverage, test with and without dualstack sockets. */
- for (grpc_forbid_dualstack_sockets_for_testing = 0;
- grpc_forbid_dualstack_sockets_for_testing <= 1;
- grpc_forbid_dualstack_sockets_for_testing++) {
- /* :: and 0.0.0.0 are handled identically. */
- test_connect("::", "127.0.0.1", port, 1);
- test_connect("::", "::ffff:127.0.0.1", port, 1);
- test_connect("::", "localhost", port, 1);
- test_connect("0.0.0.0", "127.0.0.1", port, 1);
- test_connect("0.0.0.0", "::ffff:127.0.0.1", port, 1);
- test_connect("0.0.0.0", "localhost", port, 1);
- if (do_ipv6) {
- test_connect("::", "::1", port, 1);
- test_connect("0.0.0.0", "::1", port, 1);
- }
+ for (grpc_forbid_dualstack_sockets_for_testing = 0;
+ grpc_forbid_dualstack_sockets_for_testing <= 1;
+ grpc_forbid_dualstack_sockets_for_testing++) {
+ /* :: and 0.0.0.0 are handled identically. */
+ test_connect("::", "127.0.0.1", 0, 1);
+ test_connect("::", "::ffff:127.0.0.1", 0, 1);
+ test_connect("::", "localhost", 0, 1);
+ test_connect("0.0.0.0", "127.0.0.1", 0, 1);
+ test_connect("0.0.0.0", "::ffff:127.0.0.1", 0, 1);
+ test_connect("0.0.0.0", "localhost", 0, 1);
+ if (do_ipv6) {
+ test_connect("::", "::1", 0, 1);
+ test_connect("0.0.0.0", "::1", 0, 1);
+ }
- /* These only work when the families agree. */
- test_connect("127.0.0.1", "127.0.0.1", port, 1);
- if (do_ipv6) {
- test_connect("::1", "::1", port, 1);
- test_connect("::1", "127.0.0.1", port, 0);
- test_connect("127.0.0.1", "::1", port, 0);
- }
+ /* These only work when the families agree. */
+ test_connect("127.0.0.1", "127.0.0.1", 0, 1);
+ if (do_ipv6) {
+ test_connect("::1", "::1", 0, 1);
+ test_connect("::1", "127.0.0.1", 0, 0);
+ test_connect("127.0.0.1", "::1", 0, 0);
}
}
diff --git a/test/core/end2end/tests/max_concurrent_streams.c b/test/core/end2end/tests/max_concurrent_streams.c
index d85c935..af29e17 100644
--- a/test/core/end2end/tests/max_concurrent_streams.c
+++ b/test/core/end2end/tests/max_concurrent_streams.c
@@ -196,7 +196,7 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_invoke_old(c2, f.client_cq, tag(401), tag(402), 0));
GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done_old(c1, tag(303)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done_old(c2, tag(303)));
+ GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done_old(c2, tag(403)));
ev = grpc_completion_queue_next(
f.client_cq, gpr_time_add(gpr_now(), gpr_time_from_seconds(10)));
@@ -230,8 +230,8 @@
/* first request is finished, we should be able to start the second */
cq_expect_finished_with_status(v_client, tag(live_call + 2),
GRPC_STATUS_UNIMPLEMENTED, "xyz", NULL);
- cq_expect_finish_accepted(v_client, tag(live_call + 3), GRPC_OP_OK);
live_call = (live_call == 300) ? 400 : 300;
+ cq_expect_finish_accepted(v_client, tag(live_call + 3), GRPC_OP_OK);
cq_verify(v_client);
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call_old(f.server, tag(200)));
diff --git a/test/core/end2end/tests/max_concurrent_streams_legacy.c b/test/core/end2end/tests/max_concurrent_streams_legacy.c
index d85c935..af29e17 100644
--- a/test/core/end2end/tests/max_concurrent_streams_legacy.c
+++ b/test/core/end2end/tests/max_concurrent_streams_legacy.c
@@ -196,7 +196,7 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_invoke_old(c2, f.client_cq, tag(401), tag(402), 0));
GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done_old(c1, tag(303)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done_old(c2, tag(303)));
+ GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done_old(c2, tag(403)));
ev = grpc_completion_queue_next(
f.client_cq, gpr_time_add(gpr_now(), gpr_time_from_seconds(10)));
@@ -230,8 +230,8 @@
/* first request is finished, we should be able to start the second */
cq_expect_finished_with_status(v_client, tag(live_call + 2),
GRPC_STATUS_UNIMPLEMENTED, "xyz", NULL);
- cq_expect_finish_accepted(v_client, tag(live_call + 3), GRPC_OP_OK);
live_call = (live_call == 300) ? 400 : 300;
+ cq_expect_finish_accepted(v_client, tag(live_call + 3), GRPC_OP_OK);
cq_verify(v_client);
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call_old(f.server, tag(200)));
diff --git a/test/core/end2end/tests/simple_request.c b/test/core/end2end/tests/simple_request.c
index 1263155..4d4d48a 100644
--- a/test/core/end2end/tests/simple_request.c
+++ b/test/core/end2end/tests/simple_request.c
@@ -122,7 +122,7 @@
int was_cancelled = 2;
c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
- "foo.test.google.fr", deadline);
+ "foo.test.google.fr:1234", deadline);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
@@ -177,7 +177,7 @@
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
- GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr"));
+ GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234"));
GPR_ASSERT(was_cancelled == 0);
gpr_free(details);
diff --git a/tools/gce_setup/cloud_prod_runner.sh b/tools/gce_setup/cloud_prod_runner.sh
index 4732f95..e11185c 100755
--- a/tools/gce_setup/cloud_prod_runner.sh
+++ b/tools/gce_setup/cloud_prod_runner.sh
@@ -31,7 +31,7 @@
main() {
source grpc_docker.sh
- test_cases=(large_unary empty_unary ping_pong client_streaming server_streaming service_account_creds compute_engine_creds)
+ test_cases=(large_unary empty_unary ping_pong client_streaming server_streaming cancel_after_begin cancel_after_first_response)
auth_test_cases=(service_account_creds compute_engine_creds)
clients=(cxx java go ruby node)
for test_case in "${test_cases[@]}"
diff --git a/tools/gce_setup/grpc_docker.sh b/tools/gce_setup/grpc_docker.sh
index 231625e..df8605b 100755
--- a/tools/gce_setup/grpc_docker.sh
+++ b/tools/gce_setup/grpc_docker.sh
@@ -926,7 +926,8 @@
local cmd_prefix="sudo docker run grpc/go /bin/bash -c"
local test_script="cd src/google.golang.org/grpc/interop/client"
local test_script+=" && go run client.go --use_tls=true"
- local gfe_flags=" --tls_ca_file=\"\" --tls_server_name=\"\" --server_port=443 --server_host=grpc-test.sandbox.google.com"
+ local gfe_flags=$(_grpc_prod_gfe_flags)
+ local gfe_flags+=" --tls_ca_file=\"\""
local added_gfe_flags=$(_grpc_svc_acc_test_flags)
local the_cmd="$cmd_prefix '$test_script $gfe_flags $added_gfe_flags $@'"
echo $the_cmd
@@ -941,7 +942,8 @@
local cmd_prefix="sudo docker run grpc/go /bin/bash -c"
local test_script="cd src/google.golang.org/grpc/interop/client"
local test_script+=" && go run client.go --use_tls=true"
- local gfe_flags=" --tls_ca_file=\"\" --tls_server_name=\"\" --server_port=443 --server_host=grpc-test.sandbox.google.com"
+ local gfe_flags=$(_grpc_prod_gfe_flags)
+ local gfe_flags+=" --tls_ca_file=\"\""
local added_gfe_flags=$(_grpc_gce_test_flags)
local the_cmd="$cmd_prefix '$test_script $gfe_flags $added_gfe_flags $@'"
echo $the_cmd
@@ -957,9 +959,10 @@
local test_script="/var/local/git/grpc/src/ruby/bin/interop/interop_client.rb"
local test_script+=" --use_tls"
local gfe_flags=$(_grpc_prod_gfe_flags)
- local added_gfe_flags=$(_grpc_svc_acc_test_flags)
+ local added_gfe_flags=$(_grpc_default_creds_test_flags)
local env_prefix="SSL_CERT_FILE=/cacerts/roots.pem"
- local the_cmd="$cmd_prefix '$env_prefix ruby $test_script $gfe_flags $added_gfe_flag $@'"
+ env_prefix+=" GOOGLE_APPLICATION_CREDENTIALS=/service_account/stubbyCloudTestingTest-7dd63462c60c.json"
+ local the_cmd="$cmd_prefix '$env_prefix ruby $test_script $gfe_flags $added_gfe_flags $@'"
echo $the_cmd
}
@@ -975,7 +978,7 @@
local gfe_flags=$(_grpc_prod_gfe_flags)
local added_gfe_flags=$(_grpc_gce_test_flags)
local env_prefix="SSL_CERT_FILE=/cacerts/roots.pem"
- local the_cmd="$cmd_prefix '$env_prefix ruby $test_script $gfe_flags $added_gfe_flag $@'"
+ local the_cmd="$cmd_prefix '$env_prefix ruby $test_script $gfe_flags $added_gfe_flags $@'"
echo $the_cmd
}
@@ -1001,7 +1004,8 @@
local cmd_prefix="sudo docker run grpc/go /bin/bash -c"
local test_script="cd src/google.golang.org/grpc/interop/client"
local test_script+=" && go run client.go --use_tls=true"
- local gfe_flags=" --tls_ca_file=\"\" --tls_server_name=\"\" --server_port=443 --server_host=grpc-test.sandbox.google.com"
+ local gfe_flags=$(_grpc_prod_gfe_flags)
+ local gfe_flags+=" --tls_ca_file=\"\""
local the_cmd="$cmd_prefix '$test_script $gfe_flags $@'"
echo $the_cmd
}
@@ -1153,6 +1157,11 @@
echo " --service_account_key_file=/service_account/stubbyCloudTestingTest-7dd63462c60c.json --oauth_scope=https://www.googleapis.com/auth/xapi.zoo"
}
+# default credentials test flag
+_grpc_default_creds_test_flags() {
+ echo " --oauth_scope=https://www.googleapis.com/auth/xapi.zoo"
+}
+
# outputs the flags passed to the gcloud auth tests
_grpc_gce_test_flags() {
echo " --default_service_account=155450119199-r5aaqa2vqoa9g5mv2m6s3m1l293rlmel@developer.gserviceaccount.com --oauth_scope=https://www.googleapis.com/auth/xapi.zoo"
diff --git a/tools/gce_setup/interop_test_runner.sh b/tools/gce_setup/interop_test_runner.sh
index 465c2ab..5f8c0e7 100755
--- a/tools/gce_setup/interop_test_runner.sh
+++ b/tools/gce_setup/interop_test_runner.sh
@@ -35,7 +35,7 @@
main() {
source grpc_docker.sh
- test_cases=(large_unary empty_unary ping_pong client_streaming server_streaming cancel_after_being cancel_after_first_response)
+ test_cases=(large_unary empty_unary ping_pong client_streaming server_streaming cancel_after_begin cancel_after_first_response)
clients=(cxx java go ruby node)
servers=(cxx java go ruby node python)
for test_case in "${test_cases[@]}"
diff --git a/tools/run_tests/run_python.sh b/tools/run_tests/run_python.sh
index fe40b51..19f1458 100755
--- a/tools/run_tests/run_python.sh
+++ b/tools/run_tests/run_python.sh
@@ -37,7 +37,8 @@
export LD_LIBRARY_PATH=$root/libs/opt
source python2.7_virtual_environment/bin/activate
# TODO(issue 215): Properly itemize these in run_tests.py so that they can be parallelized.
-python2.7 -B test/compiler/python_plugin_test.py
+# TODO(atash): Enable dynamic unused port discovery for this test.
+python2.7 -B test/compiler/python_plugin_test.py --build_mode=opt --port=40987
python2.7 -B -m grpc._adapter._blocking_invocation_inline_service_test
python2.7 -B -m grpc._adapter._c_test
python2.7 -B -m grpc._adapter._event_invocation_synchronous_event_service_test
@@ -45,6 +46,7 @@
python2.7 -B -m grpc._adapter._links_test
python2.7 -B -m grpc._adapter._lonely_rear_link_test
python2.7 -B -m grpc._adapter._low_test
+python2.7 -B -m grpc.early_adopter.implementations_test
python2.7 -B -m grpc.framework.assembly.implementations_test
python2.7 -B -m grpc.framework.base.packets.implementations_test
python2.7 -B -m grpc.framework.face.blocking_invocation_inline_service_test