diff --git a/INSTALL b/INSTALL index de13d65d413..a0df57dcd35 100644 --- a/INSTALL +++ b/INSTALL @@ -11,12 +11,12 @@ wiki pages: On Linux (Debian): - Note: you will need to add the Debian 'testing' distribution to your sources + Note: you will need to add the Debian 'jessie-backports' distribution to your sources file first. Add the following line to your `/etc/apt/sources.list` file: - deb http://ftp.us.debian.org/debian testing main contrib non-free + deb http://http.debian.net/debian jessie-backports main Install the gRPC library: diff --git a/src/compiler/python_generator.cc b/src/compiler/python_generator.cc index fe2b9fad99b..83133f2b6ea 100644 --- a/src/compiler/python_generator.cc +++ b/src/compiler/python_generator.cc @@ -339,7 +339,7 @@ bool PrintAlphaServerFactory(const grpc::string& package_qualified_service_name, } out->Print("}\n"); out->Print( - "return implementations.server(" + "return early_adopter_implementations.server(" "\"$PackageQualifiedServiceName$\"," " method_service_descriptions, port, private_key=private_key," " certificate_chain=certificate_chain)\n", @@ -422,7 +422,7 @@ bool PrintAlphaStubFactory(const grpc::string& package_qualified_service_name, } out->Print("}\n"); out->Print( - "return implementations.stub(" + "return early_adopter_implementations.stub(" "\"$PackageQualifiedServiceName$\"," " method_invocation_descriptions, host, port," " metadata_transformer=metadata_transformer, secure=secure," @@ -586,13 +586,13 @@ bool PrintBetaServerFactory(const grpc::string& package_qualified_service_name, "Constructor", name_and_implementation_constructor->second); } out->Print("}\n"); - out->Print("server_options = beta.server_options(" + out->Print("server_options = beta_implementations.server_options(" "request_deserializers=request_deserializers, " "response_serializers=response_serializers, " "thread_pool=pool, thread_pool_size=pool_size, " "default_timeout=default_timeout, " "maximum_timeout=maximum_timeout)\n"); - out->Print("return beta.server(method_implementations, " + out->Print("return beta_implementations.server(method_implementations, " "options=server_options)\n"); } return true; @@ -685,13 +685,13 @@ bool PrintBetaStubFactory(const grpc::string& package_qualified_service_name, "Cardinality", name_and_cardinality->second); } out->Print("}\n"); - out->Print("stub_options = beta.stub_options(" + out->Print("stub_options = beta_implementations.stub_options(" "host=host, metadata_transformer=metadata_transformer, " "request_serializers=request_serializers, " "response_deserializers=response_deserializers, " "thread_pool=pool, thread_pool_size=pool_size)\n"); out->Print( - "return beta.dynamic_stub(channel, \'$PackageQualifiedServiceName$\', " + "return beta_implementations.dynamic_stub(channel, \'$PackageQualifiedServiceName$\', " "cardinalities, options=stub_options)\n", "PackageQualifiedServiceName", package_qualified_service_name); } @@ -701,9 +701,9 @@ bool PrintBetaStubFactory(const grpc::string& package_qualified_service_name, bool PrintPreamble(const FileDescriptor* file, const GeneratorConfiguration& config, Printer* out) { out->Print("import abc\n"); - out->Print("from $Package$ import beta\n", + out->Print("from $Package$ import implementations as beta_implementations\n", "Package", config.beta_package_root); - out->Print("from $Package$ import implementations\n", + out->Print("from $Package$ import implementations as early_adopter_implementations\n", "Package", config.early_adopter_package_root); out->Print("from grpc.framework.alpha import utilities as alpha_utilities\n"); out->Print("from grpc.framework.common import cardinality\n"); diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c index 2b61d33c29d..ec832a03672 100644 --- a/src/core/channel/http_client_filter.c +++ b/src/core/channel/http_client_filter.c @@ -85,16 +85,14 @@ static grpc_mdelem *client_filter(void *user_data, grpc_mdelem *md) { static void hc_on_recv(void *user_data, int success) { grpc_call_element *elem = user_data; call_data *calld = elem->call_data; - if (success) { - size_t i; - size_t nops = calld->recv_ops->nops; - grpc_stream_op *ops = calld->recv_ops->ops; - for (i = 0; i < nops; i++) { - grpc_stream_op *op = &ops[i]; - if (op->type != GRPC_OP_METADATA) continue; - calld->got_initial_metadata = 1; - grpc_metadata_batch_filter(&op->data.metadata, client_filter, elem); - } + size_t i; + size_t nops = calld->recv_ops->nops; + grpc_stream_op *ops = calld->recv_ops->ops; + for (i = 0; i < nops; i++) { + grpc_stream_op *op = &ops[i]; + if (op->type != GRPC_OP_METADATA) continue; + calld->got_initial_metadata = 1; + grpc_metadata_batch_filter(&op->data.metadata, client_filter, elem); } calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success); } diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c index fe66ebed779..8f62ce2954b 100644 --- a/src/core/iomgr/pollset_multipoller_with_epoll.c +++ b/src/core/iomgr/pollset_multipoller_with_epoll.c @@ -99,7 +99,6 @@ static void perform_delayed_add(void *arg, int iomgr_status) { if (da->pollset->shutting_down) { /* We don't care about this pollset anymore. */ if (da->pollset->in_flight_cbs == 0 && !da->pollset->called_shutdown) { - GPR_ASSERT(!grpc_pollset_has_workers(da->pollset)); da->pollset->called_shutdown = 1; do_shutdown_cb = 1; } diff --git a/src/core/iomgr/socket_windows.c b/src/core/iomgr/socket_windows.c index 2cbe945ca3e..557ca82226d 100644 --- a/src/core/iomgr/socket_windows.c +++ b/src/core/iomgr/socket_windows.c @@ -35,8 +35,12 @@ #ifdef GPR_WINSOCK_SOCKET +#include +#include + #include #include +#include #include #include "src/core/iomgr/iocp_windows.h" @@ -63,6 +67,24 @@ grpc_winsocket *grpc_winsocket_create(SOCKET socket, const char *name) { various callsites of that function, which happens to be in various mutex hold states, and that'd be unsafe to call them directly. */ void grpc_winsocket_shutdown(grpc_winsocket *winsocket) { + /* Grab the function pointer for DisconnectEx for that specific socket. + It may change depending on the interface. */ + int status; + GUID guid = WSAID_DISCONNECTEX; + LPFN_DISCONNECTEX DisconnectEx; + DWORD ioctl_num_bytes; + + status = WSAIoctl(winsocket->socket, SIO_GET_EXTENSION_FUNCTION_POINTER, + &guid, sizeof(guid), &DisconnectEx, sizeof(DisconnectEx), + &ioctl_num_bytes, NULL, NULL); + + if (status == 0) { + DisconnectEx(winsocket->socket, NULL, 0, 0); + } else { + char *utf8_message = gpr_format_message(WSAGetLastError()); + gpr_log(GPR_ERROR, "Unable to retrieve DisconnectEx pointer : %s", utf8_message); + gpr_free(utf8_message); + } closesocket(winsocket->socket); } diff --git a/src/cpp/server/secure_server_credentials.cc b/src/cpp/server/secure_server_credentials.cc index dfa9229c981..90afebfd2e2 100644 --- a/src/cpp/server/secure_server_credentials.cc +++ b/src/cpp/server/secure_server_credentials.cc @@ -101,7 +101,9 @@ void AuthMetadataProcessorAyncWrapper::InvokeProcessor( 0, {{nullptr, nullptr, nullptr, nullptr}}}); } - cb(user_data, &consumed_md[0], consumed_md.size(), &response_md[0], + auto consumed_md_data = consumed_md.empty() ? nullptr : &consumed_md[0]; + auto response_md_data = response_md.empty() ? nullptr : &response_md[0]; + cb(user_data, consumed_md_data, consumed_md.size(), response_md_data, response_md.size(), static_cast(status.error_code()), status.error_message().c_str()); } @@ -130,8 +132,8 @@ std::shared_ptr SslServerCredentials( } grpc_server_credentials* c_creds = grpc_ssl_server_credentials_create( options.pem_root_certs.empty() ? nullptr : options.pem_root_certs.c_str(), - &pem_key_cert_pairs[0], pem_key_cert_pairs.size(), - options.force_client_auth, nullptr); + pem_key_cert_pairs.empty() ? nullptr : &pem_key_cert_pairs[0], + pem_key_cert_pairs.size(), options.force_client_auth, nullptr); return std::shared_ptr( new SecureServerCredentials(c_creds)); } diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc index b739cbfe62b..99bc8147a0d 100644 --- a/src/cpp/server/server_builder.cc +++ b/src/cpp/server/server_builder.cc @@ -128,7 +128,8 @@ std::unique_ptr ServerBuilder::BuildAndStart() { *port->selected_port = r; } } - if (!server->Start(&cqs_[0], cqs_.size())) { + auto cqs_data = cqs_.empty() ? nullptr : &cqs_[0]; + if (!server->Start(cqs_data, cqs_.size())) { return nullptr; } return server; diff --git a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj index a5945be922b..a0bcf431f7b 100644 --- a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj +++ b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj @@ -42,6 +42,9 @@ False ..\packages\BouncyCastle.1.7.0\lib\Net40-Client\BouncyCastle.Crypto.dll + + ..\packages\CommandLineParser.1.9.71\lib\net45\CommandLine.dll + False ..\packages\Google.Apis.Auth.1.9.3\lib\net40\Google.Apis.Auth.dll diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs index 8343e54122c..0884c6ea60a 100644 --- a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs +++ b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs @@ -37,6 +37,7 @@ using System.Text.RegularExpressions; using System.Threading; using System.Threading.Tasks; +using CommandLine; using Google.Apis.Auth.OAuth2; using Google.Protobuf; using Grpc.Auth; @@ -44,25 +45,54 @@ using Grpc.Core; using Grpc.Core.Utils; using Grpc.Testing; using NUnit.Framework; +using CommandLine.Text; +using System.IO; namespace Grpc.IntegrationTesting { public class InteropClient { - private const string ServiceAccountUser = "155450119199-3psnrh1sdr3d8cpj1v46naggf81mhdnk@developer.gserviceaccount.com"; - private const string ComputeEngineUser = "155450119199-r5aaqa2vqoa9g5mv2m6s3m1l293rlmel@developer.gserviceaccount.com"; - private const string AuthScope = "https://www.googleapis.com/auth/xapi.zoo"; - private const string AuthScopeResponse = "xapi.zoo"; - private class ClientOptions { - public bool help; - public string serverHost = "127.0.0.1"; - public string serverHostOverride = TestCredentials.DefaultHostOverride; - public int? serverPort; - public string testCase = "large_unary"; - public bool useTls; - public bool useTestCa; + [Option("server_host", DefaultValue = "127.0.0.1")] + public string ServerHost { get; set; } + + [Option("server_host_override", DefaultValue = TestCredentials.DefaultHostOverride)] + public string ServerHostOverride { get; set; } + + [Option("server_port", Required = true)] + public int ServerPort { get; set; } + + [Option("test_case", DefaultValue = "large_unary")] + public string TestCase { get; set; } + + [Option("use_tls")] + public bool UseTls { get; set; } + + [Option("use_test_ca")] + public bool UseTestCa { get; set; } + + [Option("default_service_account", Required = false)] + public string DefaultServiceAccount { get; set; } + + [Option("oauth_scope", Required = false)] + public string OAuthScope { get; set; } + + [Option("service_account_key_file", Required = false)] + public string ServiceAccountKeyFile { get; set; } + + [HelpOption] + public string GetUsage() + { + var help = new HelpText + { + Heading = "gRPC C# interop testing client", + AddDashesToOption = true + }; + help.AddPreOptionsLine("Usage:"); + help.AddOptions(this); + return help; + } } ClientOptions options; @@ -74,26 +104,9 @@ namespace Grpc.IntegrationTesting public static void Run(string[] args) { - Console.WriteLine("gRPC C# interop testing client"); - ClientOptions options = ParseArguments(args); - - if (options.serverHost == null || !options.serverPort.HasValue || options.testCase == null) - { - Console.WriteLine("Missing required argument."); - Console.WriteLine(); - options.help = true; - } - - if (options.help) + var options = new ClientOptions(); + if (!Parser.Default.ParseArguments(args, options)) { - Console.WriteLine("Usage:"); - Console.WriteLine(" --server_host=HOSTNAME"); - Console.WriteLine(" --server_host_override=HOSTNAME"); - Console.WriteLine(" --server_port=PORT"); - Console.WriteLine(" --test_case=TESTCASE"); - Console.WriteLine(" --use_tls=BOOLEAN"); - Console.WriteLine(" --use_test_ca=BOOLEAN"); - Console.WriteLine(); Environment.Exit(1); } @@ -103,30 +116,27 @@ namespace Grpc.IntegrationTesting private async Task Run() { - Credentials credentials = null; - if (options.useTls) - { - credentials = TestCredentials.CreateTestClientCredentials(options.useTestCa); - } - + var credentials = options.UseTls ? TestCredentials.CreateTestClientCredentials(options.UseTestCa) : Credentials.Insecure; + List channelOptions = null; - if (!string.IsNullOrEmpty(options.serverHostOverride)) + if (!string.IsNullOrEmpty(options.ServerHostOverride)) { channelOptions = new List { - new ChannelOption(ChannelOptions.SslTargetNameOverride, options.serverHostOverride) + new ChannelOption(ChannelOptions.SslTargetNameOverride, options.ServerHostOverride) }; } - - var channel = new Channel(options.serverHost, options.serverPort.Value, credentials, channelOptions); + Console.WriteLine(options.ServerHost); + Console.WriteLine(options.ServerPort); + var channel = new Channel(options.ServerHost, options.ServerPort, credentials, channelOptions); TestService.TestServiceClient client = new TestService.TestServiceClient(channel); - await RunTestCaseAsync(options.testCase, client); + await RunTestCaseAsync(client, options); channel.ShutdownAsync().Wait(); } - private async Task RunTestCaseAsync(string testCase, TestService.TestServiceClient client) + private async Task RunTestCaseAsync(TestService.TestServiceClient client, ClientOptions options) { - switch (testCase) + switch (options.TestCase) { case "empty_unary": RunEmptyUnary(client); @@ -146,20 +156,17 @@ namespace Grpc.IntegrationTesting case "empty_stream": await RunEmptyStreamAsync(client); break; - case "service_account_creds": - await RunServiceAccountCredsAsync(client); - break; case "compute_engine_creds": - await RunComputeEngineCredsAsync(client); + await RunComputeEngineCredsAsync(client, options.DefaultServiceAccount, options.OAuthScope); break; case "jwt_token_creds": - await RunJwtTokenCredsAsync(client); + await RunJwtTokenCredsAsync(client, options.DefaultServiceAccount); break; case "oauth2_auth_token": - await RunOAuth2AuthTokenAsync(client); + await RunOAuth2AuthTokenAsync(client, options.DefaultServiceAccount, options.OAuthScope); break; case "per_rpc_creds": - await RunPerRpcCredsAsync(client); + await RunPerRpcCredsAsync(client, options.DefaultServiceAccount, options.OAuthScope); break; case "cancel_after_begin": await RunCancelAfterBeginAsync(client); @@ -174,7 +181,7 @@ namespace Grpc.IntegrationTesting RunBenchmarkEmptyUnary(client); break; default: - throw new ArgumentException("Unknown test case " + testCase); + throw new ArgumentException("Unknown test case " + options.TestCase); } } @@ -313,32 +320,7 @@ namespace Grpc.IntegrationTesting Console.WriteLine("Passed!"); } - public static async Task RunServiceAccountCredsAsync(TestService.TestServiceClient client) - { - Console.WriteLine("running service_account_creds"); - var credential = await GoogleCredential.GetApplicationDefaultAsync(); - credential = credential.CreateScoped(new[] { AuthScope }); - client.HeaderInterceptor = AuthInterceptors.FromCredential(credential); - - var request = new SimpleRequest - { - ResponseType = PayloadType.COMPRESSABLE, - ResponseSize = 314159, - Payload = CreateZerosPayload(271828), - FillUsername = true, - FillOauthScope = true - }; - - var response = client.UnaryCall(request); - - Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); - Assert.AreEqual(314159, response.Payload.Body.Length); - Assert.AreEqual(AuthScopeResponse, response.OauthScope); - Assert.AreEqual(ServiceAccountUser, response.Username); - Console.WriteLine("Passed!"); - } - - public static async Task RunComputeEngineCredsAsync(TestService.TestServiceClient client) + public static async Task RunComputeEngineCredsAsync(TestService.TestServiceClient client, string defaultServiceAccount, string oauthScope) { Console.WriteLine("running compute_engine_creds"); var credential = await GoogleCredential.GetApplicationDefaultAsync(); @@ -358,16 +340,16 @@ namespace Grpc.IntegrationTesting Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); Assert.AreEqual(314159, response.Payload.Body.Length); - Assert.AreEqual(AuthScopeResponse, response.OauthScope); - Assert.AreEqual(ComputeEngineUser, response.Username); + Assert.False(string.IsNullOrEmpty(response.OauthScope)); + Assert.True(oauthScope.Contains(response.OauthScope)); + Assert.AreEqual(defaultServiceAccount, response.Username); Console.WriteLine("Passed!"); } - public static async Task RunJwtTokenCredsAsync(TestService.TestServiceClient client) + public static async Task RunJwtTokenCredsAsync(TestService.TestServiceClient client, string defaultServiceAccount) { Console.WriteLine("running jwt_token_creds"); var credential = await GoogleCredential.GetApplicationDefaultAsync(); - // check this a credential with scope support, but don't add the scope. Assert.IsTrue(credential.IsCreateScopedRequired); client.HeaderInterceptor = AuthInterceptors.FromCredential(credential); @@ -377,21 +359,20 @@ namespace Grpc.IntegrationTesting ResponseSize = 314159, Payload = CreateZerosPayload(271828), FillUsername = true, - FillOauthScope = true }; var response = client.UnaryCall(request); Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type); Assert.AreEqual(314159, response.Payload.Body.Length); - Assert.AreEqual(ServiceAccountUser, response.Username); + Assert.AreEqual(defaultServiceAccount, response.Username); Console.WriteLine("Passed!"); } - public static async Task RunOAuth2AuthTokenAsync(TestService.TestServiceClient client) + public static async Task RunOAuth2AuthTokenAsync(TestService.TestServiceClient client, string defaultServiceAccount, string oauthScope) { Console.WriteLine("running oauth2_auth_token"); - ITokenAccess credential = (await GoogleCredential.GetApplicationDefaultAsync()).CreateScoped(new[] { AuthScope }); + ITokenAccess credential = (await GoogleCredential.GetApplicationDefaultAsync()).CreateScoped(new[] { oauthScope }); string oauth2Token = await credential.GetAccessTokenForRequestAsync(); client.HeaderInterceptor = AuthInterceptors.FromAccessToken(oauth2Token); @@ -404,31 +385,29 @@ namespace Grpc.IntegrationTesting var response = client.UnaryCall(request); - Assert.AreEqual(AuthScopeResponse, response.OauthScope); - Assert.AreEqual(ServiceAccountUser, response.Username); + Assert.False(string.IsNullOrEmpty(response.OauthScope)); + Assert.True(oauthScope.Contains(response.OauthScope)); + Assert.AreEqual(defaultServiceAccount, response.Username); Console.WriteLine("Passed!"); } - public static async Task RunPerRpcCredsAsync(TestService.TestServiceClient client) + public static async Task RunPerRpcCredsAsync(TestService.TestServiceClient client, string defaultServiceAccount, string oauthScope) { Console.WriteLine("running per_rpc_creds"); - - ITokenAccess credential = (await GoogleCredential.GetApplicationDefaultAsync()).CreateScoped(new[] { AuthScope }); - string oauth2Token = await credential.GetAccessTokenForRequestAsync(); - var headerInterceptor = AuthInterceptors.FromAccessToken(oauth2Token); + ITokenAccess credential = (await GoogleCredential.GetApplicationDefaultAsync()).CreateScoped(new[] { oauthScope }); + string accessToken = await credential.GetAccessTokenForRequestAsync(); + var headerInterceptor = AuthInterceptors.FromAccessToken(accessToken); var request = new SimpleRequest { FillUsername = true, - FillOauthScope = true }; var headers = new Metadata(); headerInterceptor(null, "", headers); var response = client.UnaryCall(request, headers: headers); - Assert.AreEqual(AuthScopeResponse, response.OauthScope); - Assert.AreEqual(ServiceAccountUser, response.Username); + Assert.AreEqual(defaultServiceAccount, response.Username); Console.WriteLine("Passed!"); } @@ -508,68 +487,5 @@ namespace Grpc.IntegrationTesting { return new Payload { Body = ByteString.CopyFrom(new byte[size]) }; } - - private static ClientOptions ParseArguments(string[] args) - { - var options = new ClientOptions(); - foreach (string arg in args) - { - ParseArgument(arg, options); - if (options.help) - { - break; - } - } - return options; - } - - private static void ParseArgument(string arg, ClientOptions options) - { - Match match; - match = Regex.Match(arg, "--server_host=(.*)"); - if (match.Success) - { - options.serverHost = match.Groups[1].Value.Trim(); - return; - } - - match = Regex.Match(arg, "--server_host_override=(.*)"); - if (match.Success) - { - options.serverHostOverride = match.Groups[1].Value.Trim(); - return; - } - - match = Regex.Match(arg, "--server_port=(.*)"); - if (match.Success) - { - options.serverPort = int.Parse(match.Groups[1].Value.Trim()); - return; - } - - match = Regex.Match(arg, "--test_case=(.*)"); - if (match.Success) - { - options.testCase = match.Groups[1].Value.Trim(); - return; - } - - match = Regex.Match(arg, "--use_tls=(.*)"); - if (match.Success) - { - options.useTls = bool.Parse(match.Groups[1].Value.Trim()); - return; - } - - match = Regex.Match(arg, "--use_test_ca=(.*)"); - if (match.Success) - { - options.useTestCa = bool.Parse(match.Groups[1].Value.Trim()); - return; - } - - Console.WriteLine(string.Format("Unrecognized argument \"{0}\"", arg)); - options.help = true; - } } } diff --git a/src/csharp/Grpc.IntegrationTesting/InteropServer.cs b/src/csharp/Grpc.IntegrationTesting/InteropServer.cs index 718278f30a6..513f8722d64 100644 --- a/src/csharp/Grpc.IntegrationTesting/InteropServer.cs +++ b/src/csharp/Grpc.IntegrationTesting/InteropServer.cs @@ -37,6 +37,9 @@ using System.Diagnostics; using System.IO; using System.Text.RegularExpressions; using System.Threading.Tasks; + +using CommandLine; +using CommandLine.Text; using Grpc.Core; using Grpc.Core.Utils; using Grpc.Testing; @@ -48,9 +51,24 @@ namespace Grpc.IntegrationTesting { private class ServerOptions { - public bool help; - public int? port = 8070; - public bool useTls; + [Option("port", DefaultValue = 8070)] + public int Port { get; set; } + + [Option("use_tls")] + public bool UseTls { get; set; } + + [HelpOption] + public string GetUsage() + { + var help = new HelpText + { + Heading = "gRPC C# interop testing server", + AddDashesToOption = true + }; + help.AddPreOptionsLine("Usage:"); + help.AddOptions(this); + return help; + } } ServerOptions options; @@ -62,22 +80,9 @@ namespace Grpc.IntegrationTesting public static void Run(string[] args) { - Console.WriteLine("gRPC C# interop testing server"); - ServerOptions options = ParseArguments(args); - - if (!options.port.HasValue) - { - Console.WriteLine("Missing required argument."); - Console.WriteLine(); - options.help = true; - } - - if (options.help) + var options = new ServerOptions(); + if (!Parser.Default.ParseArguments(args, options)) { - Console.WriteLine("Usage:"); - Console.WriteLine(" --port=PORT"); - Console.WriteLine(" --use_tls=BOOLEAN"); - Console.WriteLine(); Environment.Exit(1); } @@ -93,54 +98,19 @@ namespace Grpc.IntegrationTesting }; string host = "0.0.0.0"; - int port = options.port.Value; - if (options.useTls) + int port = options.Port; + if (options.UseTls) { server.Ports.Add(host, port, TestCredentials.CreateTestServerCredentials()); } else { - server.Ports.Add(host, options.port.Value, ServerCredentials.Insecure); + server.Ports.Add(host, options.Port, ServerCredentials.Insecure); } Console.WriteLine("Running server on " + string.Format("{0}:{1}", host, port)); server.Start(); server.ShutdownTask.Wait(); } - - private static ServerOptions ParseArguments(string[] args) - { - var options = new ServerOptions(); - foreach (string arg in args) - { - ParseArgument(arg, options); - if (options.help) - { - break; - } - } - return options; - } - - private static void ParseArgument(string arg, ServerOptions options) - { - Match match; - match = Regex.Match(arg, "--port=(.*)"); - if (match.Success) - { - options.port = int.Parse(match.Groups[1].Value.Trim()); - return; - } - - match = Regex.Match(arg, "--use_tls=(.*)"); - if (match.Success) - { - options.useTls = bool.Parse(match.Groups[1].Value.Trim()); - return; - } - - Console.WriteLine(string.Format("Unrecognized argument \"{0}\"", arg)); - options.help = true; - } } } diff --git a/src/csharp/Grpc.IntegrationTesting/packages.config b/src/csharp/Grpc.IntegrationTesting/packages.config index 8dfded1964f..bdb3dadf445 100644 --- a/src/csharp/Grpc.IntegrationTesting/packages.config +++ b/src/csharp/Grpc.IntegrationTesting/packages.config @@ -1,6 +1,7 @@  + diff --git a/src/node/README.md b/src/node/README.md index 0b97680feb9..7719d082900 100644 --- a/src/node/README.md +++ b/src/node/README.md @@ -11,10 +11,10 @@ Beta **Linux (Debian):** -Add [Debian testing][] to your `sources.list` file. Example: +Add [Debian jessie-backports][] to your `sources.list` file. Example: ```sh -echo "deb http://ftp.us.debian.org/debian testing main contrib non-free" | \ +echo "deb http://http.debian.net/debian jessie-backports main" | \ sudo tee -a /etc/apt/sources.list ``` @@ -113,4 +113,4 @@ An object with factory methods for creating credential objects for servers. [homebrew]:http://brew.sh [gRPC install script]:https://raw.githubusercontent.com/grpc/homebrew-grpc/master/scripts/install -[Debian testing]:https://www.debian.org/releases/stretch/ +[Debian jessie-backports]:http://backports.debian.org/Instructions/ diff --git a/src/php/README.md b/src/php/README.md index afa09d79a1b..51322c75269 100644 --- a/src/php/README.md +++ b/src/php/README.md @@ -32,10 +32,10 @@ $ sudo php -d detect_unicode=0 go-pear.phar **Linux (Debian):** -Add [Debian testing][] to your `sources.list` file. Example: +Add [Debian jessie-backports][] to your `sources.list` file. Example: ```sh -echo "deb http://ftp.us.debian.org/debian testing main contrib non-free" | \ +echo "deb http://http.debian.net/debian jessie-backports main" | \ sudo tee -a /etc/apt/sources.list ``` @@ -167,4 +167,4 @@ $ ./bin/run_gen_code_test.sh [homebrew]:http://brew.sh [gRPC install script]:https://raw.githubusercontent.com/grpc/homebrew-grpc/master/scripts/install [Node]:https://github.com/grpc/grpc/tree/master/src/node/examples -[Debian testing]:https://www.debian.org/releases/stretch/ +[Debian jessie-backports]:http://backports.debian.org/Instructions/ diff --git a/src/python/README.md b/src/python/README.md index 67d1a173a27..afe7c731f1e 100644 --- a/src/python/README.md +++ b/src/python/README.md @@ -16,10 +16,10 @@ INSTALLATION **Linux (Debian):** -Add [Debian testing][] to your `sources.list` file. Example: +Add [Debian jessie-backports][] to your `sources.list` file. Example: ```sh -echo "deb http://ftp.us.debian.org/debian testing main contrib non-free" | \ +echo "deb http://http.debian.net/debian jessie-backports main" | \ sudo tee -a /etc/apt/sources.list ``` @@ -92,4 +92,4 @@ $ ../../tools/distrib/python/submit.py [gRPC install script]:https://raw.githubusercontent.com/grpc/homebrew-grpc/master/scripts/install [Quick Start]:http://www.grpc.io/docs/tutorials/basic/python.html [detailed example]:http://www.grpc.io/docs/installation/python.html -[Debian testing]:https://www.debian.org/releases/stretch/ +[Debian jessie-backports]:http://backports.debian.org/Instructions/ diff --git a/src/python/grpcio/grpc/_adapter/_intermediary_low.py b/src/python/grpcio/grpc/_adapter/_intermediary_low.py index 735ad205a4b..e2feec6ffba 100644 --- a/src/python/grpcio/grpc/_adapter/_intermediary_low.py +++ b/src/python/grpcio/grpc/_adapter/_intermediary_low.py @@ -59,6 +59,7 @@ from grpc._adapter import _types _IGNORE_ME_TAG = object() Code = _types.StatusCode +WriteFlags = _types.OpWriteFlags class Status(collections.namedtuple('Status', ['code', 'details'])): @@ -125,9 +126,9 @@ class Call(object): ], _TagAdapter(finish_tag, Event.Kind.FINISH)) return err0 if err0 != _types.CallError.OK else err1 if err1 != _types.CallError.OK else err2 if err2 != _types.CallError.OK else _types.CallError.OK - def write(self, message, tag): + def write(self, message, tag, flags): return self._internal.start_batch([ - _types.OpArgs.send_message(message, 0) + _types.OpArgs.send_message(message, flags) ], _TagAdapter(tag, Event.Kind.WRITE_ACCEPTED)) def complete(self, tag): @@ -163,8 +164,11 @@ class Call(object): def cancel(self): return self._internal.cancel() + def peer(self): + return self._internal.peer() + def set_credentials(self, creds): - return self._internal.set_credentials(creds) + return self._internal.set_credentials(creds._internal) class Channel(object): diff --git a/src/python/grpcio/grpc/_adapter/fore.py b/src/python/grpcio/grpc/_adapter/fore.py index daa41e8bde5..acdd69c4206 100644 --- a/src/python/grpcio/grpc/_adapter/fore.py +++ b/src/python/grpcio/grpc/_adapter/fore.py @@ -56,7 +56,7 @@ class _LowWrite(enum.Enum): def _write(call, rpc_state, payload): serialized_payload = rpc_state.serializer(payload) if rpc_state.write.low is _LowWrite.OPEN: - call.write(serialized_payload, call) + call.write(serialized_payload, call, 0) rpc_state.write.low = _LowWrite.ACTIVE else: rpc_state.write.pending.append(serialized_payload) @@ -164,7 +164,7 @@ class ForeLink(base_interfaces.ForeLink, activated.Activated): if rpc_state.write.pending: serialized_payload = rpc_state.write.pending.pop(0) - call.write(serialized_payload, call) + call.write(serialized_payload, call, 0) elif rpc_state.write.high is _common.HighWrite.CLOSED: _status(call, rpc_state) else: diff --git a/src/python/grpcio/grpc/_adapter/rear.py b/src/python/grpcio/grpc/_adapter/rear.py index fd6f45f7a79..17fa47f7460 100644 --- a/src/python/grpcio/grpc/_adapter/rear.py +++ b/src/python/grpcio/grpc/_adapter/rear.py @@ -78,7 +78,7 @@ class _RPCState(object): def _write(operation_id, call, outstanding, write_state, serialized_payload): if write_state.low is _LowWrite.OPEN: - call.write(serialized_payload, operation_id) + call.write(serialized_payload, operation_id, 0) outstanding.add(_low.Event.Kind.WRITE_ACCEPTED) write_state.low = _LowWrite.ACTIVE elif write_state.low is _LowWrite.ACTIVE: @@ -144,7 +144,7 @@ class RearLink(base_interfaces.RearLink, activated.Activated): if event.write_accepted: if rpc_state.common.write.pending: rpc_state.call.write( - rpc_state.common.write.pending.pop(0), operation_id) + rpc_state.common.write.pending.pop(0), operation_id, 0) rpc_state.outstanding.add(_low.Event.Kind.WRITE_ACCEPTED) elif rpc_state.common.write.high is _common.HighWrite.CLOSED: rpc_state.call.complete(operation_id) @@ -263,7 +263,7 @@ class RearLink(base_interfaces.RearLink, activated.Activated): low_state = _LowWrite.OPEN else: serialized_payload = request_serializer(payload) - call.write(serialized_payload, operation_id) + call.write(serialized_payload, operation_id, 0) outstanding.add(_low.Event.Kind.WRITE_ACCEPTED) low_state = _LowWrite.ACTIVE diff --git a/src/python/grpcio/grpc/_links/invocation.py b/src/python/grpcio/grpc/_links/invocation.py index fecb550ae05..67ef86a1769 100644 --- a/src/python/grpcio/grpc/_links/invocation.py +++ b/src/python/grpcio/grpc/_links/invocation.py @@ -37,6 +37,7 @@ import time from grpc._adapter import _intermediary_low from grpc._links import _constants +from grpc.beta import interfaces as beta_interfaces from grpc.framework.foundation import activated from grpc.framework.foundation import logging_pool from grpc.framework.foundation import relay @@ -73,11 +74,28 @@ class _LowWrite(enum.Enum): CLOSED = 'CLOSED' +class _Context(beta_interfaces.GRPCInvocationContext): + + def __init__(self): + self._lock = threading.Lock() + self._disable_next_compression = False + + def disable_next_request_compression(self): + with self._lock: + self._disable_next_compression = True + + def next_compression_disabled(self): + with self._lock: + disabled = self._disable_next_compression + self._disable_next_compression = False + return disabled + + class _RPCState(object): def __init__( self, call, request_serializer, response_deserializer, sequence_number, - read, allowance, high_write, low_write, due): + read, allowance, high_write, low_write, due, context): self.call = call self.request_serializer = request_serializer self.response_deserializer = response_deserializer @@ -87,6 +105,7 @@ class _RPCState(object): self.high_write = high_write self.low_write = low_write self.due = due + self.context = context def _no_longer_due(kind, rpc_state, key, rpc_states): @@ -209,7 +228,7 @@ class _Kernel(object): def _invoke( self, operation_id, group, method, initial_metadata, payload, termination, - timeout, allowance): + timeout, allowance, options): """Invoke an RPC. Args: @@ -224,6 +243,7 @@ class _Kernel(object): timeout: A duration of time in seconds to allow for the RPC. allowance: The number of payloads (beyond the free first one) that the local ticket exchange mate has granted permission to be read. + options: A beta_interfaces.GRPCCallOptions value or None. """ if termination is links.Ticket.Termination.COMPLETION: high_write = _HighWrite.CLOSED @@ -241,6 +261,8 @@ class _Kernel(object): call = _intermediary_low.Call( self._channel, self._completion_queue, '/%s/%s' % (group, method), self._host, time.time() + timeout) + if options is not None and options.credentials is not None: + call.set_credentials(options.credentials._intermediary_low_credentials) if transformed_initial_metadata is not None: for metadata_key, metadata_value in transformed_initial_metadata: call.add_metadata(metadata_key, metadata_value) @@ -254,17 +276,33 @@ class _Kernel(object): low_write = _LowWrite.OPEN due = set((_METADATA, _FINISH,)) else: - call.write(request_serializer(payload), operation_id) + if options is not None and options.disable_compression: + flags = _intermediary_low.WriteFlags.WRITE_NO_COMPRESS + else: + flags = 0 + call.write(request_serializer(payload), operation_id, flags) low_write = _LowWrite.ACTIVE due = set((_WRITE, _METADATA, _FINISH,)) + context = _Context() self._rpc_states[operation_id] = _RPCState( - call, request_serializer, response_deserializer, 0, + call, request_serializer, response_deserializer, 1, _Read.AWAITING_METADATA, 1 if allowance is None else (1 + allowance), - high_write, low_write, due) + high_write, low_write, due, context) + protocol = links.Protocol(links.Protocol.Kind.INVOCATION_CONTEXT, context) + ticket = links.Ticket( + operation_id, 0, None, None, None, None, None, None, None, None, None, + None, None, protocol) + self._relay.add_value(ticket) def _advance(self, operation_id, rpc_state, payload, termination, allowance): if payload is not None: - rpc_state.call.write(rpc_state.request_serializer(payload), operation_id) + disable_compression = rpc_state.context.next_compression_disabled() + if disable_compression: + flags = _intermediary_low.WriteFlags.WRITE_NO_COMPRESS + else: + flags = 0 + rpc_state.call.write( + rpc_state.request_serializer(payload), operation_id, flags) rpc_state.low_write = _LowWrite.ACTIVE rpc_state.due.add(_WRITE) @@ -292,10 +330,15 @@ class _Kernel(object): if self._completion_queue is None: logging.error('Received invocation ticket %s after stop!', ticket) else: + if (ticket.protocol is not None and + ticket.protocol.kind is links.Protocol.Kind.CALL_OPTION): + grpc_call_options = ticket.protocol.value + else: + grpc_call_options = None self._invoke( ticket.operation_id, ticket.group, ticket.method, ticket.initial_metadata, ticket.payload, ticket.termination, - ticket.timeout, ticket.allowance) + ticket.timeout, ticket.allowance, grpc_call_options) else: rpc_state = self._rpc_states.get(ticket.operation_id) if rpc_state is not None: diff --git a/src/python/grpcio/grpc/_links/service.py b/src/python/grpcio/grpc/_links/service.py index 34d3b262c98..f56df84007d 100644 --- a/src/python/grpcio/grpc/_links/service.py +++ b/src/python/grpcio/grpc/_links/service.py @@ -37,6 +37,7 @@ import time from grpc._adapter import _intermediary_low from grpc._links import _constants +from grpc.beta import interfaces as beta_interfaces from grpc.framework.foundation import logging_pool from grpc.framework.foundation import relay from grpc.framework.interfaces.links import links @@ -89,12 +90,34 @@ class _LowWrite(enum.Enum): CLOSED = 'CLOSED' +class _Context(beta_interfaces.GRPCServicerContext): + + def __init__(self, call): + self._lock = threading.Lock() + self._call = call + self._disable_next_compression = False + + def peer(self): + with self._lock: + return self._call.peer() + + def disable_next_response_compression(self): + with self._lock: + self._disable_next_compression = True + + def next_compression_disabled(self): + with self._lock: + disabled = self._disable_next_compression + self._disable_next_compression = False + return disabled + + class _RPCState(object): def __init__( self, request_deserializer, response_serializer, sequence_number, read, early_read, allowance, high_write, low_write, premetadataed, - terminal_metadata, code, message, due): + terminal_metadata, code, message, due, context): self.request_deserializer = request_deserializer self.response_serializer = response_serializer self.sequence_number = sequence_number @@ -110,6 +133,7 @@ class _RPCState(object): self.code = code self.message = message self.due = due + self.context = context def _no_longer_due(kind, rpc_state, key, rpc_states): @@ -163,14 +187,16 @@ class _Kernel(object): (group, method), _IDENTITY) call.read(call) + context = _Context(call) self._rpc_states[call] = _RPCState( request_deserializer, response_serializer, 1, _Read.READING, None, 1, _HighWrite.OPEN, _LowWrite.OPEN, False, None, None, None, - set((_READ, _FINISH,))) + set((_READ, _FINISH,)), context) + protocol = links.Protocol(links.Protocol.Kind.SERVICER_CONTEXT, context) ticket = links.Ticket( call, 0, group, method, links.Ticket.Subscription.FULL, service_acceptance.deadline - time.time(), None, event.metadata, None, - None, None, None, None, 'TODO: Service Context Object!') + None, None, None, None, protocol) self._relay.add_value(ticket) def _on_read_event(self, event): @@ -311,7 +337,12 @@ class _Kernel(object): self._relay.add_value(early_read_ticket) if ticket.payload is not None: - call.write(rpc_state.response_serializer(ticket.payload), call) + disable_compression = rpc_state.context.next_compression_disabled() + if disable_compression: + flags = _intermediary_low.WriteFlags.WRITE_NO_COMPRESS + else: + flags = 0 + call.write(rpc_state.response_serializer(ticket.payload), call, flags) rpc_state.due.add(_WRITE) rpc_state.low_write = _LowWrite.ACTIVE diff --git a/src/python/grpcio/grpc/beta/_connectivity_channel.py b/src/python/grpcio/grpc/beta/_connectivity_channel.py index 457ede79f2e..61674a70add 100644 --- a/src/python/grpcio/grpc/beta/_connectivity_channel.py +++ b/src/python/grpcio/grpc/beta/_connectivity_channel.py @@ -33,18 +33,24 @@ import threading import time from grpc._adapter import _low +from grpc._adapter import _types +from grpc.beta import interfaces from grpc.framework.foundation import callable_util _CHANNEL_SUBSCRIPTION_CALLBACK_ERROR_LOG_MESSAGE = ( 'Exception calling channel subscription callback!') +_LOW_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY = { + state: connectivity for state, connectivity in zip( + _types.ConnectivityState, interfaces.ChannelConnectivity) +} + class ConnectivityChannel(object): - def __init__(self, low_channel, mapping): + def __init__(self, low_channel): self._lock = threading.Lock() self._low_channel = low_channel - self._mapping = mapping self._polling = False self._connectivity = None @@ -88,7 +94,8 @@ class ConnectivityChannel(object): try_to_connect = initial_try_to_connect low_connectivity = low_channel.check_connectivity_state(try_to_connect) with self._lock: - self._connectivity = self._mapping[low_connectivity] + self._connectivity = _LOW_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[ + low_connectivity] callbacks = tuple( callback for callback, unused_but_known_to_be_none_connectivity in self._callbacks_and_connectivities) @@ -112,7 +119,8 @@ class ConnectivityChannel(object): if event.success or try_to_connect: low_connectivity = low_channel.check_connectivity_state(try_to_connect) with self._lock: - self._connectivity = self._mapping[low_connectivity] + self._connectivity = _LOW_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[ + low_connectivity] if not self._delivering: callbacks = self._deliveries(self._connectivity) if callbacks: diff --git a/src/python/grpcio/grpc/beta/_server.py b/src/python/grpcio/grpc/beta/_server.py index ebf91d80ab1..daa42c475ad 100644 --- a/src/python/grpcio/grpc/beta/_server.py +++ b/src/python/grpcio/grpc/beta/_server.py @@ -72,7 +72,7 @@ def _disassemble(grpc_link, end_link, pool, event, grace): event.set() -class Server(object): +class Server(interfaces.Server): def __init__(self, grpc_link, end_link, pool): self._grpc_link = grpc_link @@ -82,9 +82,9 @@ class Server(object): def add_insecure_port(self, address): return self._grpc_link.add_port(address, None) - def add_secure_port(self, address, intermediary_low_server_credentials): + def add_secure_port(self, address, server_credentials): return self._grpc_link.add_port( - address, intermediary_low_server_credentials) + address, server_credentials._intermediary_low_credentials) # pylint: disable=protected-access def start(self): self._grpc_link.join_link(self._end_link) diff --git a/src/python/grpcio/grpc/beta/beta.py b/src/python/grpcio/grpc/beta/implementations.py similarity index 75% rename from src/python/grpcio/grpc/beta/beta.py rename to src/python/grpcio/grpc/beta/implementations.py index b3a161087f3..9b461fb3ddf 100644 --- a/src/python/grpcio/grpc/beta/beta.py +++ b/src/python/grpcio/grpc/beta/implementations.py @@ -40,6 +40,7 @@ from grpc._adapter import _types from grpc.beta import _connectivity_channel from grpc.beta import _server from grpc.beta import _stub +from grpc.beta import interfaces from grpc.framework.common import cardinality # pylint: disable=unused-import from grpc.framework.interfaces.face import face # pylint: disable=unused-import @@ -47,32 +48,6 @@ _CHANNEL_SUBSCRIPTION_CALLBACK_ERROR_LOG_MESSAGE = ( 'Exception calling channel subscription callback!') -@enum.unique -class ChannelConnectivity(enum.Enum): - """Mirrors grpc_connectivity_state in the gRPC Core. - - Attributes: - IDLE: The channel is idle. - CONNECTING: The channel is connecting. - READY: The channel is ready to conduct RPCs. - TRANSIENT_FAILURE: The channel has seen a failure from which it expects to - recover. - FATAL_FAILURE: The channel has seen a failure from which it cannot recover. - """ - - IDLE = (_types.ConnectivityState.IDLE, 'idle',) - CONNECTING = (_types.ConnectivityState.CONNECTING, 'connecting',) - READY = (_types.ConnectivityState.READY, 'ready',) - TRANSIENT_FAILURE = ( - _types.ConnectivityState.TRANSIENT_FAILURE, 'transient failure',) - FATAL_FAILURE = (_types.ConnectivityState.FATAL_FAILURE, 'fatal failure',) - -_LOW_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY = { - state: connectivity for state, connectivity in zip( - _types.ConnectivityState, ChannelConnectivity) -} - - class ClientCredentials(object): """A value encapsulating the data required to create a secure Channel. @@ -118,13 +93,14 @@ class Channel(object): self._low_channel = low_channel self._intermediary_low_channel = intermediary_low_channel self._connectivity_channel = _connectivity_channel.ConnectivityChannel( - low_channel, _LOW_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY) + low_channel) def subscribe(self, callback, try_to_connect=None): """Subscribes to this Channel's connectivity. Args: - callback: A callable to be invoked and passed this Channel's connectivity. + callback: A callable to be invoked and passed an + interfaces.ChannelConnectivity identifying this Channel's connectivity. The callable will be invoked immediately upon subscription and again for every change to this Channel's connectivity thereafter until it is unsubscribed. @@ -144,7 +120,7 @@ class Channel(object): self._connectivity_channel.unsubscribe(callback) -def create_insecure_channel(host, port): +def insecure_channel(host, port): """Creates an insecure Channel to a remote host. Args: @@ -159,7 +135,7 @@ def create_insecure_channel(host, port): return Channel(intermediary_low_channel._internal, intermediary_low_channel) # pylint: disable=protected-access -def create_secure_channel(host, port, client_credentials): +def secure_channel(host, port, client_credentials): """Creates a secure Channel to a remote host. Args: @@ -313,86 +289,6 @@ def ssl_server_credentials( intermediary_low_credentials._internal, intermediary_low_credentials) # pylint: disable=protected-access -class Server(object): - """Services RPCs.""" - __metaclass__ = abc.ABCMeta - - @abc.abstractmethod - def add_insecure_port(self, address): - """Reserves a port for insecure RPC service once this Server becomes active. - - This method may only be called before calling this Server's start method is - called. - - Args: - address: The address for which to open a port. - - Returns: - An integer port on which RPCs will be serviced after this link has been - started. This is typically the same number as the port number contained - in the passed address, but will likely be different if the port number - contained in the passed address was zero. - """ - raise NotImplementedError() - - @abc.abstractmethod - def add_secure_port(self, address, server_credentials): - """Reserves a port for secure RPC service after this Server becomes active. - - This method may only be called before calling this Server's start method is - called. - - Args: - address: The address for which to open a port. - server_credentials: A ServerCredentials. - - Returns: - An integer port on which RPCs will be serviced after this link has been - started. This is typically the same number as the port number contained - in the passed address, but will likely be different if the port number - contained in the passed address was zero. - """ - raise NotImplementedError() - - @abc.abstractmethod - def start(self): - """Starts this Server's service of RPCs. - - This method may only be called while the server is not serving RPCs (i.e. it - is not idempotent). - """ - raise NotImplementedError() - - @abc.abstractmethod - def stop(self, grace): - """Stops this Server's service of RPCs. - - All calls to this method immediately stop service of new RPCs. When existing - RPCs are aborted is controlled by the grace period parameter passed to this - method. - - This method may be called at any time and is idempotent. Passing a smaller - grace value than has been passed in a previous call will have the effect of - stopping the Server sooner. Passing a larger grace value than has been - passed in a previous call will not have the effect of stopping the sooner - later. - - Args: - grace: A duration of time in seconds to allow existing RPCs to complete - before being aborted by this Server's stopping. May be zero for - immediate abortion of all in-progress RPCs. - - Returns: - A threading.Event that will be set when this Server has completely - stopped. The returned event may not be set until after the full grace - period (if some ongoing RPC continues for the full length of the period) - of it may be set much sooner (such as if this Server had no RPCs underway - at the time it was stopped or if all RPCs that it had underway completed - very early in the grace period). - """ - raise NotImplementedError() - - class ServerOptions(object): """A value encapsulating the various options for creation of a Server. @@ -450,27 +346,8 @@ def server_options( thread_pool, thread_pool_size, default_timeout, maximum_timeout) -class _Server(Server): - - def __init__(self, underserver): - self._underserver = underserver - - def add_insecure_port(self, address): - return self._underserver.add_insecure_port(address) - - def add_secure_port(self, address, server_credentials): - return self._underserver.add_secure_port( - address, server_credentials._intermediary_low_credentials) # pylint: disable=protected-access - - def start(self): - self._underserver.start() - - def stop(self, grace): - return self._underserver.stop(grace) - - def server(service_implementations, options=None): - """Creates a Server with which RPCs can be serviced. + """Creates an interfaces.Server with which RPCs can be serviced. Args: service_implementations: A dictionary from service name-method name pair to @@ -479,13 +356,12 @@ def server(service_implementations, options=None): functionality of the returned Server. Returns: - A Server with which RPCs can be serviced. + An interfaces.Server with which RPCs can be serviced. """ effective_options = _EMPTY_SERVER_OPTIONS if options is None else options - underserver = _server.server( + return _server.server( service_implementations, effective_options.multi_method_implementation, effective_options.request_deserializers, effective_options.response_serializers, effective_options.thread_pool, effective_options.thread_pool_size, effective_options.default_timeout, effective_options.maximum_timeout) - return _Server(underserver) diff --git a/src/python/grpcio/grpc/beta/interfaces.py b/src/python/grpcio/grpc/beta/interfaces.py index 25e6a9c66b4..07c8618f70c 100644 --- a/src/python/grpcio/grpc/beta/interfaces.py +++ b/src/python/grpcio/grpc/beta/interfaces.py @@ -29,8 +29,31 @@ """Constants and interfaces of the Beta API of gRPC Python.""" +import abc import enum +from grpc._adapter import _types + + +@enum.unique +class ChannelConnectivity(enum.Enum): + """Mirrors grpc_connectivity_state in the gRPC Core. + + Attributes: + IDLE: The channel is idle. + CONNECTING: The channel is connecting. + READY: The channel is ready to conduct RPCs. + TRANSIENT_FAILURE: The channel has seen a failure from which it expects to + recover. + FATAL_FAILURE: The channel has seen a failure from which it cannot recover. + """ + IDLE = (_types.ConnectivityState.IDLE, 'idle',) + CONNECTING = (_types.ConnectivityState.CONNECTING, 'connecting',) + READY = (_types.ConnectivityState.READY, 'ready',) + TRANSIENT_FAILURE = ( + _types.ConnectivityState.TRANSIENT_FAILURE, 'transient failure',) + FATAL_FAILURE = (_types.ConnectivityState.FATAL_FAILURE, 'fatal failure',) + @enum.unique class StatusCode(enum.Enum): @@ -52,3 +75,140 @@ class StatusCode(enum.Enum): UNAVAILABLE = 14 DATA_LOSS = 15 UNAUTHENTICATED = 16 + + +class GRPCCallOptions(object): + """A value encapsulating gRPC-specific options passed on RPC invocation. + + This class and its instances have no supported interface - it exists to + define the type of its instances and its instances exist to be passed to + other functions. + """ + + def __init__(self, disable_compression, subcall_of, credentials): + self.disable_compression = disable_compression + self.subcall_of = subcall_of + self.credentials = credentials + + +def grpc_call_options(disable_compression=False, credentials=None): + """Creates a GRPCCallOptions value to be passed at RPC invocation. + + All parameters are optional and should always be passed by keyword. + + Args: + disable_compression: A boolean indicating whether or not compression should + be disabled for the request object of the RPC. Only valid for + request-unary RPCs. + credentials: A ClientCredentials object to use for the invoked RPC. + """ + return GRPCCallOptions(disable_compression, None, credentials) + + +class GRPCServicerContext(object): + """Exposes gRPC-specific options and behaviors to code servicing RPCs.""" + __metaclass__ = abc.ABCMeta + + @abc.abstractmethod + def peer(self): + """Identifies the peer that invoked the RPC being serviced. + + Returns: + A string identifying the peer that invoked the RPC being serviced. + """ + raise NotImplementedError() + + @abc.abstractmethod + def disable_next_response_compression(self): + """Disables compression of the next response passed by the application.""" + raise NotImplementedError() + + +class GRPCInvocationContext(object): + """Exposes gRPC-specific options and behaviors to code invoking RPCs.""" + __metaclass__ = abc.ABCMeta + + @abc.abstractmethod + def disable_next_request_compression(self): + """Disables compression of the next request passed by the application.""" + raise NotImplementedError() + + +class Server(object): + """Services RPCs.""" + __metaclass__ = abc.ABCMeta + + @abc.abstractmethod + def add_insecure_port(self, address): + """Reserves a port for insecure RPC service once this Server becomes active. + + This method may only be called before calling this Server's start method is + called. + + Args: + address: The address for which to open a port. + + Returns: + An integer port on which RPCs will be serviced after this link has been + started. This is typically the same number as the port number contained + in the passed address, but will likely be different if the port number + contained in the passed address was zero. + """ + raise NotImplementedError() + + @abc.abstractmethod + def add_secure_port(self, address, server_credentials): + """Reserves a port for secure RPC service after this Server becomes active. + + This method may only be called before calling this Server's start method is + called. + + Args: + address: The address for which to open a port. + server_credentials: A ServerCredentials. + + Returns: + An integer port on which RPCs will be serviced after this link has been + started. This is typically the same number as the port number contained + in the passed address, but will likely be different if the port number + contained in the passed address was zero. + """ + raise NotImplementedError() + + @abc.abstractmethod + def start(self): + """Starts this Server's service of RPCs. + + This method may only be called while the server is not serving RPCs (i.e. it + is not idempotent). + """ + raise NotImplementedError() + + @abc.abstractmethod + def stop(self, grace): + """Stops this Server's service of RPCs. + + All calls to this method immediately stop service of new RPCs. When existing + RPCs are aborted is controlled by the grace period parameter passed to this + method. + + This method may be called at any time and is idempotent. Passing a smaller + grace value than has been passed in a previous call will have the effect of + stopping the Server sooner. Passing a larger grace value than has been + passed in a previous call will not have the effect of stopping the sooner + later. + + Args: + grace: A duration of time in seconds to allow existing RPCs to complete + before being aborted by this Server's stopping. May be zero for + immediate abortion of all in-progress RPCs. + + Returns: + A threading.Event that will be set when this Server has completely + stopped. The returned event may not be set until after the full grace + period (if some ongoing RPC continues for the full length of the period) + of it may be set much sooner (such as if this Server had no RPCs underway + at the time it was stopped or if all RPCs that it had underway completed + very early in the grace period). + """ + raise NotImplementedError() diff --git a/src/python/grpcio/grpc/beta/utilities.py b/src/python/grpcio/grpc/beta/utilities.py index 1b5356e3ad9..fb07a765795 100644 --- a/src/python/grpcio/grpc/beta/utilities.py +++ b/src/python/grpcio/grpc/beta/utilities.py @@ -32,7 +32,9 @@ import threading import time -from grpc.beta import beta +# implementations is referenced from specification in this module. +from grpc.beta import implementations # pylint: disable=unused-import +from grpc.beta import interfaces from grpc.framework.foundation import callable_util from grpc.framework.foundation import future @@ -70,7 +72,8 @@ class _ChannelReadyFuture(future.Future): def _update(self, connectivity): with self._condition: - if not self._cancelled and connectivity is beta.ChannelConnectivity.READY: + if (not self._cancelled and + connectivity is interfaces.ChannelConnectivity.READY): self._matured = True self._channel.unsubscribe(self._update) self._condition.notify_all() @@ -141,19 +144,19 @@ class _ChannelReadyFuture(future.Future): def channel_ready_future(channel): - """Creates a future.Future that matures when a beta.Channel is ready. + """Creates a future.Future tracking when an implementations.Channel is ready. - Cancelling the returned future.Future does not tell the given beta.Channel to - abandon attempts it may have been making to connect; cancelling merely - deactivates the return future.Future's subscription to the given - beta.Channel's connectivity. + Cancelling the returned future.Future does not tell the given + implementations.Channel to abandon attempts it may have been making to + connect; cancelling merely deactivates the return future.Future's + subscription to the given implementations.Channel's connectivity. Args: - channel: A beta.Channel. + channel: An implementations.Channel. Returns: A future.Future that matures when the given Channel has connectivity - beta.ChannelConnectivity.READY. + interfaces.ChannelConnectivity.READY. """ ready_future = _ChannelReadyFuture(channel) ready_future.start() diff --git a/src/python/grpcio/grpc/framework/core/_end.py b/src/python/grpcio/grpc/framework/core/_end.py index 336e9c21fd9..8e07d9061e0 100644 --- a/src/python/grpcio/grpc/framework/core/_end.py +++ b/src/python/grpcio/grpc/framework/core/_end.py @@ -168,7 +168,7 @@ class _End(End): def operate( self, group, method, subscription, timeout, initial_metadata=None, - payload=None, completion=None): + payload=None, completion=None, protocol_options=None): """See base.End.operate for specification.""" operation_id = uuid.uuid4() with self._lock: @@ -177,9 +177,9 @@ class _End(End): termination_action = _termination_action( self._lock, self._stats, operation_id, self._cycle) operation = _operation.invocation_operate( - operation_id, group, method, subscription, timeout, initial_metadata, - payload, completion, self._mate.accept_ticket, termination_action, - self._cycle.pool) + operation_id, group, method, subscription, timeout, protocol_options, + initial_metadata, payload, completion, self._mate.accept_ticket, + termination_action, self._cycle.pool) self._cycle.operations[operation_id] = operation return operation.context, operation.operator diff --git a/src/python/grpcio/grpc/framework/core/_ingestion.py b/src/python/grpcio/grpc/framework/core/_ingestion.py index 9a7959a2ddb..4129a8ce43e 100644 --- a/src/python/grpcio/grpc/framework/core/_ingestion.py +++ b/src/python/grpcio/grpc/framework/core/_ingestion.py @@ -140,7 +140,7 @@ class _IngestionManager(_interfaces.IngestionManager): def __init__( self, lock, pool, subscription, subscription_creator, termination_manager, - transmission_manager, expiration_manager): + transmission_manager, expiration_manager, protocol_manager): """Constructor. Args: @@ -157,12 +157,14 @@ class _IngestionManager(_interfaces.IngestionManager): transmission_manager: The _interfaces.TransmissionManager for the operation. expiration_manager: The _interfaces.ExpirationManager for the operation. + protocol_manager: The _interfaces.ProtocolManager for the operation. """ self._lock = lock self._pool = pool self._termination_manager = termination_manager self._transmission_manager = transmission_manager self._expiration_manager = expiration_manager + self._protocol_manager = protocol_manager if subscription is None: self._subscription_creator = subscription_creator @@ -296,6 +298,8 @@ class _IngestionManager(_interfaces.IngestionManager): self._abort_and_notify( base.Outcome.Kind.REMOTE_FAILURE, code, details) elif outcome.return_value.subscription.kind is base.Subscription.Kind.FULL: + self._protocol_manager.set_protocol_receiver( + outcome.return_value.subscription.protocol_receiver) self._operator_post_create(outcome.return_value.subscription) else: # TODO(nathaniel): Support other subscriptions. @@ -378,7 +382,7 @@ class _IngestionManager(_interfaces.IngestionManager): def invocation_ingestion_manager( subscription, lock, pool, termination_manager, transmission_manager, - expiration_manager): + expiration_manager, protocol_manager): """Creates an IngestionManager appropriate for invocation-side use. Args: @@ -390,18 +394,20 @@ def invocation_ingestion_manager( transmission_manager: The _interfaces.TransmissionManager for the operation. expiration_manager: The _interfaces.ExpirationManager for the operation. + protocol_manager: The _interfaces.ProtocolManager for the operation. Returns: An IngestionManager appropriate for invocation-side use. """ return _IngestionManager( lock, pool, subscription, None, termination_manager, transmission_manager, - expiration_manager) + expiration_manager, protocol_manager) def service_ingestion_manager( servicer, operation_context, output_operator, lock, pool, - termination_manager, transmission_manager, expiration_manager): + termination_manager, transmission_manager, expiration_manager, + protocol_manager): """Creates an IngestionManager appropriate for service-side use. The returned IngestionManager will require its set_group_and_name method to be @@ -420,6 +426,7 @@ def service_ingestion_manager( transmission_manager: The _interfaces.TransmissionManager for the operation. expiration_manager: The _interfaces.ExpirationManager for the operation. + protocol_manager: The _interfaces.ProtocolManager for the operation. Returns: An IngestionManager appropriate for service-side use. @@ -428,4 +435,4 @@ def service_ingestion_manager( servicer, operation_context, output_operator) return _IngestionManager( lock, pool, None, subscription_creator, termination_manager, - transmission_manager, expiration_manager) + transmission_manager, expiration_manager, protocol_manager) diff --git a/src/python/grpcio/grpc/framework/core/_interfaces.py b/src/python/grpcio/grpc/framework/core/_interfaces.py index 2a534cb7e7b..ffa686b2b79 100644 --- a/src/python/grpcio/grpc/framework/core/_interfaces.py +++ b/src/python/grpcio/grpc/framework/core/_interfaces.py @@ -111,8 +111,8 @@ class TransmissionManager(object): @abc.abstractmethod def kick_off( - self, group, method, timeout, initial_metadata, payload, completion, - allowance): + self, group, method, timeout, protocol_options, initial_metadata, + payload, completion, allowance): """Transmits the values associated with operation invocation.""" raise NotImplementedError() @@ -203,6 +203,31 @@ class ExpirationManager(object): raise NotImplementedError() +class ProtocolManager(object): + """A manager of protocol-specific values passing through an operation.""" + __metaclass__ = abc.ABCMeta + + @abc.abstractmethod + def set_protocol_receiver(self, protocol_receiver): + """Registers the customer object that will receive protocol objects. + + Args: + protocol_receiver: A base.ProtocolReceiver to which protocol objects for + the operation should be passed. + """ + raise NotImplementedError() + + @abc.abstractmethod + def accept_protocol_context(self, protocol_context): + """Accepts the protocol context object for the operation. + + Args: + protocol_context: An object designated for use as the protocol context + of the operation, with further semantics implementation-determined. + """ + raise NotImplementedError() + + class EmissionManager(base.Operator): """A manager of values emitted by customer code.""" __metaclass__ = abc.ABCMeta diff --git a/src/python/grpcio/grpc/framework/core/_operation.py b/src/python/grpcio/grpc/framework/core/_operation.py index f5679d03569..020c0c9ed9a 100644 --- a/src/python/grpcio/grpc/framework/core/_operation.py +++ b/src/python/grpcio/grpc/framework/core/_operation.py @@ -36,6 +36,7 @@ from grpc.framework.core import _emission from grpc.framework.core import _expiration from grpc.framework.core import _ingestion from grpc.framework.core import _interfaces +from grpc.framework.core import _protocol from grpc.framework.core import _reception from grpc.framework.core import _termination from grpc.framework.core import _transmission @@ -84,8 +85,9 @@ class _EasyOperation(_interfaces.Operation): def invocation_operate( - operation_id, group, method, subscription, timeout, initial_metadata, - payload, completion, ticket_sink, termination_action, pool): + operation_id, group, method, subscription, timeout, protocol_options, + initial_metadata, payload, completion, ticket_sink, termination_action, + pool): """Constructs objects necessary for front-side operation management. Args: @@ -95,6 +97,8 @@ def invocation_operate( subscription: A base.Subscription describing the customer's interest in the results of the operation. timeout: A length of time in seconds to allow for the operation. + protocol_options: A transport-specific, application-specific, and/or + protocol-specific value relating to the invocation. May be None. initial_metadata: An initial metadata value to be sent to the other side of the operation. May be None if the initial metadata will be passed later or if there will be no initial metadata passed at all. @@ -120,23 +124,27 @@ def invocation_operate( operation_id, ticket_sink, lock, pool, termination_manager) expiration_manager = _expiration.invocation_expiration_manager( timeout, lock, termination_manager, transmission_manager) + protocol_manager = _protocol.invocation_protocol_manager( + subscription, lock, pool, termination_manager, transmission_manager, + expiration_manager) operation_context = _context.OperationContext( lock, termination_manager, transmission_manager, expiration_manager) emission_manager = _emission.EmissionManager( lock, termination_manager, transmission_manager, expiration_manager) ingestion_manager = _ingestion.invocation_ingestion_manager( subscription, lock, pool, termination_manager, transmission_manager, - expiration_manager) + expiration_manager, protocol_manager) reception_manager = _reception.ReceptionManager( termination_manager, transmission_manager, expiration_manager, - ingestion_manager) + protocol_manager, ingestion_manager) termination_manager.set_expiration_manager(expiration_manager) transmission_manager.set_expiration_manager(expiration_manager) emission_manager.set_ingestion_manager(ingestion_manager) transmission_manager.kick_off( - group, method, timeout, initial_metadata, payload, completion, None) + group, method, timeout, protocol_options, initial_metadata, payload, + completion, None) return _EasyOperation( lock, termination_manager, transmission_manager, expiration_manager, @@ -170,16 +178,20 @@ def service_operate( ticket.timeout, servicer_package.default_timeout, servicer_package.maximum_timeout, lock, termination_manager, transmission_manager) + protocol_manager = _protocol.service_protocol_manager( + lock, pool, termination_manager, transmission_manager, + expiration_manager) operation_context = _context.OperationContext( lock, termination_manager, transmission_manager, expiration_manager) emission_manager = _emission.EmissionManager( lock, termination_manager, transmission_manager, expiration_manager) ingestion_manager = _ingestion.service_ingestion_manager( servicer_package.servicer, operation_context, emission_manager, lock, - pool, termination_manager, transmission_manager, expiration_manager) + pool, termination_manager, transmission_manager, expiration_manager, + protocol_manager) reception_manager = _reception.ReceptionManager( termination_manager, transmission_manager, expiration_manager, - ingestion_manager) + protocol_manager, ingestion_manager) termination_manager.set_expiration_manager(expiration_manager) transmission_manager.set_expiration_manager(expiration_manager) diff --git a/src/python/grpcio/grpc/framework/core/_protocol.py b/src/python/grpcio/grpc/framework/core/_protocol.py new file mode 100644 index 00000000000..3177b5e302a --- /dev/null +++ b/src/python/grpcio/grpc/framework/core/_protocol.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. + +"""State and behavior for passing protocol objects in an operation.""" + +import collections +import enum + +from grpc.framework.core import _constants +from grpc.framework.core import _interfaces +from grpc.framework.core import _utilities +from grpc.framework.foundation import callable_util +from grpc.framework.interfaces.base import base + +_EXCEPTION_LOG_MESSAGE = 'Exception delivering protocol object!' + +_LOCAL_FAILURE_OUTCOME = _utilities.Outcome( + base.Outcome.Kind.LOCAL_FAILURE, None, None) + + +class _Awaited( + collections.namedtuple('_Awaited', ('kind', 'value',))): + + @enum.unique + class Kind(enum.Enum): + NOT_YET_ARRIVED = 'not yet arrived' + ARRIVED = 'arrived' + +_NOT_YET_ARRIVED = _Awaited(_Awaited.Kind.NOT_YET_ARRIVED, None) +_ARRIVED_AND_NONE = _Awaited(_Awaited.Kind.ARRIVED, None) + + +class _Transitory( + collections.namedtuple('_Transitory', ('kind', 'value',))): + + @enum.unique + class Kind(enum.Enum): + NOT_YET_SEEN = 'not yet seen' + PRESENT = 'present' + GONE = 'gone' + +_NOT_YET_SEEN = _Transitory(_Transitory.Kind.NOT_YET_SEEN, None) +_GONE = _Transitory(_Transitory.Kind.GONE, None) + + +class _ProtocolManager(_interfaces.ProtocolManager): + """An implementation of _interfaces.ExpirationManager.""" + + def __init__( + self, protocol_receiver, lock, pool, termination_manager, + transmission_manager, expiration_manager): + """Constructor. + + Args: + protocol_receiver: An _Awaited wrapping of the base.ProtocolReceiver to + which protocol objects should be passed during the operation. May be + of kind _Awaited.Kind.NOT_YET_ARRIVED if the customer's subscription is + not yet known and may be of kind _Awaited.Kind.ARRIVED but with a value + of None if the customer's subscription did not include a + ProtocolReceiver. + lock: The operation-wide lock. + pool: A thread pool. + termination_manager: The _interfaces.TerminationManager for the operation. + transmission_manager: The _interfaces.TransmissionManager for the + operation. + expiration_manager: The _interfaces.ExpirationManager for the operation. + """ + self._lock = lock + self._pool = pool + self._termination_manager = termination_manager + self._transmission_manager = transmission_manager + self._expiration_manager = expiration_manager + + self._protocol_receiver = protocol_receiver + self._context = _NOT_YET_SEEN + + def _abort_and_notify(self, outcome): + if self._termination_manager.outcome is None: + self._termination_manager.abort(outcome) + self._transmission_manager.abort(outcome) + self._expiration_manager.terminate() + + def _deliver(self, behavior, value): + def deliver(): + delivery_outcome = callable_util.call_logging_exceptions( + behavior, _EXCEPTION_LOG_MESSAGE, value) + if delivery_outcome.kind is callable_util.Outcome.Kind.RAISED: + with self._lock: + self._abort_and_notify(_LOCAL_FAILURE_OUTCOME) + self._pool.submit( + callable_util.with_exceptions_logged( + deliver, _constants.INTERNAL_ERROR_LOG_MESSAGE)) + + def set_protocol_receiver(self, protocol_receiver): + """See _interfaces.ProtocolManager.set_protocol_receiver for spec.""" + self._protocol_receiver = _Awaited(_Awaited.Kind.ARRIVED, protocol_receiver) + if (self._context.kind is _Transitory.Kind.PRESENT and + protocol_receiver is not None): + self._deliver(protocol_receiver.context, self._context.value) + self._context = _GONE + + def accept_protocol_context(self, protocol_context): + """See _interfaces.ProtocolManager.accept_protocol_context for spec.""" + if self._protocol_receiver.kind is _Awaited.Kind.ARRIVED: + if self._protocol_receiver.value is not None: + self._deliver(self._protocol_receiver.value.context, protocol_context) + self._context = _GONE + else: + self._context = _Transitory(_Transitory.Kind.PRESENT, protocol_context) + + +def invocation_protocol_manager( + subscription, lock, pool, termination_manager, transmission_manager, + expiration_manager): + """Creates an _interfaces.ProtocolManager for invocation-side use. + + Args: + subscription: The local customer's subscription to the operation. + lock: The operation-wide lock. + pool: A thread pool. + termination_manager: The _interfaces.TerminationManager for the operation. + transmission_manager: The _interfaces.TransmissionManager for the + operation. + expiration_manager: The _interfaces.ExpirationManager for the operation. + """ + if subscription.kind is base.Subscription.Kind.FULL: + awaited_protocol_receiver = _Awaited( + _Awaited.Kind.ARRIVED, subscription.protocol_receiver) + else: + awaited_protocol_receiver = _ARRIVED_AND_NONE + return _ProtocolManager( + awaited_protocol_receiver, lock, pool, termination_manager, + transmission_manager, expiration_manager) + + +def service_protocol_manager( + lock, pool, termination_manager, transmission_manager, expiration_manager): + """Creates an _interfaces.ProtocolManager for service-side use. + + Args: + lock: The operation-wide lock. + pool: A thread pool. + termination_manager: The _interfaces.TerminationManager for the operation. + transmission_manager: The _interfaces.TransmissionManager for the + operation. + expiration_manager: The _interfaces.ExpirationManager for the operation. + """ + return _ProtocolManager( + _NOT_YET_ARRIVED, lock, pool, termination_manager, transmission_manager, + expiration_manager) diff --git a/src/python/grpcio/grpc/framework/core/_reception.py b/src/python/grpcio/grpc/framework/core/_reception.py index d374cf0c8ce..ff81450dee9 100644 --- a/src/python/grpcio/grpc/framework/core/_reception.py +++ b/src/python/grpcio/grpc/framework/core/_reception.py @@ -51,23 +51,31 @@ _RECEPTION_FAILURE_OUTCOME = _utilities.Outcome( base.Outcome.Kind.RECEPTION_FAILURE, None, None) +def _carrying_protocol_context(ticket): + return ticket.protocol is not None and ticket.protocol.kind in ( + links.Protocol.Kind.INVOCATION_CONTEXT, + links.Protocol.Kind.SERVICER_CONTEXT,) + + class ReceptionManager(_interfaces.ReceptionManager): """A ReceptionManager based around a _Receiver passed to it.""" def __init__( self, termination_manager, transmission_manager, expiration_manager, - ingestion_manager): + protocol_manager, ingestion_manager): """Constructor. Args: termination_manager: The operation's _interfaces.TerminationManager. transmission_manager: The operation's _interfaces.TransmissionManager. expiration_manager: The operation's _interfaces.ExpirationManager. + protocol_manager: The operation's _interfaces.ProtocolManager. ingestion_manager: The operation's _interfaces.IngestionManager. """ self._termination_manager = termination_manager self._transmission_manager = transmission_manager self._expiration_manager = expiration_manager + self._protocol_manager = protocol_manager self._ingestion_manager = ingestion_manager self._lowest_unseen_sequence_number = 0 @@ -100,6 +108,10 @@ class ReceptionManager(_interfaces.ReceptionManager): def _process_one(self, ticket): if ticket.sequence_number == 0: self._ingestion_manager.set_group_and_method(ticket.group, ticket.method) + if _carrying_protocol_context(ticket): + self._protocol_manager.accept_protocol_context(ticket.protocol.value) + else: + self._protocol_manager.accept_protocol_context(None) if ticket.timeout is not None: self._expiration_manager.change_timeout(ticket.timeout) if ticket.termination is None: diff --git a/src/python/grpcio/grpc/framework/core/_transmission.py b/src/python/grpcio/grpc/framework/core/_transmission.py index 8f852cfe9f6..65b12c4160e 100644 --- a/src/python/grpcio/grpc/framework/core/_transmission.py +++ b/src/python/grpcio/grpc/framework/core/_transmission.py @@ -207,18 +207,19 @@ class TransmissionManager(_interfaces.TransmissionManager): self._transmitting = True def kick_off( - self, group, method, timeout, initial_metadata, payload, completion, - allowance): + self, group, method, timeout, protocol_options, initial_metadata, + payload, completion, allowance): """See _interfaces.TransmissionManager.kickoff for specification.""" # TODO(nathaniel): Support other subscriptions. subscription = links.Ticket.Subscription.FULL terminal_metadata, code, message, termination = _explode_completion( completion) self._remote_allowance = 1 if payload is None else 0 + protocol = links.Protocol(links.Protocol.Kind.CALL_OPTION, protocol_options) ticket = links.Ticket( self._operation_id, 0, group, method, subscription, timeout, allowance, initial_metadata, payload, terminal_metadata, code, message, - termination, None) + termination, protocol) self._lowest_unused_sequence_number = 1 self._transmit(ticket) diff --git a/src/python/grpcio/grpc/framework/crust/_calls.py b/src/python/grpcio/grpc/framework/crust/_calls.py index 4c6bf16f432..bff940d7471 100644 --- a/src/python/grpcio/grpc/framework/crust/_calls.py +++ b/src/python/grpcio/grpc/framework/crust/_calls.py @@ -38,10 +38,14 @@ _ITERATOR_EXCEPTION_LOG_MESSAGE = 'Exception iterating over requests!' _EMPTY_COMPLETION = utilities.completion(None, None, None) -def _invoke(end, group, method, timeout, initial_metadata, payload, complete): +def _invoke( + end, group, method, timeout, protocol_options, initial_metadata, payload, + complete): rendezvous = _control.Rendezvous(None, None) + subscription = utilities.full_subscription( + rendezvous, _control.protocol_receiver(rendezvous)) operation_context, operator = end.operate( - group, method, utilities.full_subscription(rendezvous), timeout, + group, method, subscription, timeout, protocol_options=protocol_options, initial_metadata=initial_metadata, payload=payload, completion=_EMPTY_COMPLETION if complete else None) rendezvous.set_operator_and_context(operator, operation_context) @@ -93,36 +97,43 @@ def _event_return_stream( def blocking_unary_unary( - end, group, method, timeout, with_call, initial_metadata, payload): + end, group, method, timeout, with_call, protocol_options, initial_metadata, + payload): """Services in a blocking fashion a unary-unary servicer method.""" rendezvous, unused_operation_context, unused_outcome = _invoke( - end, group, method, timeout, initial_metadata, payload, True) + end, group, method, timeout, protocol_options, initial_metadata, payload, + True) if with_call: return next(rendezvous), rendezvous else: return next(rendezvous) -def future_unary_unary(end, group, method, timeout, initial_metadata, payload): +def future_unary_unary( + end, group, method, timeout, protocol_options, initial_metadata, payload): """Services a value-in value-out servicer method by returning a Future.""" rendezvous, unused_operation_context, unused_outcome = _invoke( - end, group, method, timeout, initial_metadata, payload, True) + end, group, method, timeout, protocol_options, initial_metadata, payload, + True) return rendezvous -def inline_unary_stream(end, group, method, timeout, initial_metadata, payload): +def inline_unary_stream( + end, group, method, timeout, protocol_options, initial_metadata, payload): """Services a value-in stream-out servicer method.""" rendezvous, unused_operation_context, unused_outcome = _invoke( - end, group, method, timeout, initial_metadata, payload, True) + end, group, method, timeout, protocol_options, initial_metadata, payload, + True) return rendezvous def blocking_stream_unary( - end, group, method, timeout, with_call, initial_metadata, payload_iterator, - pool): + end, group, method, timeout, with_call, protocol_options, initial_metadata, + payload_iterator, pool): """Services in a blocking fashion a stream-in value-out servicer method.""" rendezvous, operation_context, outcome = _invoke( - end, group, method, timeout, initial_metadata, None, False) + end, group, method, timeout, protocol_options, initial_metadata, None, + False) if outcome is None: def in_pool(): for payload in payload_iterator: @@ -141,10 +152,12 @@ def blocking_stream_unary( def future_stream_unary( - end, group, method, timeout, initial_metadata, payload_iterator, pool): + end, group, method, timeout, protocol_options, initial_metadata, + payload_iterator, pool): """Services a stream-in value-out servicer method by returning a Future.""" rendezvous, operation_context, outcome = _invoke( - end, group, method, timeout, initial_metadata, None, False) + end, group, method, timeout, protocol_options, initial_metadata, None, + False) if outcome is None: def in_pool(): for payload in payload_iterator: @@ -155,10 +168,12 @@ def future_stream_unary( def inline_stream_stream( - end, group, method, timeout, initial_metadata, payload_iterator, pool): + end, group, method, timeout, protocol_options, initial_metadata, + payload_iterator, pool): """Services a stream-in stream-out servicer method.""" rendezvous, operation_context, outcome = _invoke( - end, group, method, timeout, initial_metadata, None, False) + end, group, method, timeout, protocol_options, initial_metadata, None, + False) if outcome is None: def in_pool(): for payload in payload_iterator: @@ -169,36 +184,40 @@ def inline_stream_stream( def event_unary_unary( - end, group, method, timeout, initial_metadata, payload, receiver, - abortion_callback, pool): + end, group, method, timeout, protocol_options, initial_metadata, payload, + receiver, abortion_callback, pool): rendezvous, operation_context, outcome = _invoke( - end, group, method, timeout, initial_metadata, payload, True) + end, group, method, timeout, protocol_options, initial_metadata, payload, + True) return _event_return_unary( receiver, abortion_callback, rendezvous, operation_context, outcome, pool) def event_unary_stream( - end, group, method, timeout, initial_metadata, payload, + end, group, method, timeout, protocol_options, initial_metadata, payload, receiver, abortion_callback, pool): rendezvous, operation_context, outcome = _invoke( - end, group, method, timeout, initial_metadata, payload, True) + end, group, method, timeout, protocol_options, initial_metadata, payload, + True) return _event_return_stream( receiver, abortion_callback, rendezvous, operation_context, outcome, pool) def event_stream_unary( - end, group, method, timeout, initial_metadata, receiver, abortion_callback, - pool): + end, group, method, timeout, protocol_options, initial_metadata, receiver, + abortion_callback, pool): rendezvous, operation_context, outcome = _invoke( - end, group, method, timeout, initial_metadata, None, False) + end, group, method, timeout, protocol_options, initial_metadata, None, + False) return _event_return_unary( receiver, abortion_callback, rendezvous, operation_context, outcome, pool) def event_stream_stream( - end, group, method, timeout, initial_metadata, receiver, abortion_callback, - pool): + end, group, method, timeout, protocol_options, initial_metadata, receiver, + abortion_callback, pool): rendezvous, operation_context, outcome = _invoke( - end, group, method, timeout, initial_metadata, None, False) + end, group, method, timeout, protocol_options, initial_metadata, None, + False) return _event_return_stream( receiver, abortion_callback, rendezvous, operation_context, outcome, pool) diff --git a/src/python/grpcio/grpc/framework/crust/_control.py b/src/python/grpcio/grpc/framework/crust/_control.py index 7bddf46a572..5e9efdf7322 100644 --- a/src/python/grpcio/grpc/framework/crust/_control.py +++ b/src/python/grpcio/grpc/framework/crust/_control.py @@ -182,6 +182,8 @@ class Rendezvous(base.Operator, future.Future, stream.Consumer, face.Call): self._operator = operator self._operation_context = operation_context + self._protocol_context = _NOT_YET_ARRIVED + self._up_initial_metadata = _NOT_YET_ARRIVED self._up_payload = None self._up_allowance = 1 @@ -442,6 +444,16 @@ class Rendezvous(base.Operator, future.Future, stream.Consumer, face.Call): else: return self._termination.abortion + def protocol_context(self): + with self._condition: + while True: + if self._protocol_context.kind is _Awaited.Kind.ARRIVED: + return self._protocol_context.value + elif self._termination.abortion_error is not None: + raise self._termination.abortion_error + else: + self._condition.wait() + def initial_metadata(self): with self._condition: while True: @@ -514,11 +526,30 @@ class Rendezvous(base.Operator, future.Future, stream.Consumer, face.Call): else: self._down_details = _Transitory(_Transitory.Kind.PRESENT, details) + def set_protocol_context(self, protocol_context): + with self._condition: + self._protocol_context = _Awaited( + _Awaited.Kind.ARRIVED, protocol_context) + self._condition.notify_all() + def set_outcome(self, outcome): with self._condition: return self._set_outcome(outcome) +class _ProtocolReceiver(base.ProtocolReceiver): + + def __init__(self, rendezvous): + self._rendezvous = rendezvous + + def context(self, protocol_context): + self._rendezvous.set_protocol_context(protocol_context) + + +def protocol_receiver(rendezvous): + return _ProtocolReceiver(rendezvous) + + def pool_wrap(behavior, operation_context): """Wraps an operation-related behavior so that it may be called in a pool. diff --git a/src/python/grpcio/grpc/framework/crust/_service.py b/src/python/grpcio/grpc/framework/crust/_service.py index 6ff7249e75c..9903415c099 100644 --- a/src/python/grpcio/grpc/framework/crust/_service.py +++ b/src/python/grpcio/grpc/framework/crust/_service.py @@ -52,6 +52,9 @@ class _ServicerContext(face.ServicerContext): def cancel(self): self._rendezvous.cancel() + def protocol_context(self): + return self._rendezvous.protocol_context() + def invocation_metadata(self): return self._rendezvous.initial_metadata() @@ -71,10 +74,12 @@ class _ServicerContext(face.ServicerContext): def _adaptation(pool, in_pool): def adaptation(operator, operation_context): rendezvous = _control.Rendezvous(operator, operation_context) + subscription = utilities.full_subscription( + rendezvous, _control.protocol_receiver(rendezvous)) outcome = operation_context.add_termination_callback(rendezvous.set_outcome) if outcome is None: pool.submit(_control.pool_wrap(in_pool, operation_context), rendezvous) - return utilities.full_subscription(rendezvous) + return subscription else: raise abandonment.Abandoned() return adaptation @@ -151,6 +156,8 @@ def adapt_event_stream_stream(method, pool): def adapt_multi_method(multi_method, pool): def adaptation(group, method, operator, operation_context): rendezvous = _control.Rendezvous(operator, operation_context) + subscription = utilities.full_subscription( + rendezvous, _control.protocol_receiver(rendezvous)) outcome = operation_context.add_termination_callback(rendezvous.set_outcome) if outcome is None: def in_pool(): @@ -160,7 +167,7 @@ def adapt_multi_method(multi_method, pool): request_consumer.consume(request) request_consumer.terminate() pool.submit(_control.pool_wrap(in_pool, operation_context), rendezvous) - return utilities.full_subscription(rendezvous) + return subscription else: raise abandonment.Abandoned() return adaptation diff --git a/src/python/grpcio/grpc/framework/crust/implementations.py b/src/python/grpcio/grpc/framework/crust/implementations.py index d38fab8ba01..4ebc4e9ae85 100644 --- a/src/python/grpcio/grpc/framework/crust/implementations.py +++ b/src/python/grpcio/grpc/framework/crust/implementations.py @@ -66,22 +66,23 @@ class _UnaryUnaryMultiCallable(face.UnaryUnaryMultiCallable): self._pool = pool def __call__( - self, request, timeout, metadata=None, with_call=False): + self, request, timeout, metadata=None, with_call=False, + protocol_options=None): return _calls.blocking_unary_unary( self._end, self._group, self._method, timeout, with_call, - metadata, request) + protocol_options, metadata, request) - def future(self, request, timeout, metadata=None): + def future(self, request, timeout, metadata=None, protocol_options=None): return _calls.future_unary_unary( - self._end, self._group, self._method, timeout, metadata, - request) + self._end, self._group, self._method, timeout, protocol_options, + metadata, request) def event( self, request, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): return _calls.event_unary_unary( - self._end, self._group, self._method, timeout, metadata, - request, receiver, abortion_callback, self._pool) + self._end, self._group, self._method, timeout, protocol_options, + metadata, request, receiver, abortion_callback, self._pool) class _UnaryStreamMultiCallable(face.UnaryStreamMultiCallable): @@ -92,17 +93,17 @@ class _UnaryStreamMultiCallable(face.UnaryStreamMultiCallable): self._method = method self._pool = pool - def __call__(self, request, timeout, metadata=None): + def __call__(self, request, timeout, metadata=None, protocol_options=None): return _calls.inline_unary_stream( - self._end, self._group, self._method, timeout, metadata, - request) + self._end, self._group, self._method, timeout, protocol_options, + metadata, request) def event( self, request, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): return _calls.event_unary_stream( - self._end, self._group, self._method, timeout, metadata, - request, receiver, abortion_callback, self._pool) + self._end, self._group, self._method, timeout, protocol_options, + metadata, request, receiver, abortion_callback, self._pool) class _StreamUnaryMultiCallable(face.StreamUnaryMultiCallable): @@ -115,21 +116,23 @@ class _StreamUnaryMultiCallable(face.StreamUnaryMultiCallable): def __call__( self, request_iterator, timeout, metadata=None, - with_call=False): + with_call=False, protocol_options=None): return _calls.blocking_stream_unary( self._end, self._group, self._method, timeout, with_call, - metadata, request_iterator, self._pool) + protocol_options, metadata, request_iterator, self._pool) - def future(self, request_iterator, timeout, metadata=None): + def future( + self, request_iterator, timeout, metadata=None, protocol_options=None): return _calls.future_stream_unary( - self._end, self._group, self._method, timeout, metadata, - request_iterator, self._pool) + self._end, self._group, self._method, timeout, protocol_options, + metadata, request_iterator, self._pool) def event( - self, receiver, abortion_callback, timeout, metadata=None): + self, receiver, abortion_callback, timeout, metadata=None, + protocol_options=None): return _calls.event_stream_unary( - self._end, self._group, self._method, timeout, metadata, - receiver, abortion_callback, self._pool) + self._end, self._group, self._method, timeout, protocol_options, + metadata, receiver, abortion_callback, self._pool) class _StreamStreamMultiCallable(face.StreamStreamMultiCallable): @@ -140,16 +143,18 @@ class _StreamStreamMultiCallable(face.StreamStreamMultiCallable): self._method = method self._pool = pool - def __call__(self, request_iterator, timeout, metadata=None): + def __call__( + self, request_iterator, timeout, metadata=None, protocol_options=None): return _calls.inline_stream_stream( - self._end, self._group, self._method, timeout, metadata, - request_iterator, self._pool) + self._end, self._group, self._method, timeout, protocol_options, + metadata, request_iterator, self._pool) def event( - self, receiver, abortion_callback, timeout, metadata=None): + self, receiver, abortion_callback, timeout, metadata=None, + protocol_options=None): return _calls.event_stream_stream( - self._end, self._group, self._method, timeout, metadata, - receiver, abortion_callback, self._pool) + self._end, self._group, self._method, timeout, protocol_options, + metadata, receiver, abortion_callback, self._pool) class _GenericStub(face.GenericStub): @@ -161,66 +166,70 @@ class _GenericStub(face.GenericStub): def blocking_unary_unary( self, group, method, request, timeout, metadata=None, - with_call=None): + with_call=None, protocol_options=None): return _calls.blocking_unary_unary( - self._end, group, method, timeout, with_call, metadata, - request) + self._end, group, method, timeout, with_call, protocol_options, + metadata, request) def future_unary_unary( - self, group, method, request, timeout, metadata=None): + self, group, method, request, timeout, metadata=None, + protocol_options=None): return _calls.future_unary_unary( - self._end, group, method, timeout, metadata, request) + self._end, group, method, timeout, protocol_options, metadata, request) def inline_unary_stream( - self, group, method, request, timeout, metadata=None): + self, group, method, request, timeout, metadata=None, + protocol_options=None): return _calls.inline_unary_stream( - self._end, group, method, timeout, metadata, request) + self._end, group, method, timeout, protocol_options, metadata, request) def blocking_stream_unary( self, group, method, request_iterator, timeout, metadata=None, - with_call=None): + with_call=None, protocol_options=None): return _calls.blocking_stream_unary( - self._end, group, method, timeout, with_call, metadata, - request_iterator, self._pool) + self._end, group, method, timeout, with_call, protocol_options, + metadata, request_iterator, self._pool) def future_stream_unary( - self, group, method, request_iterator, timeout, metadata=None): + self, group, method, request_iterator, timeout, metadata=None, + protocol_options=None): return _calls.future_stream_unary( - self._end, group, method, timeout, metadata, + self._end, group, method, timeout, protocol_options, metadata, request_iterator, self._pool) def inline_stream_stream( - self, group, method, request_iterator, timeout, metadata=None): + self, group, method, request_iterator, timeout, metadata=None, + protocol_options=None): return _calls.inline_stream_stream( - self._end, group, method, timeout, metadata, + self._end, group, method, timeout, protocol_options, metadata, request_iterator, self._pool) def event_unary_unary( self, group, method, request, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): return _calls.event_unary_unary( - self._end, group, method, timeout, metadata, request, + self._end, group, method, timeout, protocol_options, metadata, request, receiver, abortion_callback, self._pool) def event_unary_stream( self, group, method, request, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): return _calls.event_unary_stream( - self._end, group, method, timeout, metadata, request, + self._end, group, method, timeout, protocol_options, metadata, request, receiver, abortion_callback, self._pool) def event_stream_unary( self, group, method, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): return _calls.event_stream_unary( - self._end, group, method, timeout, metadata, receiver, + self._end, group, method, timeout, protocol_options, metadata, receiver, abortion_callback, self._pool) def event_stream_stream( self, group, method, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): return _calls.event_stream_stream( - self._end, group, method, timeout, metadata, receiver, + self._end, group, method, timeout, protocol_options, metadata, receiver, abortion_callback, self._pool) def unary_unary(self, group, method): diff --git a/src/python/grpcio/grpc/framework/interfaces/base/base.py b/src/python/grpcio/grpc/framework/interfaces/base/base.py index 0d9d6b464e2..a1e70be5e8d 100644 --- a/src/python/grpcio/grpc/framework/interfaces/base/base.py +++ b/src/python/grpcio/grpc/framework/interfaces/base/base.py @@ -184,6 +184,19 @@ class Operator(object): """ raise NotImplementedError() +class ProtocolReceiver(object): + """A means of receiving protocol values during an operation.""" + __metaclass__ = abc.ABCMeta + + @abc.abstractmethod + def context(self, protocol_context): + """Accepts the protocol context object for the operation. + + Args: + protocol_context: The protocol context object for the operation. + """ + raise NotImplementedError() + class Subscription(object): """Describes customer code's interest in values from the other side. @@ -199,7 +212,11 @@ class Subscription(object): otherwise. operator: An Operator to be passed values from the other side of the operation. Must be non-None if kind is Kind.FULL. Must be None otherwise. + protocol_receiver: A ProtocolReceiver to be passed protocol objects as they + become available during the operation. Must be non-None if kind is + Kind.FULL. """ + __metaclass__ = abc.ABCMeta @enum.unique class Kind(enum.Enum): @@ -274,7 +291,7 @@ class End(object): @abc.abstractmethod def operate( self, group, method, subscription, timeout, initial_metadata=None, - payload=None, completion=None): + payload=None, completion=None, protocol_options=None): """Commences an operation. Args: @@ -290,6 +307,8 @@ class End(object): payload: An initial payload for the operation. completion: A Completion value indicating the end of transmission to the other side of the operation. + protocol_options: A value specified by the provider of a Base interface + implementation affording custom state and behavior. Returns: A pair of objects affording information about the operation and action diff --git a/src/python/grpcio/grpc/framework/interfaces/base/utilities.py b/src/python/grpcio/grpc/framework/interfaces/base/utilities.py index a9ee1a09816..87a85018f52 100644 --- a/src/python/grpcio/grpc/framework/interfaces/base/utilities.py +++ b/src/python/grpcio/grpc/framework/interfaces/base/utilities.py @@ -45,11 +45,12 @@ class _Subscription( base.Subscription, collections.namedtuple( '_Subscription', - ('kind', 'termination_callback', 'allowance', 'operator',))): + ('kind', 'termination_callback', 'allowance', 'operator', + 'protocol_receiver',))): """A trivial implementation of base.Subscription.""" _NONE_SUBSCRIPTION = _Subscription( - base.Subscription.Kind.NONE, None, None, None) + base.Subscription.Kind.NONE, None, None, None, None) def completion(terminal_metadata, code, message): @@ -66,14 +67,16 @@ def completion(terminal_metadata, code, message): return _Completion(terminal_metadata, code, message) -def full_subscription(operator): +def full_subscription(operator, protocol_receiver): """Creates a "full" base.Subscription for the given base.Operator. Args: operator: A base.Operator to be used in an operation. + protocol_receiver: A base.ProtocolReceiver to be used in an operation. Returns: A base.Subscription of kind base.Subscription.Kind.FULL wrapping the given - base.Operator. + base.Operator and base.ProtocolReceiver. """ - return _Subscription(base.Subscription.Kind.FULL, None, None, operator) + return _Subscription( + base.Subscription.Kind.FULL, None, None, operator, protocol_receiver) diff --git a/src/python/grpcio/grpc/framework/interfaces/face/face.py b/src/python/grpcio/grpc/framework/interfaces/face/face.py index 948e7505b6d..bc9a434a760 100644 --- a/src/python/grpcio/grpc/framework/interfaces/face/face.py +++ b/src/python/grpcio/grpc/framework/interfaces/face/face.py @@ -184,6 +184,16 @@ class RpcContext(object): """ raise NotImplementedError() + @abc.abstractmethod + def protocol_context(self): + """Accesses a custom object specified by an implementation provider. + + Returns: + A value specified by the provider of a Face interface implementation + affording custom state and behavior. + """ + raise NotImplementedError() + class Call(RpcContext): """Invocation-side utility object for an RPC.""" @@ -354,7 +364,8 @@ class UnaryUnaryMultiCallable(object): @abc.abstractmethod def __call__( - self, request, timeout, metadata=None, with_call=False): + self, request, timeout, metadata=None, with_call=False, + protocol_options=None): """Synchronously invokes the underlying RPC. Args: @@ -364,6 +375,8 @@ class UnaryUnaryMultiCallable(object): the RPC. with_call: Whether or not to include return a Call for the RPC in addition to the reponse. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: The response value for the RPC, and a Call for the RPC if with_call was @@ -375,7 +388,7 @@ class UnaryUnaryMultiCallable(object): raise NotImplementedError() @abc.abstractmethod - def future(self, request, timeout, metadata=None): + def future(self, request, timeout, metadata=None, protocol_options=None): """Asynchronously invokes the underlying RPC. Args: @@ -383,6 +396,8 @@ class UnaryUnaryMultiCallable(object): timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: An object that is both a Call for the RPC and a future.Future. In the @@ -395,7 +410,7 @@ class UnaryUnaryMultiCallable(object): @abc.abstractmethod def event( self, request, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): """Asynchronously invokes the underlying RPC. Args: @@ -406,6 +421,8 @@ class UnaryUnaryMultiCallable(object): timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: A Call for the RPC. @@ -418,7 +435,7 @@ class UnaryStreamMultiCallable(object): __metaclass__ = abc.ABCMeta @abc.abstractmethod - def __call__(self, request, timeout, metadata=None): + def __call__(self, request, timeout, metadata=None, protocol_options=None): """Invokes the underlying RPC. Args: @@ -426,6 +443,8 @@ class UnaryStreamMultiCallable(object): timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: An object that is both a Call for the RPC and an iterator of response @@ -437,7 +456,7 @@ class UnaryStreamMultiCallable(object): @abc.abstractmethod def event( self, request, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): """Asynchronously invokes the underlying RPC. Args: @@ -448,6 +467,8 @@ class UnaryStreamMultiCallable(object): timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: A Call object for the RPC. @@ -462,7 +483,7 @@ class StreamUnaryMultiCallable(object): @abc.abstractmethod def __call__( self, request_iterator, timeout, metadata=None, - with_call=False): + with_call=False, protocol_options=None): """Synchronously invokes the underlying RPC. Args: @@ -472,6 +493,8 @@ class StreamUnaryMultiCallable(object): the RPC. with_call: Whether or not to include return a Call for the RPC in addition to the reponse. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: The response value for the RPC, and a Call for the RPC if with_call was @@ -483,7 +506,8 @@ class StreamUnaryMultiCallable(object): raise NotImplementedError() @abc.abstractmethod - def future(self, request_iterator, timeout, metadata=None): + def future( + self, request_iterator, timeout, metadata=None, protocol_options=None): """Asynchronously invokes the underlying RPC. Args: @@ -491,6 +515,8 @@ class StreamUnaryMultiCallable(object): timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: An object that is both a Call for the RPC and a future.Future. In the @@ -502,7 +528,8 @@ class StreamUnaryMultiCallable(object): @abc.abstractmethod def event( - self, receiver, abortion_callback, timeout, metadata=None): + self, receiver, abortion_callback, timeout, metadata=None, + protocol_options=None): """Asynchronously invokes the underlying RPC. Args: @@ -512,6 +539,8 @@ class StreamUnaryMultiCallable(object): timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: A single object that is both a Call object for the RPC and a @@ -525,7 +554,8 @@ class StreamStreamMultiCallable(object): __metaclass__ = abc.ABCMeta @abc.abstractmethod - def __call__(self, request_iterator, timeout, metadata=None): + def __call__( + self, request_iterator, timeout, metadata=None, protocol_options=None): """Invokes the underlying RPC. Args: @@ -533,6 +563,8 @@ class StreamStreamMultiCallable(object): timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: An object that is both a Call for the RPC and an iterator of response @@ -543,7 +575,8 @@ class StreamStreamMultiCallable(object): @abc.abstractmethod def event( - self, receiver, abortion_callback, timeout, metadata=None): + self, receiver, abortion_callback, timeout, metadata=None, + protocol_options=None): """Asynchronously invokes the underlying RPC. Args: @@ -553,6 +586,8 @@ class StreamStreamMultiCallable(object): timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: A single object that is both a Call object for the RPC and a @@ -646,7 +681,7 @@ class GenericStub(object): @abc.abstractmethod def blocking_unary_unary( self, group, method, request, timeout, metadata=None, - with_call=False): + with_call=False, protocol_options=None): """Invokes a unary-request-unary-response method. This method blocks until either returning the response value of the RPC @@ -661,6 +696,8 @@ class GenericStub(object): metadata: A metadata value to be passed to the service-side of the RPC. with_call: Whether or not to include return a Call for the RPC in addition to the reponse. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: The response value for the RPC, and a Call for the RPC if with_call was @@ -673,7 +710,8 @@ class GenericStub(object): @abc.abstractmethod def future_unary_unary( - self, group, method, request, timeout, metadata=None): + self, group, method, request, timeout, metadata=None, + protocol_options=None): """Invokes a unary-request-unary-response method. Args: @@ -682,6 +720,8 @@ class GenericStub(object): request: The request value for the RPC. timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: An object that is both a Call for the RPC and a future.Future. In the @@ -693,7 +733,8 @@ class GenericStub(object): @abc.abstractmethod def inline_unary_stream( - self, group, method, request, timeout, metadata=None): + self, group, method, request, timeout, metadata=None, + protocol_options=None): """Invokes a unary-request-stream-response method. Args: @@ -702,6 +743,8 @@ class GenericStub(object): request: The request value for the RPC. timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: An object that is both a Call for the RPC and an iterator of response @@ -713,7 +756,7 @@ class GenericStub(object): @abc.abstractmethod def blocking_stream_unary( self, group, method, request_iterator, timeout, metadata=None, - with_call=False): + with_call=False, protocol_options=None): """Invokes a stream-request-unary-response method. This method blocks until either returning the response value of the RPC @@ -728,6 +771,8 @@ class GenericStub(object): metadata: A metadata value to be passed to the service-side of the RPC. with_call: Whether or not to include return a Call for the RPC in addition to the reponse. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: The response value for the RPC, and a Call for the RPC if with_call was @@ -740,7 +785,8 @@ class GenericStub(object): @abc.abstractmethod def future_stream_unary( - self, group, method, request_iterator, timeout, metadata=None): + self, group, method, request_iterator, timeout, metadata=None, + protocol_options=None): """Invokes a stream-request-unary-response method. Args: @@ -749,6 +795,8 @@ class GenericStub(object): request_iterator: An iterator that yields request values for the RPC. timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: An object that is both a Call for the RPC and a future.Future. In the @@ -760,7 +808,8 @@ class GenericStub(object): @abc.abstractmethod def inline_stream_stream( - self, group, method, request_iterator, timeout, metadata=None): + self, group, method, request_iterator, timeout, metadata=None, + protocol_options=None): """Invokes a stream-request-stream-response method. Args: @@ -769,6 +818,8 @@ class GenericStub(object): request_iterator: An iterator that yields request values for the RPC. timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: An object that is both a Call for the RPC and an iterator of response @@ -780,7 +831,7 @@ class GenericStub(object): @abc.abstractmethod def event_unary_unary( self, group, method, request, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): """Event-driven invocation of a unary-request-unary-response method. Args: @@ -792,6 +843,8 @@ class GenericStub(object): in the event of RPC abortion. timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: A Call for the RPC. @@ -801,7 +854,7 @@ class GenericStub(object): @abc.abstractmethod def event_unary_stream( self, group, method, request, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): """Event-driven invocation of a unary-request-stream-response method. Args: @@ -813,6 +866,8 @@ class GenericStub(object): in the event of RPC abortion. timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: A Call for the RPC. @@ -822,7 +877,7 @@ class GenericStub(object): @abc.abstractmethod def event_stream_unary( self, group, method, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): """Event-driven invocation of a unary-request-unary-response method. Args: @@ -833,6 +888,8 @@ class GenericStub(object): in the event of RPC abortion. timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: A pair of a Call object for the RPC and a stream.Consumer to which the @@ -843,7 +900,7 @@ class GenericStub(object): @abc.abstractmethod def event_stream_stream( self, group, method, receiver, abortion_callback, timeout, - metadata=None): + metadata=None, protocol_options=None): """Event-driven invocation of a unary-request-stream-response method. Args: @@ -854,6 +911,8 @@ class GenericStub(object): in the event of RPC abortion. timeout: A duration of time in seconds to allow for the RPC. metadata: A metadata value to be passed to the service-side of the RPC. + protocol_options: A value specified by the provider of a Face interface + implementation affording custom state and behavior. Returns: A pair of a Call object for the RPC and a stream.Consumer to which the diff --git a/src/python/grpcio/grpc/framework/interfaces/links/links.py b/src/python/grpcio/grpc/framework/interfaces/links/links.py index b98a30a3990..24f0e3b3545 100644 --- a/src/python/grpcio/grpc/framework/interfaces/links/links.py +++ b/src/python/grpcio/grpc/framework/interfaces/links/links.py @@ -34,14 +34,13 @@ import collections import enum -class Transport(collections.namedtuple('Transport', ('kind', 'value',))): - """A sum type for handles to an underlying transport system. +class Protocol(collections.namedtuple('Protocol', ('kind', 'value',))): + """A sum type for handles to a system that transmits tickets. Attributes: - kind: A Kind value identifying the kind of value being passed to or from - the underlying transport. - value: The value being passed through RPC Framework between the high-level - application and the underlying transport. + kind: A Kind value identifying the kind of value being passed. + value: The value being passed between the high-level application and the + system affording ticket transport. """ @enum.unique @@ -56,8 +55,7 @@ class Ticket( 'Ticket', ('operation_id', 'sequence_number', 'group', 'method', 'subscription', 'timeout', 'allowance', 'initial_metadata', 'payload', - 'terminal_metadata', 'code', 'message', 'termination', - 'transport',))): + 'terminal_metadata', 'code', 'message', 'termination', 'protocol',))): """A sum type for all values sent from a front to a back. Attributes: @@ -99,8 +97,8 @@ class Ticket( termination: A Termination value describing the end of the operation, or None if the operation has not yet terminated. If set, no further tickets may be sent in the same direction. - transport: A Transport value or None, with further semantics being a matter - between high-level application and underlying transport. + protocol: A Protocol value or None, with further semantics being a matter + between high-level application and underlying ticket transport. """ @enum.unique diff --git a/src/python/grpcio_test/grpc_protoc_plugin/beta_python_plugin_test.py b/src/python/grpcio_test/grpc_protoc_plugin/beta_python_plugin_test.py index 4c8c64b06d4..259b978de25 100644 --- a/src/python/grpcio_test/grpc_protoc_plugin/beta_python_plugin_test.py +++ b/src/python/grpcio_test/grpc_protoc_plugin/beta_python_plugin_test.py @@ -42,7 +42,7 @@ import threading import time import unittest -from grpc.beta import beta +from grpc.beta import implementations from grpc.framework.foundation import future from grpc.framework.interfaces.face import face from grpc_test.framework.common import test_constants @@ -170,7 +170,7 @@ def _CreateService(test_pb2): server = getattr(test_pb2, SERVER_FACTORY_IDENTIFIER)(servicer) port = server.add_insecure_port('[::]:0') server.start() - channel = beta.create_insecure_channel('localhost', port) + channel = implementations.insecure_channel('localhost', port) stub = getattr(test_pb2, STUB_FACTORY_IDENTIFIER)(channel) yield servicer_methods, stub server.stop(0) diff --git a/src/python/grpcio_test/grpc_test/_adapter/_intermediary_low_test.py b/src/python/grpcio_test/grpc_test/_adapter/_intermediary_low_test.py index 27a5b82e9c4..90ad0b9bcb5 100644 --- a/src/python/grpcio_test/grpc_test/_adapter/_intermediary_low_test.py +++ b/src/python/grpcio_test/grpc_test/_adapter/_intermediary_low_test.py @@ -191,7 +191,7 @@ class EchoTest(unittest.TestCase): metadata[server_leading_binary_metadata_key]) for datum in test_data: - client_call.write(datum, write_tag) + client_call.write(datum, write_tag, _low.WriteFlags.WRITE_NO_COMPRESS) write_accepted = self.client_events.get() self.assertIsNotNone(write_accepted) self.assertIs(write_accepted.kind, _low.Event.Kind.WRITE_ACCEPTED) @@ -206,7 +206,7 @@ class EchoTest(unittest.TestCase): self.assertIsNotNone(read_accepted.bytes) server_data.append(read_accepted.bytes) - server_call.write(read_accepted.bytes, write_tag) + server_call.write(read_accepted.bytes, write_tag, 0) write_accepted = self.server_events.get() self.assertIsNotNone(write_accepted) self.assertEqual(_low.Event.Kind.WRITE_ACCEPTED, write_accepted.kind) @@ -370,14 +370,14 @@ class CancellationTest(unittest.TestCase): self.assertIsNotNone(metadata_accepted) for datum in test_data: - client_call.write(datum, write_tag) + client_call.write(datum, write_tag, 0) write_accepted = self.client_events.get() server_call.read(read_tag) read_accepted = self.server_events.get() server_data.append(read_accepted.bytes) - server_call.write(read_accepted.bytes, write_tag) + server_call.write(read_accepted.bytes, write_tag, 0) write_accepted = self.server_events.get() self.assertIsNotNone(write_accepted) diff --git a/src/python/grpcio_test/grpc_test/beta/_beta_features_test.py b/src/python/grpcio_test/grpc_test/beta/_beta_features_test.py new file mode 100644 index 00000000000..fad57da9d06 --- /dev/null +++ b/src/python/grpcio_test/grpc_test/beta/_beta_features_test.py @@ -0,0 +1,232 @@ +# 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. + +"""Tests Face interface compliance of the gRPC Python Beta API.""" + +import threading +import unittest + +from grpc.beta import implementations +from grpc.beta import interfaces +from grpc.framework.common import cardinality +from grpc.framework.interfaces.face import utilities +from grpc_test import resources +from grpc_test.beta import test_utilities +from grpc_test.framework.common import test_constants + +_SERVER_HOST_OVERRIDE = 'foo.test.google.fr' + +_GROUP = 'group' +_UNARY_UNARY = 'unary-unary' +_UNARY_STREAM = 'unary-stream' +_STREAM_UNARY = 'stream-unary' +_STREAM_STREAM = 'stream-stream' + +_REQUEST = b'abc' +_RESPONSE = b'123' + + +class _Servicer(object): + + def __init__(self): + self._condition = threading.Condition() + self._peer = None + self._serviced = False + + def unary_unary(self, request, context): + with self._condition: + self._request = request + self._peer = context.protocol_context().peer() + context.protocol_context().disable_next_response_compression() + self._serviced = True + self._condition.notify_all() + return _RESPONSE + + def unary_stream(self, request, context): + with self._condition: + self._request = request + self._peer = context.protocol_context().peer() + context.protocol_context().disable_next_response_compression() + self._serviced = True + self._condition.notify_all() + return + yield + + def stream_unary(self, request_iterator, context): + for request in request_iterator: + self._request = request + with self._condition: + self._peer = context.protocol_context().peer() + context.protocol_context().disable_next_response_compression() + self._serviced = True + self._condition.notify_all() + return _RESPONSE + + def stream_stream(self, request_iterator, context): + for request in request_iterator: + with self._condition: + self._peer = context.protocol_context().peer() + context.protocol_context().disable_next_response_compression() + yield _RESPONSE + with self._condition: + self._serviced = True + self._condition.notify_all() + + def peer(self): + with self._condition: + return self._peer + + def block_until_serviced(self): + with self._condition: + while not self._serviced: + self._condition.wait() + + +class _BlockingIterator(object): + + def __init__(self, upstream): + self._condition = threading.Condition() + self._upstream = upstream + self._allowed = [] + + def __iter__(self): + return self + + def next(self): + with self._condition: + while True: + if self._allowed is None: + raise StopIteration() + elif self._allowed: + return self._allowed.pop(0) + else: + self._condition.wait() + + def allow(self): + with self._condition: + try: + self._allowed.append(next(self._upstream)) + except StopIteration: + self._allowed = None + self._condition.notify_all() + + +class BetaFeaturesTest(unittest.TestCase): + + def setUp(self): + self._servicer = _Servicer() + method_implementations = { + (_GROUP, _UNARY_UNARY): + utilities.unary_unary_inline(self._servicer.unary_unary), + (_GROUP, _UNARY_STREAM): + utilities.unary_stream_inline(self._servicer.unary_stream), + (_GROUP, _STREAM_UNARY): + utilities.stream_unary_inline(self._servicer.stream_unary), + (_GROUP, _STREAM_STREAM): + utilities.stream_stream_inline(self._servicer.stream_stream), + } + + cardinalities = { + _UNARY_UNARY: cardinality.Cardinality.UNARY_UNARY, + _UNARY_STREAM: cardinality.Cardinality.UNARY_STREAM, + _STREAM_UNARY: cardinality.Cardinality.STREAM_UNARY, + _STREAM_STREAM: cardinality.Cardinality.STREAM_STREAM, + } + + server_options = implementations.server_options( + thread_pool_size=test_constants.POOL_SIZE) + self._server = implementations.server( + method_implementations, options=server_options) + server_credentials = implementations.ssl_server_credentials( + [(resources.private_key(), resources.certificate_chain(),),]) + port = self._server.add_secure_port('[::]:0', server_credentials) + self._server.start() + self._client_credentials = implementations.ssl_client_credentials( + resources.test_root_certificates(), None, None) + channel = test_utilities.not_really_secure_channel( + 'localhost', port, self._client_credentials, _SERVER_HOST_OVERRIDE) + stub_options = implementations.stub_options( + thread_pool_size=test_constants.POOL_SIZE) + self._dynamic_stub = implementations.dynamic_stub( + channel, _GROUP, cardinalities, options=stub_options) + + def tearDown(self): + self._dynamic_stub = None + self._server.stop(test_constants.SHORT_TIMEOUT).wait() + + def test_unary_unary(self): + call_options = interfaces.grpc_call_options( + disable_compression=True, credentials=self._client_credentials) + response = getattr(self._dynamic_stub, _UNARY_UNARY)( + _REQUEST, test_constants.LONG_TIMEOUT, protocol_options=call_options) + self.assertEqual(_RESPONSE, response) + self.assertIsNotNone(self._servicer.peer()) + + def test_unary_stream(self): + call_options = interfaces.grpc_call_options( + disable_compression=True, credentials=self._client_credentials) + response_iterator = getattr(self._dynamic_stub, _UNARY_STREAM)( + _REQUEST, test_constants.LONG_TIMEOUT, protocol_options=call_options) + self._servicer.block_until_serviced() + self.assertIsNotNone(self._servicer.peer()) + + def test_stream_unary(self): + call_options = interfaces.grpc_call_options( + credentials=self._client_credentials) + request_iterator = _BlockingIterator(iter((_REQUEST,))) + response_future = getattr(self._dynamic_stub, _STREAM_UNARY).future( + request_iterator, test_constants.LONG_TIMEOUT, + protocol_options=call_options) + response_future.protocol_context().disable_next_request_compression() + request_iterator.allow() + response_future.protocol_context().disable_next_request_compression() + request_iterator.allow() + self._servicer.block_until_serviced() + self.assertIsNotNone(self._servicer.peer()) + self.assertEqual(_RESPONSE, response_future.result()) + + def test_stream_stream(self): + call_options = interfaces.grpc_call_options( + credentials=self._client_credentials) + request_iterator = _BlockingIterator(iter((_REQUEST,))) + response_iterator = getattr(self._dynamic_stub, _STREAM_STREAM)( + request_iterator, test_constants.SHORT_TIMEOUT, + protocol_options=call_options) + response_iterator.protocol_context().disable_next_request_compression() + request_iterator.allow() + response = next(response_iterator) + response_iterator.protocol_context().disable_next_request_compression() + request_iterator.allow() + self._servicer.block_until_serviced() + self.assertIsNotNone(self._servicer.peer()) + self.assertEqual(_RESPONSE, response) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/grpcio_test/grpc_test/beta/_connectivity_channel_test.py b/src/python/grpcio_test/grpc_test/beta/_connectivity_channel_test.py index 038464889d6..b3c05bdb0c9 100644 --- a/src/python/grpcio_test/grpc_test/beta/_connectivity_channel_test.py +++ b/src/python/grpcio_test/grpc_test/beta/_connectivity_channel_test.py @@ -36,14 +36,9 @@ import unittest from grpc._adapter import _low from grpc._adapter import _types from grpc.beta import _connectivity_channel +from grpc.beta import interfaces from grpc_test.framework.common import test_constants -_MAPPING_FUNCTION = lambda integer: integer * 200 + 17 -_MAPPING = { - state: _MAPPING_FUNCTION(state) for state in _types.ConnectivityState} -_IDLE, _CONNECTING, _READY, _TRANSIENT_FAILURE, _FATAL_FAILURE = map( - _MAPPING_FUNCTION, _types.ConnectivityState) - def _drive_completion_queue(completion_queue): while True: @@ -84,7 +79,7 @@ class ChannelConnectivityTest(unittest.TestCase): callback = _Callback() connectivity_channel = _connectivity_channel.ConnectivityChannel( - low_channel, _MAPPING) + low_channel) connectivity_channel.subscribe(callback.update, try_to_connect=False) first_connectivities = callback.block_until_connectivities_satisfy(bool) connectivity_channel.subscribe(callback.update, try_to_connect=True) @@ -98,11 +93,16 @@ class ChannelConnectivityTest(unittest.TestCase): connectivity_channel.unsubscribe(callback.update) fifth_connectivities = callback.connectivities() - self.assertSequenceEqual((_IDLE,), first_connectivities) - self.assertNotIn(_READY, second_connectivities) - self.assertNotIn(_READY, third_connectivities) - self.assertNotIn(_READY, fourth_connectivities) - self.assertNotIn(_READY, fifth_connectivities) + self.assertSequenceEqual( + (interfaces.ChannelConnectivity.IDLE,), first_connectivities) + self.assertNotIn( + interfaces.ChannelConnectivity.READY, second_connectivities) + self.assertNotIn( + interfaces.ChannelConnectivity.READY, third_connectivities) + self.assertNotIn( + interfaces.ChannelConnectivity.READY, fourth_connectivities) + self.assertNotIn( + interfaces.ChannelConnectivity.READY, fifth_connectivities) def test_immediately_connectable_channel_connectivity(self): server_completion_queue = _low.CompletionQueue() @@ -117,7 +117,7 @@ class ChannelConnectivityTest(unittest.TestCase): second_callback = _Callback() connectivity_channel = _connectivity_channel.ConnectivityChannel( - low_channel, _MAPPING) + low_channel) connectivity_channel.subscribe(first_callback.update, try_to_connect=False) first_connectivities = first_callback.block_until_connectivities_satisfy( bool) @@ -132,9 +132,11 @@ class ChannelConnectivityTest(unittest.TestCase): bool) # Wait for a connection that will happen (or may already have happened). first_callback.block_until_connectivities_satisfy( - lambda connectivities: _READY in connectivities) + lambda connectivities: + interfaces.ChannelConnectivity.READY in connectivities) second_callback.block_until_connectivities_satisfy( - lambda connectivities: _READY in connectivities) + lambda connectivities: + interfaces.ChannelConnectivity.READY in connectivities) connectivity_channel.unsubscribe(first_callback.update) connectivity_channel.unsubscribe(second_callback.update) @@ -142,12 +144,19 @@ class ChannelConnectivityTest(unittest.TestCase): server_completion_queue.shutdown() server_completion_queue_thread.join() - self.assertSequenceEqual((_IDLE,), first_connectivities) - self.assertSequenceEqual((_IDLE,), second_connectivities) - self.assertNotIn(_TRANSIENT_FAILURE, third_connectivities) - self.assertNotIn(_FATAL_FAILURE, third_connectivities) - self.assertNotIn(_TRANSIENT_FAILURE, fourth_connectivities) - self.assertNotIn(_FATAL_FAILURE, fourth_connectivities) + self.assertSequenceEqual( + (interfaces.ChannelConnectivity.IDLE,), first_connectivities) + self.assertSequenceEqual( + (interfaces.ChannelConnectivity.IDLE,), second_connectivities) + self.assertNotIn( + interfaces.ChannelConnectivity.TRANSIENT_FAILURE, third_connectivities) + self.assertNotIn( + interfaces.ChannelConnectivity.FATAL_FAILURE, third_connectivities) + self.assertNotIn( + interfaces.ChannelConnectivity.TRANSIENT_FAILURE, + fourth_connectivities) + self.assertNotIn( + interfaces.ChannelConnectivity.FATAL_FAILURE, fourth_connectivities) def test_reachable_then_unreachable_channel_connectivity(self): server_completion_queue = _low.CompletionQueue() @@ -161,14 +170,16 @@ class ChannelConnectivityTest(unittest.TestCase): callback = _Callback() connectivity_channel = _connectivity_channel.ConnectivityChannel( - low_channel, _MAPPING) + low_channel) connectivity_channel.subscribe(callback.update, try_to_connect=True) callback.block_until_connectivities_satisfy( - lambda connectivities: _READY in connectivities) + lambda connectivities: + interfaces.ChannelConnectivity.READY in connectivities) # Now take down the server and confirm that channel readiness is repudiated. server.shutdown() callback.block_until_connectivities_satisfy( - lambda connectivities: connectivities[-1] is not _READY) + lambda connectivities: + connectivities[-1] is not interfaces.ChannelConnectivity.READY) connectivity_channel.unsubscribe(callback.update) server.shutdown() diff --git a/src/python/grpcio_test/grpc_test/beta/_face_interface_test.py b/src/python/grpcio_test/grpc_test/beta/_face_interface_test.py index e9087a79493..aa33e1e6f86 100644 --- a/src/python/grpcio_test/grpc_test/beta/_face_interface_test.py +++ b/src/python/grpcio_test/grpc_test/beta/_face_interface_test.py @@ -32,7 +32,7 @@ import collections import unittest -from grpc.beta import beta +from grpc.beta import implementations from grpc.beta import interfaces from grpc_test import resources from grpc_test import test_common as grpc_test_common @@ -81,25 +81,26 @@ class _Implementation(test_interfaces.Implementation): method: method_object.cardinality() for (group, method), method_object in methods.iteritems()} - server_options = beta.server_options( + server_options = implementations.server_options( request_deserializers=serialization_behaviors.request_deserializers, response_serializers=serialization_behaviors.response_serializers, thread_pool_size=test_constants.POOL_SIZE) - server = beta.server(method_implementations, options=server_options) - server_credentials = beta.ssl_server_credentials( + server = implementations.server( + method_implementations, options=server_options) + server_credentials = implementations.ssl_server_credentials( [(resources.private_key(), resources.certificate_chain(),),]) port = server.add_secure_port('[::]:0', server_credentials) server.start() - client_credentials = beta.ssl_client_credentials( + client_credentials = implementations.ssl_client_credentials( resources.test_root_certificates(), None, None) - channel = test_utilities.create_not_really_secure_channel( + channel = test_utilities.not_really_secure_channel( 'localhost', port, client_credentials, _SERVER_HOST_OVERRIDE) - stub_options = beta.stub_options( + stub_options = implementations.stub_options( request_serializers=serialization_behaviors.request_serializers, response_deserializers=serialization_behaviors.response_deserializers, thread_pool_size=test_constants.POOL_SIZE) - generic_stub = beta.generic_stub(channel, options=stub_options) - dynamic_stub = beta.dynamic_stub( + generic_stub = implementations.generic_stub(channel, options=stub_options) + dynamic_stub = implementations.dynamic_stub( channel, service, cardinalities, options=stub_options) return generic_stub, {service: dynamic_stub}, server diff --git a/src/python/grpcio_test/grpc_test/beta/_not_found_test.py b/src/python/grpcio_test/grpc_test/beta/_not_found_test.py index ecd10f2175a..5feb997fef6 100644 --- a/src/python/grpcio_test/grpc_test/beta/_not_found_test.py +++ b/src/python/grpcio_test/grpc_test/beta/_not_found_test.py @@ -31,7 +31,7 @@ import unittest -from grpc.beta import beta +from grpc.beta import implementations from grpc.beta import interfaces from grpc.framework.interfaces.face import face from grpc_test.framework.common import test_constants @@ -40,10 +40,10 @@ from grpc_test.framework.common import test_constants class NotFoundTest(unittest.TestCase): def setUp(self): - self._server = beta.server({}) + self._server = implementations.server({}) port = self._server.add_insecure_port('[::]:0') - channel = beta.create_insecure_channel('localhost', port) - self._generic_stub = beta.generic_stub(channel) + channel = implementations.insecure_channel('localhost', port) + self._generic_stub = implementations.generic_stub(channel) self._server.start() def tearDown(self): diff --git a/src/python/grpcio_test/grpc_test/beta/_utilities_test.py b/src/python/grpcio_test/grpc_test/beta/_utilities_test.py index 998e74ccf48..996cea91189 100644 --- a/src/python/grpcio_test/grpc_test/beta/_utilities_test.py +++ b/src/python/grpcio_test/grpc_test/beta/_utilities_test.py @@ -35,7 +35,7 @@ import unittest from grpc._adapter import _low from grpc._adapter import _types -from grpc.beta import beta +from grpc.beta import implementations from grpc.beta import utilities from grpc.framework.foundation import future from grpc_test.framework.common import test_constants @@ -69,7 +69,7 @@ class _Callback(object): class ChannelConnectivityTest(unittest.TestCase): def test_lonely_channel_connectivity(self): - channel = beta.create_insecure_channel('localhost', 12345) + channel = implementations.insecure_channel('localhost', 12345) callback = _Callback() ready_future = utilities.channel_ready_future(channel) @@ -94,7 +94,7 @@ class ChannelConnectivityTest(unittest.TestCase): server_completion_queue_thread = threading.Thread( target=_drive_completion_queue, args=(server_completion_queue,)) server_completion_queue_thread.start() - channel = beta.create_insecure_channel('localhost', port) + channel = implementations.insecure_channel('localhost', port) callback = _Callback() try: diff --git a/src/python/grpcio_test/grpc_test/beta/test_utilities.py b/src/python/grpcio_test/grpc_test/beta/test_utilities.py index 338670478d5..24a8600e12b 100644 --- a/src/python/grpcio_test/grpc_test/beta/test_utilities.py +++ b/src/python/grpcio_test/grpc_test/beta/test_utilities.py @@ -30,25 +30,27 @@ """Test-appropriate entry points into the gRPC Python Beta API.""" from grpc._adapter import _intermediary_low -from grpc.beta import beta +from grpc.beta import implementations -def create_not_really_secure_channel( +def not_really_secure_channel( host, port, client_credentials, server_host_override): """Creates an insecure Channel to a remote host. Args: host: The name of the remote host to which to connect. port: The port of the remote host to which to connect. - client_credentials: The beta.ClientCredentials with which to connect. + client_credentials: The implementations.ClientCredentials with which to + connect. server_host_override: The target name used for SSL host name checking. Returns: - A beta.Channel to the remote host through which RPCs may be conducted. + An implementations.Channel to the remote host through which RPCs may be + conducted. """ hostport = '%s:%d' % (host, port) intermediary_low_channel = _intermediary_low.Channel( hostport, client_credentials._intermediary_low_credentials, server_host_override=server_host_override) - return beta.Channel( + return implementations.Channel( intermediary_low_channel._internal, intermediary_low_channel) diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py index 5065a3f38a8..ddda1018c34 100644 --- a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py +++ b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py @@ -119,6 +119,17 @@ class _Operator(base.Operator): 'Deliberately raised exception from Operator.advance (in a test)!') +class _ProtocolReceiver(base.ProtocolReceiver): + + def __init__(self): + self._condition = threading.Condition() + self._contexts = [] + + def context(self, protocol_context): + with self._condition: + self._contexts.append(protocol_context) + + class _Servicer(base.Servicer): """A base.Servicer with instrumented for testing.""" @@ -144,7 +155,7 @@ class _Servicer(base.Servicer): controller.service_on_termination) if outcome is not None: controller.service_on_termination(outcome) - return utilities.full_subscription(operator) + return utilities.full_subscription(operator, _ProtocolReceiver()) class _OperationTest(unittest.TestCase): @@ -169,7 +180,8 @@ class _OperationTest(unittest.TestCase): test_operator = _Operator( self._controller, self._controller.on_invocation_advance, self._pool, None) - subscription = utilities.full_subscription(test_operator) + subscription = utilities.full_subscription( + test_operator, _ProtocolReceiver()) else: # TODO(nathaniel): support and test other subscription kinds. self.fail('Non-full subscriptions not yet supported!') diff --git a/src/ruby/README.md b/src/ruby/README.md index 7f75c0e313d..8c56ceb1352 100644 --- a/src/ruby/README.md +++ b/src/ruby/README.md @@ -19,10 +19,10 @@ INSTALLATION **Linux (Debian):** -Add [Debian testing][] to your `sources.list` file. Example: +Add [Debian jessie-backports][] to your `sources.list` file. Example: ```sh -echo "deb http://ftp.us.debian.org/debian testing main contrib non-free" | \ +echo "deb http://http.debian.net/debian jessie-backports main" | \ sudo tee -a /etc/apt/sources.list ``` @@ -99,4 +99,4 @@ Directory structure is the layout for [ruby extensions][] [ruby extensions]:http://guides.rubygems.org/gems-with-extensions/ [rubydoc]: http://www.rubydoc.info/gems/grpc [grpc.io]: http://www.grpc.io/docs/installation/ruby.html -[Debian testing]:https://www.debian.org/releases/stretch/ +[Debian jessie-backports]:http://backports.debian.org/Instructions/ diff --git a/test/core/bad_client/bad_client.c b/test/core/bad_client/bad_client.c index 1d988796625..41ac83b7b70 100644 --- a/test/core/bad_client/bad_client.c +++ b/test/core/bad_client/bad_client.c @@ -145,6 +145,7 @@ void grpc_run_bad_client_test(grpc_bad_client_server_side_validator validator, gpr_event_wait(&a.done_write, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5))); if (flags & GRPC_BAD_CLIENT_DISCONNECT) { + grpc_endpoint_shutdown(sfd.client); grpc_endpoint_destroy(sfd.client); sfd.client = NULL; } @@ -153,6 +154,7 @@ void grpc_run_bad_client_test(grpc_bad_client_server_side_validator validator, /* Shutdown */ if (sfd.client) { + grpc_endpoint_shutdown(sfd.client); grpc_endpoint_destroy(sfd.client); } grpc_server_shutdown_and_notify(a.server, a.cq, NULL);