OLD | NEW |
---|---|
(Empty) | |
1 // Copyright 2014 The Chromium Authors. All rights reserved. | |
2 // Use of this source code is governed by a BSD-style license that can be | |
3 // found in the LICENSE file. | |
4 | |
5 #include "extensions/browser/api/cast_channel/cast_transport.h" | |
6 | |
7 #include <string> | |
8 | |
9 #include "base/bind.h" | |
10 #include "base/format_macros.h" | |
11 #include "base/numerics/safe_conversions.h" | |
12 #include "base/strings/stringprintf.h" | |
13 #include "extensions/browser/api/cast_channel/cast_channel.pb.h" | |
14 #include "extensions/browser/api/cast_channel/cast_framer.h" | |
15 #include "extensions/browser/api/cast_channel/cast_message_util.h" | |
16 #include "extensions/browser/api/cast_channel/logger.h" | |
17 #include "extensions/browser/api/cast_channel/logger_util.h" | |
18 #include "net/base/net_errors.h" | |
19 | |
20 #define VLOG_WITH_CONNECTION(level) \ | |
21 VLOG(level) << "[" << socket_->ip_endpoint().ToString() \ | |
22 << ", auth=" << socket_->channel_auth() << "] " | |
23 | |
24 namespace extensions { | |
25 namespace core_api { | |
26 namespace cast_channel { | |
27 namespace { | |
Wez
2014/09/12 19:59:59
nit: blank between { and the function decl.
| |
28 proto::ReadState ReadStateToProto(CastTransport::ReadState state) { | |
mark a. foltz
2014/09/12 19:41:00
ISTM these could get moved to logger_util.cc and t
Kevin M
2014/09/12 20:22:25
Acknowledged.
| |
29 switch (state) { | |
30 case CastTransport::READ_STATE_NONE: | |
31 return proto::READ_STATE_NONE; | |
32 case CastTransport::READ_STATE_READ: | |
33 return proto::READ_STATE_READ; | |
34 case CastTransport::READ_STATE_READ_COMPLETE: | |
35 return proto::READ_STATE_READ_COMPLETE; | |
36 case CastTransport::READ_STATE_DO_CALLBACK: | |
37 return proto::READ_STATE_DO_CALLBACK; | |
38 case CastTransport::READ_STATE_ERROR: | |
39 return proto::READ_STATE_ERROR; | |
40 default: | |
41 NOTREACHED(); | |
42 return proto::READ_STATE_NONE; | |
43 } | |
44 } | |
45 | |
46 proto::WriteState WriteStateToProto(CastTransport::WriteState state) { | |
47 switch (state) { | |
48 case CastTransport::WRITE_STATE_NONE: | |
49 return proto::WRITE_STATE_NONE; | |
50 case CastTransport::WRITE_STATE_WRITE: | |
51 return proto::WRITE_STATE_WRITE; | |
52 case CastTransport::WRITE_STATE_WRITE_COMPLETE: | |
53 return proto::WRITE_STATE_WRITE_COMPLETE; | |
54 case CastTransport::WRITE_STATE_DO_CALLBACK: | |
55 return proto::WRITE_STATE_DO_CALLBACK; | |
56 case CastTransport::WRITE_STATE_ERROR: | |
57 return proto::WRITE_STATE_ERROR; | |
58 default: | |
59 NOTREACHED(); | |
60 return proto::WRITE_STATE_NONE; | |
61 } | |
62 } | |
63 | |
64 proto::ErrorState ErrorStateToProto(ChannelError state) { | |
65 switch (state) { | |
66 case CHANNEL_ERROR_NONE: | |
67 return proto::CHANNEL_ERROR_NONE; | |
68 case CHANNEL_ERROR_CHANNEL_NOT_OPEN: | |
69 return proto::CHANNEL_ERROR_CHANNEL_NOT_OPEN; | |
70 case CHANNEL_ERROR_AUTHENTICATION_ERROR: | |
71 return proto::CHANNEL_ERROR_AUTHENTICATION_ERROR; | |
72 case CHANNEL_ERROR_CONNECT_ERROR: | |
73 return proto::CHANNEL_ERROR_CONNECT_ERROR; | |
74 case CHANNEL_ERROR_SOCKET_ERROR: | |
75 return proto::CHANNEL_ERROR_SOCKET_ERROR; | |
76 case CHANNEL_ERROR_TRANSPORT_ERROR: | |
77 return proto::CHANNEL_ERROR_TRANSPORT_ERROR; | |
78 case CHANNEL_ERROR_INVALID_MESSAGE: | |
79 return proto::CHANNEL_ERROR_INVALID_MESSAGE; | |
80 case CHANNEL_ERROR_INVALID_CHANNEL_ID: | |
81 return proto::CHANNEL_ERROR_INVALID_CHANNEL_ID; | |
82 case CHANNEL_ERROR_CONNECT_TIMEOUT: | |
83 return proto::CHANNEL_ERROR_CONNECT_TIMEOUT; | |
84 case CHANNEL_ERROR_UNKNOWN: | |
85 return proto::CHANNEL_ERROR_UNKNOWN; | |
86 default: | |
87 NOTREACHED(); | |
88 return proto::CHANNEL_ERROR_NONE; | |
89 } | |
90 } | |
91 } // namespace | |
92 | |
93 CastTransport::CastTransport(CastSocketPlaceholder* socket, | |
94 Delegate* read_delegate, | |
95 scoped_refptr<Logger> logger) | |
96 : socket_(socket), | |
97 read_delegate_(read_delegate), | |
98 write_state_(WRITE_STATE_NONE), | |
99 read_state_(READ_STATE_NONE), | |
100 logger_(logger) { | |
101 DCHECK(socket); | |
102 DCHECK(read_delegate); | |
103 | |
104 // Buffer is reused across messages. | |
105 read_buffer_ = new net::GrowableIOBuffer(); | |
106 read_buffer_->SetCapacity(MessageFramer::MessageHeader::max_message_size()); | |
107 framer_.reset(new MessageFramer(read_buffer_)); | |
108 } | |
109 | |
110 CastTransport::~CastTransport() { | |
111 DCHECK(thread_checker_.CalledOnValidThread()); | |
Wez
2014/09/12 19:59:59
nit: Won't |thread_checker_| do this itself when i
| |
112 FlushWriteQueue(); | |
113 } | |
114 | |
115 void CastTransport::FlushWriteQueue() { | |
116 for (; !write_queue_.empty(); write_queue_.pop()) { | |
117 net::CompletionCallback& callback = write_queue_.front().callback; | |
118 callback.Run(net::ERR_FAILED); | |
119 callback.Reset(); | |
120 } | |
121 } | |
122 | |
123 void CastTransport::SendMessage(const CastMessage& message, | |
124 const net::CompletionCallback& callback) { | |
125 DCHECK(thread_checker_.CalledOnValidThread()); | |
Wez
2014/09/12 20:23:04
nit: As commented in the header, I prefer longer f
| |
126 std::string serialized_message; | |
127 if (!MessageFramer::Serialize(message, &serialized_message)) { | |
128 logger_->LogSocketEventForMessage(socket_->id(), | |
129 proto::SEND_MESSAGE_FAILED, | |
130 message.namespace_(), | |
131 "Error when serializing message."); | |
132 callback.Run(net::ERR_FAILED); | |
133 return; | |
134 } | |
135 WriteRequest write_request( | |
136 message.namespace_(), serialized_message, callback); | |
137 | |
138 write_queue_.push(write_request); | |
139 logger_->LogSocketEventForMessage( | |
140 socket_->id(), | |
141 proto::MESSAGE_ENQUEUED, | |
142 message.namespace_(), | |
143 base::StringPrintf("Queue size: %" PRIuS, write_queue_.size())); | |
144 if (write_state_ == WRITE_STATE_NONE) { | |
145 SetWriteState(WRITE_STATE_WRITE); | |
146 DoWriteLoop(net::OK); | |
147 } | |
148 } | |
149 | |
150 CastTransport::WriteRequest::WriteRequest( | |
151 const std::string& namespace_, | |
152 const std::string& payload, | |
153 const net::CompletionCallback& callback) | |
154 : message_namespace(namespace_), callback(callback) { | |
155 VLOG(2) << "WriteRequest size: " << payload.size(); | |
156 io_buffer = new net::DrainableIOBuffer(new net::StringIOBuffer(payload), | |
157 payload.size()); | |
158 } | |
159 | |
160 CastTransport::WriteRequest::~WriteRequest() { | |
161 } | |
162 | |
163 void CastTransport::SetReadState(ReadState read_state) { | |
164 if (read_state_ != read_state) { | |
165 read_state_ = read_state; | |
166 logger_->LogSocketReadState(socket_->id(), ReadStateToProto(read_state_)); | |
167 } | |
168 } | |
169 | |
170 void CastTransport::SetWriteState(WriteState write_state) { | |
171 if (write_state_ != write_state) { | |
172 write_state_ = write_state; | |
173 logger_->LogSocketWriteState(socket_->id(), | |
174 WriteStateToProto(write_state_)); | |
175 } | |
176 } | |
177 | |
178 void CastTransport::SetErrorState(ChannelError error_state) { | |
179 if (error_state_ != error_state) { | |
180 error_state_ = error_state; | |
181 logger_->LogSocketErrorState(socket_->id(), | |
182 ErrorStateToProto(error_state_)); | |
183 } | |
184 } | |
185 | |
186 void CastTransport::DoWriteLoop(int result) { | |
187 DCHECK(thread_checker_.CalledOnValidThread()); | |
188 VLOG_WITH_CONNECTION(1) << "DoWriteLoop queue size: " << write_queue_.size(); | |
189 | |
190 if (write_queue_.empty()) { | |
191 SetWriteState(WRITE_STATE_NONE); | |
192 return; | |
193 } | |
194 | |
195 // Network operations can either finish synchronously or asynchronously. | |
196 // This method executes the state machine transitions in a loop so that | |
197 // write state transitions happen even when network operations finish | |
198 // synchronously. | |
199 int rv = result; | |
200 do { | |
201 WriteState state = write_state_; | |
202 write_state_ = WRITE_STATE_NONE; | |
203 switch (state) { | |
204 case WRITE_STATE_WRITE: | |
205 rv = DoWrite(); | |
206 break; | |
207 case WRITE_STATE_WRITE_COMPLETE: | |
208 rv = DoWriteComplete(rv); | |
209 break; | |
210 case WRITE_STATE_DO_CALLBACK: | |
211 rv = DoWriteCallback(); | |
212 break; | |
213 case WRITE_STATE_ERROR: | |
214 rv = DoWriteError(rv); | |
215 break; | |
216 default: | |
217 NOTREACHED() << "BUG in write flow. Unknown state: " << state; | |
218 break; | |
219 } | |
220 } while (!write_queue_.empty() && rv != net::ERR_IO_PENDING && | |
221 write_state_ != WRITE_STATE_NONE); | |
222 | |
223 // No state change occurred in do-while loop above. This means state has | |
224 // transitioned to NONE. | |
Wez
2014/09/12 20:23:04
Isn't this the wrong way around? Doesn't the state
| |
225 if (write_state_ == WRITE_STATE_NONE) { | |
226 logger_->LogSocketWriteState(socket_->id(), | |
227 WriteStateToProto(write_state_)); | |
Wez
2014/09/12 20:23:04
Are we logging this to get unexpected-state stats?
| |
228 } | |
229 | |
230 // If write loop is done because the queue is empty then set write | |
231 // state to NONE | |
Wez
2014/09/12 20:23:04
This comment is tautologous with the code below; c
| |
232 if (write_queue_.empty()) { | |
233 SetWriteState(WRITE_STATE_NONE); | |
234 } | |
235 | |
236 // Write loop is done - if the result is ERR_FAILED then close with error. | |
237 if (rv == net::ERR_FAILED) { | |
Wez
2014/09/12 20:23:04
Is ERR_FAILED the only possible error code we coul
| |
238 DCHECK_NE(CHANNEL_ERROR_NONE, error_state_); | |
239 socket_->CloseWithError(error_state_); | |
240 FlushWriteQueue(); | |
241 } | |
242 } | |
243 | |
244 int CastTransport::DoWrite() { | |
245 DCHECK(!write_queue_.empty()); | |
246 WriteRequest& request = write_queue_.front(); | |
247 | |
248 VLOG_WITH_CONNECTION(2) << "WriteData byte_count = " | |
249 << request.io_buffer->size() << " bytes_written " | |
250 << request.io_buffer->BytesConsumed(); | |
251 | |
252 SetWriteState(WRITE_STATE_WRITE_COMPLETE); | |
253 | |
254 int rv = socket_->Write( | |
255 request.io_buffer.get(), | |
256 request.io_buffer->BytesRemaining(), | |
257 base::Bind(&CastTransport::DoWriteLoop, base::Unretained(this))); | |
258 logger_->LogSocketEventWithRv(socket_->id(), proto::SOCKET_WRITE, rv); | |
259 | |
260 return rv; | |
261 } | |
262 | |
263 int CastTransport::DoWriteComplete(int result) { | |
264 VLOG_WITH_CONNECTION(2) << "DoWriteComplete result=" << result; | |
265 DCHECK(!write_queue_.empty()); | |
266 if (result <= 0) { // NOTE that 0 also indicates an error | |
267 SetErrorState(CHANNEL_ERROR_TRANSPORT_ERROR); | |
268 SetWriteState(WRITE_STATE_ERROR); | |
269 return result == 0 ? net::ERR_FAILED : result; | |
270 } | |
271 | |
272 // Some bytes were successfully written | |
273 WriteRequest& request = write_queue_.front(); | |
274 scoped_refptr<net::DrainableIOBuffer> io_buffer = request.io_buffer; | |
275 io_buffer->DidConsume(result); | |
276 if (io_buffer->BytesRemaining() == 0) { // Message fully sent | |
277 SetWriteState(WRITE_STATE_DO_CALLBACK); | |
278 } else { | |
279 SetWriteState(WRITE_STATE_WRITE); | |
280 } | |
281 | |
282 return net::OK; | |
283 } | |
284 | |
285 int CastTransport::DoWriteCallback() { | |
286 VLOG_WITH_CONNECTION(2) << "DoWriteCallback"; | |
287 DCHECK(!write_queue_.empty()); | |
288 | |
289 SetWriteState(WRITE_STATE_WRITE); | |
290 | |
291 WriteRequest& request = write_queue_.front(); | |
292 int bytes_consumed = request.io_buffer->BytesConsumed(); | |
293 logger_->LogSocketEventForMessage( | |
294 socket_->id(), | |
295 proto::MESSAGE_WRITTEN, | |
296 request.message_namespace, | |
297 base::StringPrintf("Bytes: %d", bytes_consumed)); | |
298 request.callback.Run(net::OK); | |
299 write_queue_.pop(); | |
300 return net::OK; | |
301 } | |
302 | |
303 int CastTransport::DoWriteError(int result) { | |
304 VLOG_WITH_CONNECTION(2) << "DoWriteError result=" << result; | |
305 DCHECK_NE(CHANNEL_ERROR_NONE, error_state_); | |
306 DCHECK_LT(result, 0); | |
307 return net::ERR_FAILED; | |
308 } | |
309 | |
310 void CastTransport::StartReadLoop() { | |
311 DCHECK(thread_checker_.CalledOnValidThread()); | |
312 // Read loop would have already been started if read state is not NONE | |
313 if (read_state_ == READ_STATE_NONE) { | |
314 SetReadState(READ_STATE_READ); | |
315 DoReadLoop(net::OK); | |
316 } | |
317 } | |
318 | |
319 void CastTransport::DoReadLoop(int result) { | |
320 DCHECK(thread_checker_.CalledOnValidThread()); | |
321 // Network operations can either finish synchronously or asynchronously. | |
322 // This method executes the state machine transitions in a loop so that | |
323 // write state transitions happen even when network operations finish | |
324 // synchronously. | |
325 int rv = result; | |
326 do { | |
327 ReadState state = read_state_; | |
328 read_state_ = READ_STATE_NONE; | |
329 | |
330 switch (state) { | |
331 case READ_STATE_READ: | |
332 rv = DoRead(); | |
333 break; | |
334 case READ_STATE_READ_COMPLETE: | |
335 rv = DoReadComplete(rv); | |
336 break; | |
337 case READ_STATE_DO_CALLBACK: | |
338 rv = DoReadCallback(); | |
339 break; | |
340 case READ_STATE_ERROR: | |
341 rv = DoReadError(rv); | |
342 DCHECK_EQ(read_state_, READ_STATE_NONE); | |
343 break; | |
344 default: | |
345 NOTREACHED() << "BUG in read flow. Unknown state: " << state; | |
346 break; | |
347 } | |
348 } while (rv != net::ERR_IO_PENDING && read_state_ != READ_STATE_NONE); | |
349 | |
350 // No state change occurred in do-while loop above. This means state has | |
351 // transitioned to NONE. | |
352 if (read_state_ == READ_STATE_NONE) { | |
353 logger_->LogSocketReadState(socket_->id(), ReadStateToProto(read_state_)); | |
354 } | |
355 | |
356 if (rv == net::ERR_FAILED) { | |
357 DCHECK_NE(CHANNEL_ERROR_NONE, error_state_); | |
358 socket_->CloseWithError(error_state_); | |
359 FlushWriteQueue(); | |
360 read_delegate_->OnError( | |
361 socket_, error_state_, logger_->GetLastErrors(socket_->id())); | |
362 } | |
363 } | |
364 | |
365 int CastTransport::DoRead() { | |
366 VLOG_WITH_CONNECTION(2) << "DoRead"; | |
367 SetReadState(READ_STATE_READ_COMPLETE); | |
368 | |
369 // Determine how many bytes need to be read. | |
370 size_t num_bytes_to_read = framer_->BytesRequested(); | |
371 | |
372 // Read up to num_bytes_to_read into |current_read_buffer_|. | |
373 int rv = socket_->Read( | |
374 read_buffer_.get(), | |
375 base::checked_cast<uint32>(num_bytes_to_read), | |
376 base::Bind(&CastTransport::DoReadLoop, base::Unretained(this))); | |
377 | |
378 return rv; | |
379 } | |
380 | |
381 int CastTransport::DoReadComplete(int result) { | |
382 VLOG_WITH_CONNECTION(2) << "DoReadComplete result = " << result; | |
383 | |
384 if (result <= 0) { | |
385 SetErrorState(CHANNEL_ERROR_TRANSPORT_ERROR); | |
386 SetReadState(READ_STATE_ERROR); | |
387 return result == 0 ? net::ERR_FAILED : result; | |
388 } | |
389 | |
390 size_t message_size; | |
391 DCHECK(current_message_.get() == NULL); | |
392 current_message_ = framer_->Ingest(result, &message_size, &error_state_); | |
393 if (current_message_.get()) { | |
394 DCHECK_EQ(error_state_, CHANNEL_ERROR_NONE); | |
395 DCHECK_GT(message_size, static_cast<size_t>(0)); | |
396 logger_->LogSocketEventForMessage( | |
397 socket_->id(), | |
398 proto::MESSAGE_READ, | |
399 current_message_->namespace_(), | |
400 base::StringPrintf("Message size: %zu", message_size)); | |
401 SetReadState(READ_STATE_DO_CALLBACK); | |
402 } else if (error_state_ != CHANNEL_ERROR_NONE) { | |
403 DCHECK(current_message_.get() == NULL); | |
404 SetErrorState(CHANNEL_ERROR_INVALID_MESSAGE); | |
405 SetReadState(READ_STATE_ERROR); | |
406 } else { | |
407 DCHECK(current_message_.get() == NULL); | |
408 SetReadState(READ_STATE_READ); | |
409 } | |
410 return net::OK; | |
411 } | |
412 | |
413 int CastTransport::DoReadCallback() { | |
414 VLOG_WITH_CONNECTION(2) << "DoReadCallback"; | |
415 SetReadState(READ_STATE_READ); | |
416 if (!ValidateCastMessage(*current_message_)) { | |
417 SetReadState(READ_STATE_ERROR); | |
418 SetErrorState(CHANNEL_ERROR_INVALID_MESSAGE); | |
419 return net::ERR_INVALID_RESPONSE; | |
420 } | |
421 logger_->LogSocketEventForMessage(socket_->id(), | |
422 proto::NOTIFY_ON_MESSAGE, | |
423 current_message_->namespace_(), | |
424 std::string()); | |
425 read_delegate_->OnMessage(socket_, *current_message_); | |
426 current_message_.reset(); | |
427 return net::OK; | |
428 } | |
429 | |
430 int CastTransport::DoReadError(int result) { | |
431 VLOG_WITH_CONNECTION(2) << "DoReadError"; | |
432 DCHECK_NE(CHANNEL_ERROR_NONE, error_state_); | |
433 DCHECK_LE(result, 0); | |
434 return net::ERR_FAILED; | |
435 } | |
436 } // namespace cast_channel | |
437 } // namespace core_api | |
438 } // namespace extensions | |
OLD | NEW |