mirror of https://github.com/grpc/grpc.git
Muxi Yan
6 years ago
commit
e8eac7c107
37 changed files with 651 additions and 388 deletions
@ -0,0 +1,134 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <map> |
||||
|
||||
#include <grpcpp/support/client_interceptor.h> |
||||
|
||||
#ifdef BAZEL_BUILD |
||||
#include "examples/protos/keyvaluestore.grpc.pb.h" |
||||
#else |
||||
#include "keyvaluestore.grpc.pb.h" |
||||
#endif |
||||
|
||||
// This is a naive implementation of a cache. A new cache is for each call. For
|
||||
// each new key request, the key is first searched in the map and if found, the
|
||||
// interceptor fills in the return value without making a request to the server.
|
||||
// Only if the key is not found in the cache do we make a request.
|
||||
class CachingInterceptor : public grpc::experimental::Interceptor { |
||||
public: |
||||
CachingInterceptor(grpc::experimental::ClientRpcInfo* info) {} |
||||
|
||||
void Intercept( |
||||
::grpc::experimental::InterceptorBatchMethods* methods) override { |
||||
bool hijack = false; |
||||
if (methods->QueryInterceptionHookPoint( |
||||
grpc::experimental::InterceptionHookPoints:: |
||||
PRE_SEND_INITIAL_METADATA)) { |
||||
// Hijack all calls
|
||||
hijack = true; |
||||
// Create a stream on which this interceptor can make requests
|
||||
stub_ = keyvaluestore::KeyValueStore::NewStub( |
||||
methods->GetInterceptedChannel()); |
||||
stream_ = stub_->GetValues(&context_); |
||||
} |
||||
if (methods->QueryInterceptionHookPoint( |
||||
grpc::experimental::InterceptionHookPoints::PRE_SEND_MESSAGE)) { |
||||
// We know that clients perform a Read and a Write in a loop, so we don't
|
||||
// need to maintain a list of the responses.
|
||||
std::string requested_key; |
||||
const keyvaluestore::Request* req_msg = |
||||
static_cast<const keyvaluestore::Request*>(methods->GetSendMessage()); |
||||
if (req_msg != nullptr) { |
||||
requested_key = req_msg->key(); |
||||
} else { |
||||
// The non-serialized form would not be available in certain scenarios,
|
||||
// so add a fallback
|
||||
keyvaluestore::Request req_msg; |
||||
auto* buffer = methods->GetSerializedSendMessage(); |
||||
auto copied_buffer = *buffer; |
||||
GPR_ASSERT( |
||||
grpc::SerializationTraits<keyvaluestore::Request>::Deserialize( |
||||
&copied_buffer, &req_msg) |
||||
.ok()); |
||||
requested_key = req_msg.key(); |
||||
} |
||||
|
||||
// Check if the key is present in the map
|
||||
auto search = cached_map_.find(requested_key); |
||||
if (search != cached_map_.end()) { |
||||
std::cout << "Key " << requested_key << "found in map"; |
||||
response_ = search->second; |
||||
} else { |
||||
std::cout << "Key " << requested_key << "not found in cache"; |
||||
// Key was not found in the cache, so make a request
|
||||
keyvaluestore::Request req; |
||||
req.set_key(requested_key); |
||||
stream_->Write(req); |
||||
keyvaluestore::Response resp; |
||||
stream_->Read(&resp); |
||||
response_ = resp.value(); |
||||
// Insert the pair in the cache for future requests
|
||||
cached_map_.insert({requested_key, response_}); |
||||
} |
||||
} |
||||
if (methods->QueryInterceptionHookPoint( |
||||
grpc::experimental::InterceptionHookPoints::PRE_SEND_CLOSE)) { |
||||
stream_->WritesDone(); |
||||
} |
||||
if (methods->QueryInterceptionHookPoint( |
||||
grpc::experimental::InterceptionHookPoints::PRE_RECV_MESSAGE)) { |
||||
keyvaluestore::Response* resp = |
||||
static_cast<keyvaluestore::Response*>(methods->GetRecvMessage()); |
||||
resp->set_value(response_); |
||||
} |
||||
if (methods->QueryInterceptionHookPoint( |
||||
grpc::experimental::InterceptionHookPoints::PRE_RECV_STATUS)) { |
||||
auto* status = methods->GetRecvStatus(); |
||||
*status = grpc::Status::OK; |
||||
} |
||||
// One of Hijack or Proceed always needs to be called to make progress.
|
||||
if (hijack) { |
||||
// Hijack is called only once when PRE_SEND_INITIAL_METADATA is present in
|
||||
// the hook points
|
||||
methods->Hijack(); |
||||
} else { |
||||
// Proceed is an indicator that the interceptor is done intercepting the
|
||||
// batch.
|
||||
methods->Proceed(); |
||||
} |
||||
} |
||||
|
||||
private: |
||||
grpc::ClientContext context_; |
||||
std::unique_ptr<keyvaluestore::KeyValueStore::Stub> stub_; |
||||
std::unique_ptr< |
||||
grpc::ClientReaderWriter<keyvaluestore::Request, keyvaluestore::Response>> |
||||
stream_; |
||||
std::map<std::string, std::string> cached_map_; |
||||
std::string response_; |
||||
}; |
||||
|
||||
class CachingInterceptorFactory |
||||
: public grpc::experimental::ClientInterceptorFactoryInterface { |
||||
public: |
||||
grpc::experimental::Interceptor* CreateClientInterceptor( |
||||
grpc::experimental::ClientRpcInfo* info) override { |
||||
return new CachingInterceptor(info); |
||||
} |
||||
}; |
||||
@ -0,0 +1,110 @@ |
||||
#region Copyright notice and license |
||||
|
||||
// Copyright 2019 The gRPC Authors |
||||
// |
||||
// Licensed under the Apache License, Version 2.0 (the "License"); |
||||
// you may not use this file except in compliance with the License. |
||||
// You may obtain a copy of the License at |
||||
// |
||||
// http://www.apache.org/licenses/LICENSE-2.0 |
||||
// |
||||
// Unless required by applicable law or agreed to in writing, software |
||||
// distributed under the License is distributed on an "AS IS" BASIS, |
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
// See the License for the specific language governing permissions and |
||||
// limitations under the License. |
||||
|
||||
#endregion |
||||
|
||||
using System; |
||||
using System.Threading; |
||||
using System.Threading.Tasks; |
||||
|
||||
using Grpc.Core.Internal; |
||||
|
||||
namespace Grpc.Core |
||||
{ |
||||
/// <summary> |
||||
/// Default implementation of <c>ServerCallContext</c>. |
||||
/// </summary> |
||||
internal class DefaultServerCallContext : ServerCallContext |
||||
{ |
||||
private readonly CallSafeHandle callHandle; |
||||
private readonly string method; |
||||
private readonly string host; |
||||
private readonly DateTime deadline; |
||||
private readonly Metadata requestHeaders; |
||||
private readonly CancellationToken cancellationToken; |
||||
private readonly Metadata responseTrailers; |
||||
private Status status; |
||||
private readonly IServerResponseStream serverResponseStream; |
||||
private readonly Lazy<AuthContext> authContext; |
||||
|
||||
/// <summary> |
||||
/// Creates a new instance of <c>ServerCallContext</c>. |
||||
/// To allow reuse of ServerCallContext API by different gRPC implementations, the implementation of some members is provided externally. |
||||
/// To provide state, this <c>ServerCallContext</c> instance and <c>extraData</c> will be passed to the member implementations. |
||||
/// </summary> |
||||
internal DefaultServerCallContext(CallSafeHandle callHandle, string method, string host, DateTime deadline, |
||||
Metadata requestHeaders, CancellationToken cancellationToken, IServerResponseStream serverResponseStream) |
||||
{ |
||||
this.callHandle = callHandle; |
||||
this.method = method; |
||||
this.host = host; |
||||
this.deadline = deadline; |
||||
this.requestHeaders = requestHeaders; |
||||
this.cancellationToken = cancellationToken; |
||||
this.responseTrailers = new Metadata(); |
||||
this.status = Status.DefaultSuccess; |
||||
this.serverResponseStream = serverResponseStream; |
||||
// TODO(jtattermusch): avoid unnecessary allocation of factory function and the lazy object |
||||
this.authContext = new Lazy<AuthContext>(GetAuthContextEager); |
||||
} |
||||
|
||||
protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions options) |
||||
{ |
||||
return new ContextPropagationToken(callHandle, deadline, cancellationToken, options); |
||||
} |
||||
|
||||
protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) |
||||
{ |
||||
return serverResponseStream.WriteResponseHeadersAsync(responseHeaders); |
||||
} |
||||
|
||||
protected override string MethodCore => method; |
||||
|
||||
protected override string HostCore => host; |
||||
|
||||
protected override string PeerCore => callHandle.GetPeer(); |
||||
|
||||
protected override DateTime DeadlineCore => deadline; |
||||
|
||||
protected override Metadata RequestHeadersCore => requestHeaders; |
||||
|
||||
protected override CancellationToken CancellationTokenCore => cancellationToken; |
||||
|
||||
protected override Metadata ResponseTrailersCore => responseTrailers; |
||||
|
||||
protected override Status StatusCore |
||||
{ |
||||
get => status; |
||||
set => status = value; |
||||
} |
||||
|
||||
protected override WriteOptions WriteOptionsCore |
||||
{ |
||||
get => serverResponseStream.WriteOptions; |
||||
set => serverResponseStream.WriteOptions = value; |
||||
} |
||||
|
||||
protected override AuthContext AuthContextCore => authContext.Value; |
||||
|
||||
private AuthContext GetAuthContextEager() |
||||
{ |
||||
using (var authContextNative = callHandle.GetAuthContext()) |
||||
{ |
||||
return authContextNative.ToAuthContext(); |
||||
} |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,38 @@ |
||||
#region Copyright notice and license |
||||
// Copyright 2019 The gRPC Authors |
||||
// |
||||
// Licensed under the Apache License, Version 2.0 (the "License"); |
||||
// you may not use this file except in compliance with the License. |
||||
// You may obtain a copy of the License at |
||||
// |
||||
// http://www.apache.org/licenses/LICENSE-2.0 |
||||
// |
||||
// Unless required by applicable law or agreed to in writing, software |
||||
// distributed under the License is distributed on an "AS IS" BASIS, |
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
// See the License for the specific language governing permissions and |
||||
// limitations under the License. |
||||
#endregion |
||||
|
||||
using System; |
||||
using System.Threading.Tasks; |
||||
using Grpc.Core.Internal; |
||||
|
||||
namespace Grpc.Core.Internal |
||||
{ |
||||
/// <summary> |
||||
/// Exposes non-generic members of <c>ServerReponseStream</c>. |
||||
/// </summary> |
||||
internal interface IServerResponseStream |
||||
{ |
||||
/// <summary> |
||||
/// Asynchronously sends response headers for the current call to the client. See <c>ServerCallContext.WriteResponseHeadersAsync</c> for exact semantics. |
||||
/// </summary> |
||||
Task WriteResponseHeadersAsync(Metadata responseHeaders); |
||||
|
||||
/// <summary> |
||||
/// Gets or sets the write options. |
||||
/// </summary> |
||||
WriteOptions WriteOptions { get; set; } |
||||
} |
||||
} |
||||
@ -0,0 +1,10 @@ |
||||
diff --git a/tools/dockerfile/interoptest/grpc_interop_ruby/build_interop.sh b/tools/dockerfile/interoptest/grpc_interop_ruby/build_interop.sh
|
||||
index 67f66090ae..e71ad91499 100755
|
||||
--- a/tools/dockerfile/interoptest/grpc_interop_ruby/build_interop.sh
|
||||
+++ b/tools/dockerfile/interoptest/grpc_interop_ruby/build_interop.sh
|
||||
@@ -30,4 +30,4 @@ cd /var/local/git/grpc
|
||||
rvm --default use ruby-2.5
|
||||
|
||||
# build Ruby interop client and server
|
||||
-(cd src/ruby && gem update bundler && bundle && rake compile)
|
||||
+(cd src/ruby && gem install bundler -v 1.17.3 && bundle && rake compile)
|
||||
@ -1,37 +0,0 @@ |
||||
# Copyright 2017 gRPC authors. |
||||
# |
||||
# Licensed under the Apache License, Version 2.0 (the "License"); |
||||
# you may not use this file except in compliance with the License. |
||||
# You may obtain a copy of the License at |
||||
# |
||||
# http://www.apache.org/licenses/LICENSE-2.0 |
||||
# |
||||
# Unless required by applicable law or agreed to in writing, software |
||||
# distributed under the License is distributed on an "AS IS" BASIS, |
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
# See the License for the specific language governing permissions and |
||||
# limitations under the License. |
||||
|
||||
import os |
||||
import json |
||||
import urllib2 |
||||
|
||||
|
||||
def comment_on_pr(text): |
||||
if 'JENKINS_OAUTH_TOKEN' not in os.environ: |
||||
print 'Missing JENKINS_OAUTH_TOKEN env var: not commenting' |
||||
return |
||||
if 'ghprbPullId' not in os.environ: |
||||
print 'Missing ghprbPullId env var: not commenting' |
||||
return |
||||
req = urllib2.Request( |
||||
url='https://api.github.com/repos/grpc/grpc/issues/%s/comments' % |
||||
os.environ['ghprbPullId'], |
||||
data=json.dumps({ |
||||
'body': text |
||||
}), |
||||
headers={ |
||||
'Authorization': 'token %s' % os.environ['JENKINS_OAUTH_TOKEN'], |
||||
'Content-Type': 'application/json', |
||||
}) |
||||
print urllib2.urlopen(req).read() |
||||
Loading…
Reference in new issue