| 187 | google::protobuf::Message*>::value, |
| 188 | int>::type = 0> |
| 189 | Future<Try<Response, StatusError>> call( |
| 190 | const Connection& connection, |
| 191 | Method&& method, |
| 192 | Request&& request, |
| 193 | const CallOptions& options) |
| 194 | { |
| 195 | // Create a `Promise` that will be set upon receiving a response. |
| 196 | // TODO(chhsiao): The `Promise` in the `shared_ptr` is not shared, but only |
| 197 | // to be captured by the lambda below. Use a `unique_ptr` once we get C++14. |
| 198 | std::shared_ptr<Promise<Try<Response, StatusError>>> promise( |
| 199 | new Promise<Try<Response, StatusError>>); |
| 200 | Future<Try<Response, StatusError>> future = promise->future(); |
| 201 | |
| 202 | // Send the request in the internal runtime process. |
| 203 | // TODO(chhsiao): We use `std::bind` here to forward `request` to avoid an |
| 204 | // extra copy. We should capture it by forwarding once we get C++14. |
| 205 | dispatch(data->pid, &RuntimeProcess::send, std::bind( |
| 206 | [connection, method, options, promise]( |
| 207 | const Request& request, |
| 208 | bool terminating, |
| 209 | ::grpc::CompletionQueue* queue) { |
| 210 | if (terminating) { |
| 211 | promise->fail("Runtime has been terminated"); |
| 212 | return; |
| 213 | } |
| 214 | |
| 215 | // TODO(chhsiao): The `shared_ptr`s here aren't shared, but only to be |
| 216 | // captured by the lambda below. Use `unique_ptr`s once we get C++14. |
| 217 | std::shared_ptr<::grpc::ClientContext> context( |
| 218 | new ::grpc::ClientContext()); |
| 219 | |
| 220 | context->set_wait_for_ready(options.wait_for_ready); |
| 221 | |
| 222 | // We need to ensure that we're using a |
| 223 | // `std::chrono::system_clock::time_point` because `grpc::TimePoint` |
| 224 | // provides a specialization only for this type and we cannot |
| 225 | // guarantee that the operation below will always result in this type. |
| 226 | auto time_point = |
| 227 | std::chrono::time_point_cast<std::chrono::system_clock::duration>( |
| 228 | std::chrono::system_clock::now() + |
| 229 | std::chrono::nanoseconds(options.timeout.ns())); |
| 230 | |
| 231 | context->set_deadline(time_point); |
| 232 | |
| 233 | promise->future().onDiscard([=] { context->TryCancel(); }); |
| 234 | |
| 235 | std::shared_ptr<Response> response(new Response()); |
| 236 | std::shared_ptr<::grpc::Status> status(new ::grpc::Status()); |
| 237 | |
| 238 | std::shared_ptr<::grpc::ClientAsyncResponseReader<Response>> reader = |
| 239 | (typename internal::MethodTraits<Method>::stub_type( |
| 240 | connection.channel).*method)(context.get(), request, queue); |
| 241 | |
| 242 | reader->StartCall(); |
| 243 | |
| 244 | // Create a `ReceiveCallback` as a tag in the `CompletionQueue` for |
| 245 | // the current asynchronous gRPC call. The callback will set up the |
| 246 | // above `Promise` upon receiving a response. |