OLD | NEW |
1 // Copyright 2015 The Chromium Authors. All rights reserved. | 1 // Copyright 2015 The Chromium Authors. All rights reserved. |
2 // Use of this source code is governed by a BSD-style license that can be | 2 // Use of this source code is governed by a BSD-style license that can be |
3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
4 | 4 |
5 #include "mojo/public/cpp/bindings/interface_endpoint_client.h" | 5 #include "mojo/public/cpp/bindings/interface_endpoint_client.h" |
6 | 6 |
7 #include <stdint.h> | 7 #include <stdint.h> |
8 | 8 |
9 #include <utility> | 9 #include <utility> |
10 | 10 |
(...skipping 216 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
227 message->SerializeAssociatedEndpointHandles(handle_.group_controller()); | 227 message->SerializeAssociatedEndpointHandles(handle_.group_controller()); |
228 | 228 |
229 if (encountered_error_) | 229 if (encountered_error_) |
230 return false; | 230 return false; |
231 | 231 |
232 InitControllerIfNecessary(); | 232 InitControllerIfNecessary(); |
233 | 233 |
234 return controller_->SendMessage(message); | 234 return controller_->SendMessage(message); |
235 } | 235 } |
236 | 236 |
237 bool InterfaceEndpointClient::AcceptWithResponder(Message* message, | 237 bool InterfaceEndpointClient::AcceptWithResponder( |
238 MessageReceiver* responder) { | 238 Message* message, |
| 239 std::unique_ptr<MessageReceiver> responder) { |
239 DCHECK(thread_checker_.CalledOnValidThread()); | 240 DCHECK(thread_checker_.CalledOnValidThread()); |
240 DCHECK(message->has_flag(Message::kFlagExpectsResponse)); | 241 DCHECK(message->has_flag(Message::kFlagExpectsResponse)); |
241 DCHECK(!handle_.pending_association()); | 242 DCHECK(!handle_.pending_association()); |
242 | 243 |
243 // Please see comments in Accept(). | 244 // Please see comments in Accept(). |
244 if (!message->associated_endpoint_handles()->empty()) | 245 if (!message->associated_endpoint_handles()->empty()) |
245 message->SerializeAssociatedEndpointHandles(handle_.group_controller()); | 246 message->SerializeAssociatedEndpointHandles(handle_.group_controller()); |
246 | 247 |
247 if (encountered_error_) | 248 if (encountered_error_) |
248 return false; | 249 return false; |
249 | 250 |
250 InitControllerIfNecessary(); | 251 InitControllerIfNecessary(); |
251 | 252 |
252 // Reserve 0 in case we want it to convey special meaning in the future. | 253 // Reserve 0 in case we want it to convey special meaning in the future. |
253 uint64_t request_id = next_request_id_++; | 254 uint64_t request_id = next_request_id_++; |
254 if (request_id == 0) | 255 if (request_id == 0) |
255 request_id = next_request_id_++; | 256 request_id = next_request_id_++; |
256 | 257 |
257 message->set_request_id(request_id); | 258 message->set_request_id(request_id); |
258 | 259 |
259 bool is_sync = message->has_flag(Message::kFlagIsSync); | 260 bool is_sync = message->has_flag(Message::kFlagIsSync); |
260 if (!controller_->SendMessage(message)) | 261 if (!controller_->SendMessage(message)) |
261 return false; | 262 return false; |
262 | 263 |
263 if (!is_sync) { | 264 if (!is_sync) { |
264 // We assume ownership of |responder|. | 265 async_responders_[request_id] = std::move(responder); |
265 async_responders_[request_id] = base::WrapUnique(responder); | |
266 return true; | 266 return true; |
267 } | 267 } |
268 | 268 |
269 SyncCallRestrictions::AssertSyncCallAllowed(); | 269 SyncCallRestrictions::AssertSyncCallAllowed(); |
270 | 270 |
271 bool response_received = false; | 271 bool response_received = false; |
272 std::unique_ptr<MessageReceiver> sync_responder(responder); | |
273 sync_responses_.insert(std::make_pair( | 272 sync_responses_.insert(std::make_pair( |
274 request_id, base::MakeUnique<SyncResponseInfo>(&response_received))); | 273 request_id, base::MakeUnique<SyncResponseInfo>(&response_received))); |
275 | 274 |
276 base::WeakPtr<InterfaceEndpointClient> weak_self = | 275 base::WeakPtr<InterfaceEndpointClient> weak_self = |
277 weak_ptr_factory_.GetWeakPtr(); | 276 weak_ptr_factory_.GetWeakPtr(); |
278 controller_->SyncWatch(&response_received); | 277 controller_->SyncWatch(&response_received); |
279 // Make sure that this instance hasn't been destroyed. | 278 // Make sure that this instance hasn't been destroyed. |
280 if (weak_self) { | 279 if (weak_self) { |
281 DCHECK(base::ContainsKey(sync_responses_, request_id)); | 280 DCHECK(base::ContainsKey(sync_responses_, request_id)); |
282 auto iter = sync_responses_.find(request_id); | 281 auto iter = sync_responses_.find(request_id); |
283 DCHECK_EQ(&response_received, iter->second->response_received); | 282 DCHECK_EQ(&response_received, iter->second->response_received); |
284 if (response_received) | 283 if (response_received) |
285 ignore_result(sync_responder->Accept(&iter->second->response)); | 284 ignore_result(responder->Accept(&iter->second->response)); |
286 sync_responses_.erase(iter); | 285 sync_responses_.erase(iter); |
287 } | 286 } |
288 | 287 |
289 // Return true means that we take ownership of |responder|. | |
290 return true; | 288 return true; |
291 } | 289 } |
292 | 290 |
293 bool InterfaceEndpointClient::HandleIncomingMessage(Message* message) { | 291 bool InterfaceEndpointClient::HandleIncomingMessage(Message* message) { |
294 DCHECK(thread_checker_.CalledOnValidThread()); | 292 DCHECK(thread_checker_.CalledOnValidThread()); |
295 return filters_.Accept(message); | 293 return filters_.Accept(message); |
296 } | 294 } |
297 | 295 |
298 void InterfaceEndpointClient::NotifyError( | 296 void InterfaceEndpointClient::NotifyError( |
299 const base::Optional<DisconnectReason>& reason) { | 297 const base::Optional<DisconnectReason>& reason) { |
(...skipping 68 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
368 // This message is received after error has been encountered. For associated | 366 // This message is received after error has been encountered. For associated |
369 // interfaces, this means the remote side sends a | 367 // interfaces, this means the remote side sends a |
370 // PeerAssociatedEndpointClosed event but continues to send more messages | 368 // PeerAssociatedEndpointClosed event but continues to send more messages |
371 // for the same interface. Close the pipe because this shouldn't happen. | 369 // for the same interface. Close the pipe because this shouldn't happen. |
372 DVLOG(1) << "A message is received for an interface after it has been " | 370 DVLOG(1) << "A message is received for an interface after it has been " |
373 << "disconnected. Closing the pipe."; | 371 << "disconnected. Closing the pipe."; |
374 return false; | 372 return false; |
375 } | 373 } |
376 | 374 |
377 if (message->has_flag(Message::kFlagExpectsResponse)) { | 375 if (message->has_flag(Message::kFlagExpectsResponse)) { |
378 MessageReceiverWithStatus* responder = | 376 std::unique_ptr<MessageReceiverWithStatus> responder = |
379 new ResponderThunk(weak_ptr_factory_.GetWeakPtr(), task_runner_); | 377 base::MakeUnique<ResponderThunk>(weak_ptr_factory_.GetWeakPtr(), |
380 bool ok = false; | 378 task_runner_); |
381 if (mojo::internal::ControlMessageHandler::IsControlMessage(message)) { | 379 if (mojo::internal::ControlMessageHandler::IsControlMessage(message)) { |
382 ok = control_message_handler_.AcceptWithResponder(message, responder); | 380 return control_message_handler_.AcceptWithResponder(message, |
| 381 std::move(responder)); |
383 } else { | 382 } else { |
384 ok = incoming_receiver_->AcceptWithResponder(message, responder); | 383 return incoming_receiver_->AcceptWithResponder(message, |
| 384 std::move(responder)); |
385 } | 385 } |
386 if (!ok) | |
387 delete responder; | |
388 return ok; | |
389 } else if (message->has_flag(Message::kFlagIsResponse)) { | 386 } else if (message->has_flag(Message::kFlagIsResponse)) { |
390 uint64_t request_id = message->request_id(); | 387 uint64_t request_id = message->request_id(); |
391 | 388 |
392 if (message->has_flag(Message::kFlagIsSync)) { | 389 if (message->has_flag(Message::kFlagIsSync)) { |
393 auto it = sync_responses_.find(request_id); | 390 auto it = sync_responses_.find(request_id); |
394 if (it == sync_responses_.end()) | 391 if (it == sync_responses_.end()) |
395 return false; | 392 return false; |
396 it->second->response = std::move(*message); | 393 it->second->response = std::move(*message); |
397 *it->second->response_received = true; | 394 *it->second->response_received = true; |
398 return true; | 395 return true; |
399 } | 396 } |
400 | 397 |
401 auto it = async_responders_.find(request_id); | 398 auto it = async_responders_.find(request_id); |
402 if (it == async_responders_.end()) | 399 if (it == async_responders_.end()) |
403 return false; | 400 return false; |
404 std::unique_ptr<MessageReceiver> responder = std::move(it->second); | 401 std::unique_ptr<MessageReceiver> responder = std::move(it->second); |
405 async_responders_.erase(it); | 402 async_responders_.erase(it); |
406 return responder->Accept(message); | 403 return responder->Accept(message); |
407 } else { | 404 } else { |
408 if (mojo::internal::ControlMessageHandler::IsControlMessage(message)) | 405 if (mojo::internal::ControlMessageHandler::IsControlMessage(message)) |
409 return control_message_handler_.Accept(message); | 406 return control_message_handler_.Accept(message); |
410 | 407 |
411 return incoming_receiver_->Accept(message); | 408 return incoming_receiver_->Accept(message); |
412 } | 409 } |
413 } | 410 } |
414 | 411 |
415 } // namespace mojo | 412 } // namespace mojo |
OLD | NEW |