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 40 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
51 static const size_t kMaxWriteQueueMessages = 128; | 51 static const size_t kMaxWriteQueueMessages = 128; |
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 old_send_complete_callback_(NULL), | 61 send_complete_callback_(NULL), |
62 old_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 } |
(...skipping 25 matching lines...) Expand all Loading... |
97 pending_receive_ = buf; | 97 pending_receive_ = buf; |
98 pending_receive_length_ = buf_len; | 98 pending_receive_length_ = buf_len; |
99 return ERR_IO_PENDING; | 99 return ERR_IO_PENDING; |
100 } | 100 } |
101 | 101 |
102 int bytes_read = InternalRead(buf, buf_len); | 102 int bytes_read = InternalRead(buf, buf_len); |
103 DCHECK_LT(0, bytes_read); | 103 DCHECK_LT(0, bytes_read); |
104 return bytes_read; | 104 return bytes_read; |
105 } | 105 } |
106 | 106 |
107 int Messenger::Write( | 107 int Messenger::Write(IOBuffer* buf, int buf_len, OldCompletionCallback* callback
) { |
108 IOBuffer* buf, int buf_len, OldCompletionCallback* callback) { | |
109 DCHECK(CalledOnValidThread()); | 108 DCHECK(CalledOnValidThread()); |
110 DCHECK(!pending_send_.get()); // Already a write pending! | 109 DCHECK(!pending_send_.get()); // Already a write pending! |
111 DCHECK(!old_send_complete_callback_ && send_complete_callback_.is_null()); | 110 DCHECK(!send_complete_callback_); |
112 DCHECK_LT(0, buf_len); | 111 DCHECK_LT(0, buf_len); |
113 | 112 |
114 int len = send_buffer_.write(buf->data(), buf_len); | 113 int len = send_buffer_.write(buf->data(), buf_len); |
115 if (!send_timer_.IsRunning()) | 114 if (!send_timer_.IsRunning()) |
116 send_timer_.Start(FROM_HERE, base::TimeDelta(), | 115 send_timer_.Start(FROM_HERE, base::TimeDelta(), |
117 this, &Messenger::OnSendTimer); | 116 this, &Messenger::OnSendTimer); |
118 if (len) | 117 if (len) |
119 return len; | 118 return len; |
120 | 119 |
121 // We couldn't add data to the send buffer, so block the application. | 120 // We couldn't add data to the send buffer, so block the application. |
122 pending_send_ = buf; | 121 pending_send_ = buf; |
123 pending_send_length_ = buf_len; | 122 pending_send_length_ = buf_len; |
124 old_send_complete_callback_ = callback; | |
125 return ERR_IO_PENDING; | |
126 } | |
127 int Messenger::Write( | |
128 IOBuffer* buf, int buf_len, const CompletionCallback& callback) { | |
129 DCHECK(CalledOnValidThread()); | |
130 DCHECK(!pending_send_.get()); // Already a write pending! | |
131 DCHECK(!old_send_complete_callback_ && send_complete_callback_.is_null()); | |
132 DCHECK_LT(0, buf_len); | |
133 | |
134 int len = send_buffer_.write(buf->data(), buf_len); | |
135 if (!send_timer_.IsRunning()) | |
136 send_timer_.Start(FROM_HERE, base::TimeDelta(), | |
137 this, &Messenger::OnSendTimer); | |
138 if (len) | |
139 return len; | |
140 | |
141 // We couldn't add data to the send buffer, so block the application. | |
142 pending_send_ = buf; | |
143 pending_send_length_ = buf_len; | |
144 send_complete_callback_ = callback; | 123 send_complete_callback_ = callback; |
145 return ERR_IO_PENDING; | 124 return ERR_IO_PENDING; |
146 } | 125 } |
147 | 126 |
148 void Messenger::OnConnection(ConnectionKey key) { | 127 void Messenger::OnConnection(ConnectionKey key) { |
149 LOG(ERROR) << "Client Connect: " << key.ToString(); | 128 LOG(ERROR) << "Client Connect: " << key.ToString(); |
150 key_ = key; | 129 key_ = key; |
151 } | 130 } |
152 | 131 |
153 void Messenger::OnClose(Packetizer* packetizer, ConnectionKey key) { | 132 void Messenger::OnClose(Packetizer* packetizer, ConnectionKey key) { |
(...skipping 28 matching lines...) Expand all Loading... |
182 IOBufferWithSize* Messenger::CreateBufferFromSendQueue() { | 161 IOBufferWithSize* Messenger::CreateBufferFromSendQueue() { |
183 DCHECK_LT(0, send_buffer_.length()); | 162 DCHECK_LT(0, send_buffer_.length()); |
184 | 163 |
185 int length = std::min(packetizer_->max_message_payload(), | 164 int length = std::min(packetizer_->max_message_payload(), |
186 send_buffer_.length()); | 165 send_buffer_.length()); |
187 IOBufferWithSize* rv = new IOBufferWithSize(length); | 166 IOBufferWithSize* rv = new IOBufferWithSize(length); |
188 int bytes = send_buffer_.read(rv->data(), length); | 167 int bytes = send_buffer_.read(rv->data(), length); |
189 DCHECK_EQ(bytes, length); | 168 DCHECK_EQ(bytes, length); |
190 | 169 |
191 // We consumed data, check to see if someone is waiting to write more data. | 170 // We consumed data, check to see if someone is waiting to write more data. |
192 if (old_send_complete_callback_ || !send_complete_callback_.is_null()) { | 171 if (send_complete_callback_) { |
193 DCHECK(pending_send_.get()); | 172 DCHECK(pending_send_.get()); |
194 | 173 |
195 int len = send_buffer_.write(pending_send_->data(), pending_send_length_); | 174 int len = send_buffer_.write(pending_send_->data(), pending_send_length_); |
196 if (len) { | 175 if (len) { |
197 pending_send_ = NULL; | 176 pending_send_ = NULL; |
198 if (old_send_complete_callback_) { | 177 OldCompletionCallback* callback = send_complete_callback_; |
199 OldCompletionCallback* callback = old_send_complete_callback_; | 178 send_complete_callback_ = NULL; |
200 old_send_complete_callback_ = NULL; | 179 callback->Run(len); |
201 callback->Run(len); | |
202 } else { | |
203 CompletionCallback callback = send_complete_callback_; | |
204 send_complete_callback_.Reset(); | |
205 callback.Run(len); | |
206 } | |
207 } | 180 } |
208 } | 181 } |
209 | 182 |
210 return rv; | 183 return rv; |
211 } | 184 } |
212 | 185 |
213 void Messenger::OnSendMessageComplete(int result) { | 186 void Messenger::OnSendMessageComplete(int result) { |
214 DCHECK_NE(ERR_IO_PENDING, result); | 187 DCHECK_NE(ERR_IO_PENDING, result); |
215 | 188 |
216 send_message_in_progress_ = false; | 189 send_message_in_progress_ = false; |
(...skipping 198 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
415 | 388 |
416 int rv = packetizer_->SendMessage(key_, | 389 int rv = packetizer_->SendMessage(key_, |
417 buffer->data(), | 390 buffer->data(), |
418 sizeof(Message), | 391 sizeof(Message), |
419 &send_message_callback_); | 392 &send_message_callback_); |
420 // TODO(mbelshe): Fix me! Deal with the error cases | 393 // TODO(mbelshe): Fix me! Deal with the error cases |
421 DCHECK(rv == sizeof(Message)); | 394 DCHECK(rv == sizeof(Message)); |
422 } | 395 } |
423 | 396 |
424 } // namespace net | 397 } // namespace net |
OLD | NEW |