Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(744)

Side by Side Diff: extensions/browser/api/cast_channel/cast_transport.cc

Issue 555283002: Create new class "CastTransport", which encapsulates the message read and write event loops. (Closed) Base URL: https://chromium.googlesource.com/chromium/src.git@master
Patch Set: Code review feedback. Created 6 years, 3 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
OLDNEW
(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
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698