| OLD | NEW |
| 1 // Copyright (c) 2011 The Chromium Authors. All rights reserved. | 1 // Copyright (c) 2011 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 "net/curvecp/messenger.h" | 5 #include "net/curvecp/messenger.h" |
| 6 | 6 |
| 7 #include "base/logging.h" | 7 #include "base/logging.h" |
| 8 #include "base/message_loop.h" | 8 #include "base/message_loop.h" |
| 9 #include "net/base/io_buffer.h" | 9 #include "net/base/io_buffer.h" |
| 10 #include "net/base/net_errors.h" | 10 #include "net/base/net_errors.h" |
| (...skipping 41 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 52 | 52 |
| 53 // Size of the send buffer. | 53 // Size of the send buffer. |
| 54 static const size_t kSendBufferSize = (128 * 1024); | 54 static const size_t kSendBufferSize = (128 * 1024); |
| 55 // Size of the receive buffer. | 55 // Size of the receive buffer. |
| 56 static const size_t kReceiveBufferSize = (128 * 1024); | 56 static const size_t kReceiveBufferSize = (128 * 1024); |
| 57 | 57 |
| 58 Messenger::Messenger(Packetizer* packetizer) | 58 Messenger::Messenger(Packetizer* packetizer) |
| 59 : packetizer_(packetizer), | 59 : packetizer_(packetizer), |
| 60 send_buffer_(kSendBufferSize), | 60 send_buffer_(kSendBufferSize), |
| 61 send_complete_callback_(NULL), | 61 send_complete_callback_(NULL), |
| 62 receive_complete_callback_(NULL), | 62 old_receive_complete_callback_(NULL), |
| 63 pending_receive_length_(0), | 63 pending_receive_length_(0), |
| 64 send_message_in_progress_(false), | 64 send_message_in_progress_(false), |
| 65 ALLOW_THIS_IN_INITIALIZER_LIST( | 65 ALLOW_THIS_IN_INITIALIZER_LIST( |
| 66 send_message_callback_(this, &Messenger::OnSendMessageComplete)), | 66 send_message_callback_(this, &Messenger::OnSendMessageComplete)), |
| 67 ALLOW_THIS_IN_INITIALIZER_LIST(factory_(this)) { | 67 ALLOW_THIS_IN_INITIALIZER_LIST(factory_(this)) { |
| 68 } | 68 } |
| 69 | 69 |
| 70 Messenger::~Messenger() { | 70 Messenger::~Messenger() { |
| 71 } | 71 } |
| 72 | 72 |
| 73 int Messenger::Read(IOBuffer* buf, int buf_len, OldCompletionCallback* callback)
{ | 73 int Messenger::Read(IOBuffer* buf, int buf_len, OldCompletionCallback* callback)
{ |
| 74 DCHECK(CalledOnValidThread()); | 74 DCHECK(CalledOnValidThread()); |
| 75 DCHECK(!receive_complete_callback_); | 75 DCHECK(!old_receive_complete_callback_ && |
| 76 receive_complete_callback_.is_null()); |
| 77 |
| 78 if (!received_list_.bytes_available()) { |
| 79 old_receive_complete_callback_ = callback; |
| 80 pending_receive_ = buf; |
| 81 pending_receive_length_ = buf_len; |
| 82 return ERR_IO_PENDING; |
| 83 } |
| 84 |
| 85 int bytes_read = InternalRead(buf, buf_len); |
| 86 DCHECK_LT(0, bytes_read); |
| 87 return bytes_read; |
| 88 } |
| 89 int Messenger::Read(IOBuffer* buf, int buf_len, |
| 90 const CompletionCallback& callback) { |
| 91 DCHECK(CalledOnValidThread()); |
| 92 DCHECK(!old_receive_complete_callback_ && |
| 93 receive_complete_callback_.is_null()); |
| 76 | 94 |
| 77 if (!received_list_.bytes_available()) { | 95 if (!received_list_.bytes_available()) { |
| 78 receive_complete_callback_ = callback; | 96 receive_complete_callback_ = callback; |
| 79 pending_receive_ = buf; | 97 pending_receive_ = buf; |
| 80 pending_receive_length_ = buf_len; | 98 pending_receive_length_ = buf_len; |
| 81 return ERR_IO_PENDING; | 99 return ERR_IO_PENDING; |
| 82 } | 100 } |
| 83 | 101 |
| 84 int bytes_read = InternalRead(buf, buf_len); | 102 int bytes_read = InternalRead(buf, buf_len); |
| 85 DCHECK_LT(0, bytes_read); | 103 DCHECK_LT(0, bytes_read); |
| (...skipping 245 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 331 // Add to our received block list | 349 // Add to our received block list |
| 332 uint64 position = uint64_unpack(header->position); | 350 uint64 position = uint64_unpack(header->position); |
| 333 scoped_refptr<IOBuffer> buffer(new IOBuffer(body_length)); | 351 scoped_refptr<IOBuffer> buffer(new IOBuffer(body_length)); |
| 334 memcpy(buffer->data(), message->data() + sizeof(Message), body_length); | 352 memcpy(buffer->data(), message->data() + sizeof(Message), body_length); |
| 335 received_list_.AddBlock(position, buffer, body_length); | 353 received_list_.AddBlock(position, buffer, body_length); |
| 336 | 354 |
| 337 SendAck(message_id); | 355 SendAck(message_id); |
| 338 } | 356 } |
| 339 | 357 |
| 340 // If we have data available, and a read is pending, notify the callback. | 358 // If we have data available, and a read is pending, notify the callback. |
| 341 if (received_list_.bytes_available() && receive_complete_callback_) { | 359 if (received_list_.bytes_available() && |
| 360 (old_receive_complete_callback_ || |
| 361 !receive_complete_callback_.is_null())) { |
| 342 // Pass the data up to the caller. | 362 // Pass the data up to the caller. |
| 343 int bytes_read = InternalRead(pending_receive_, pending_receive_length_); | 363 int bytes_read = InternalRead(pending_receive_, pending_receive_length_); |
| 344 OldCompletionCallback* callback = receive_complete_callback_; | 364 if (old_receive_complete_callback_) { |
| 345 receive_complete_callback_ = NULL; | 365 OldCompletionCallback* callback = old_receive_complete_callback_; |
| 346 callback->Run(bytes_read); | 366 old_receive_complete_callback_ = NULL; |
| 367 callback->Run(bytes_read); |
| 368 } else { |
| 369 CompletionCallback callback = receive_complete_callback_; |
| 370 receive_complete_callback_.Reset(); |
| 371 callback.Run(bytes_read); |
| 372 } |
| 347 } | 373 } |
| 348 } | 374 } |
| 349 | 375 |
| 350 void Messenger::SendAck(uint32 last_message_received) { | 376 void Messenger::SendAck(uint32 last_message_received) { |
| 351 scoped_refptr<IOBuffer> buffer(new IOBuffer(sizeof(Message))); | 377 scoped_refptr<IOBuffer> buffer(new IOBuffer(sizeof(Message))); |
| 352 memset(buffer->data(), 0, sizeof(Message)); | 378 memset(buffer->data(), 0, sizeof(Message)); |
| 353 Message* message = reinterpret_cast<Message*>(buffer->data()); | 379 Message* message = reinterpret_cast<Message*>(buffer->data()); |
| 354 uint32_pack(message->last_message_received, last_message_received); | 380 uint32_pack(message->last_message_received, last_message_received); |
| 355 uint64_pack(message->acknowledge_1, received_list_.bytes_received()); | 381 uint64_pack(message->acknowledge_1, received_list_.bytes_received()); |
| 356 LOG(ERROR) << "SendAck " << received_list_.bytes_received(); | 382 LOG(ERROR) << "SendAck " << received_list_.bytes_received(); |
| 357 // TODO(mbelshe): fill in remainder of selective acknowledgements | 383 // TODO(mbelshe): fill in remainder of selective acknowledgements |
| 358 | 384 |
| 359 // TODO(mbelshe): Fix this - it is totally possible to have a send message | 385 // TODO(mbelshe): Fix this - it is totally possible to have a send message |
| 360 // in progress here... | 386 // in progress here... |
| 361 DCHECK(!send_message_in_progress_); | 387 DCHECK(!send_message_in_progress_); |
| 362 | 388 |
| 363 int rv = packetizer_->SendMessage(key_, | 389 int rv = packetizer_->SendMessage(key_, |
| 364 buffer->data(), | 390 buffer->data(), |
| 365 sizeof(Message), | 391 sizeof(Message), |
| 366 &send_message_callback_); | 392 &send_message_callback_); |
| 367 // TODO(mbelshe): Fix me! Deal with the error cases | 393 // TODO(mbelshe): Fix me! Deal with the error cases |
| 368 DCHECK(rv == sizeof(Message)); | 394 DCHECK(rv == sizeof(Message)); |
| 369 } | 395 } |
| 370 | 396 |
| 371 } // namespace net | 397 } // namespace net |
| OLD | NEW |