| OLD | NEW |
|---|
| (Empty) |
| 1 // Copyright 2013 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 "google_apis/gcm/base/socket_stream.h" | |
| 6 | |
| 7 #include "base/bind.h" | |
| 8 #include "base/callback.h" | |
| 9 #include "net/base/io_buffer.h" | |
| 10 #include "net/socket/stream_socket.h" | |
| 11 | |
| 12 namespace gcm { | |
| 13 | |
| 14 namespace { | |
| 15 | |
| 16 // TODO(zea): consider having dynamically-sized buffers if this becomes too | |
| 17 // expensive. | |
| 18 const uint32 kDefaultBufferSize = 8*1024; | |
| 19 | |
| 20 } // namespace | |
| 21 | |
| 22 SocketInputStream::SocketInputStream(net::StreamSocket* socket) | |
| 23 : socket_(socket), | |
| 24 io_buffer_(new net::IOBuffer(kDefaultBufferSize)), | |
| 25 read_buffer_(new net::DrainableIOBuffer(io_buffer_.get(), | |
| 26 kDefaultBufferSize)), | |
| 27 next_pos_(0), | |
| 28 last_error_(net::OK), | |
| 29 weak_ptr_factory_(this) { | |
| 30 DCHECK(socket->IsConnected()); | |
| 31 } | |
| 32 | |
| 33 SocketInputStream::~SocketInputStream() { | |
| 34 } | |
| 35 | |
| 36 bool SocketInputStream::Next(const void** data, int* size) { | |
| 37 if (GetState() != EMPTY && GetState() != READY) { | |
| 38 NOTREACHED() << "Invalid input stream read attempt."; | |
| 39 return false; | |
| 40 } | |
| 41 | |
| 42 if (GetState() == EMPTY) { | |
| 43 DVLOG(1) << "No unread data remaining, ending read."; | |
| 44 return false; | |
| 45 } | |
| 46 | |
| 47 DCHECK_EQ(GetState(), READY) | |
| 48 << " Input stream must have pending data before reading."; | |
| 49 DCHECK_LT(next_pos_, read_buffer_->BytesConsumed()); | |
| 50 *data = io_buffer_->data() + next_pos_; | |
| 51 *size = UnreadByteCount(); | |
| 52 next_pos_ = read_buffer_->BytesConsumed(); | |
| 53 DVLOG(1) << "Consuming " << *size << " bytes in input buffer."; | |
| 54 return true; | |
| 55 } | |
| 56 | |
| 57 void SocketInputStream::BackUp(int count) { | |
| 58 DCHECK(GetState() == READY || GetState() == EMPTY); | |
| 59 DCHECK_GT(count, 0); | |
| 60 DCHECK_LE(count, next_pos_); | |
| 61 | |
| 62 next_pos_ -= count; | |
| 63 DVLOG(1) << "Backing up " << count << " bytes in input buffer. " | |
| 64 << "Current position now at " << next_pos_ | |
| 65 << " of " << read_buffer_->BytesConsumed(); | |
| 66 } | |
| 67 | |
| 68 bool SocketInputStream::Skip(int count) { | |
| 69 NOTIMPLEMENTED(); | |
| 70 return false; | |
| 71 } | |
| 72 | |
| 73 int64 SocketInputStream::ByteCount() const { | |
| 74 DCHECK_NE(GetState(), CLOSED); | |
| 75 DCHECK_NE(GetState(), READING); | |
| 76 return next_pos_; | |
| 77 } | |
| 78 | |
| 79 size_t SocketInputStream::UnreadByteCount() const { | |
| 80 DCHECK_NE(GetState(), CLOSED); | |
| 81 DCHECK_NE(GetState(), READING); | |
| 82 return read_buffer_->BytesConsumed() - next_pos_; | |
| 83 } | |
| 84 | |
| 85 net::Error SocketInputStream::Refresh(const base::Closure& callback, | |
| 86 int byte_limit) { | |
| 87 DCHECK_NE(GetState(), CLOSED); | |
| 88 DCHECK_NE(GetState(), READING); | |
| 89 DCHECK_GT(byte_limit, 0); | |
| 90 | |
| 91 if (byte_limit > read_buffer_->BytesRemaining()) { | |
| 92 NOTREACHED() << "Out of buffer space, closing input stream."; | |
| 93 CloseStream(net::ERR_UNEXPECTED, base::Closure()); | |
| 94 return net::OK; | |
| 95 } | |
| 96 | |
| 97 if (!socket_->IsConnected()) { | |
| 98 LOG(ERROR) << "Socket was disconnected, closing input stream"; | |
| 99 CloseStream(net::ERR_CONNECTION_CLOSED, base::Closure()); | |
| 100 return net::OK; | |
| 101 } | |
| 102 | |
| 103 DVLOG(1) << "Refreshing input stream, limit of " << byte_limit << " bytes."; | |
| 104 int result = socket_->Read( | |
| 105 read_buffer_, | |
| 106 byte_limit, | |
| 107 base::Bind(&SocketInputStream::RefreshCompletionCallback, | |
| 108 weak_ptr_factory_.GetWeakPtr(), | |
| 109 callback)); | |
| 110 DVLOG(1) << "Read returned " << result; | |
| 111 if (result == net::ERR_IO_PENDING) { | |
| 112 last_error_ = net::ERR_IO_PENDING; | |
| 113 return net::ERR_IO_PENDING; | |
| 114 } | |
| 115 | |
| 116 RefreshCompletionCallback(base::Closure(), result); | |
| 117 return net::OK; | |
| 118 } | |
| 119 | |
| 120 void SocketInputStream::RebuildBuffer() { | |
| 121 DVLOG(1) << "Rebuilding input stream, consumed " | |
| 122 << next_pos_ << " bytes."; | |
| 123 DCHECK_NE(GetState(), READING); | |
| 124 DCHECK_NE(GetState(), CLOSED); | |
| 125 | |
| 126 int unread_data_size = 0; | |
| 127 const void* unread_data_ptr = NULL; | |
| 128 Next(&unread_data_ptr, &unread_data_size); | |
| 129 ResetInternal(); | |
| 130 | |
| 131 if (unread_data_ptr != io_buffer_->data()) { | |
| 132 DVLOG(1) << "Have " << unread_data_size | |
| 133 << " unread bytes remaining, shifting."; | |
| 134 // Move any remaining unread data to the start of the buffer; | |
| 135 std::memmove(io_buffer_->data(), unread_data_ptr, unread_data_size); | |
| 136 } else { | |
| 137 DVLOG(1) << "Have " << unread_data_size << " unread bytes remaining."; | |
| 138 } | |
| 139 read_buffer_->DidConsume(unread_data_size); | |
| 140 } | |
| 141 | |
| 142 net::Error SocketInputStream::last_error() const { | |
| 143 return last_error_; | |
| 144 } | |
| 145 | |
| 146 SocketInputStream::State SocketInputStream::GetState() const { | |
| 147 if (last_error_ < net::ERR_IO_PENDING) | |
| 148 return CLOSED; | |
| 149 | |
| 150 if (last_error_ == net::ERR_IO_PENDING) | |
| 151 return READING; | |
| 152 | |
| 153 DCHECK_EQ(last_error_, net::OK); | |
| 154 if (read_buffer_->BytesConsumed() == next_pos_) | |
| 155 return EMPTY; | |
| 156 | |
| 157 return READY; | |
| 158 } | |
| 159 | |
| 160 void SocketInputStream::RefreshCompletionCallback( | |
| 161 const base::Closure& callback, int result) { | |
| 162 // If an error occurred before the completion callback could complete, ignore | |
| 163 // the result. | |
| 164 if (GetState() == CLOSED) | |
| 165 return; | |
| 166 | |
| 167 // Result == 0 implies EOF, which is treated as an error. | |
| 168 if (result == 0) | |
| 169 result = net::ERR_CONNECTION_CLOSED; | |
| 170 | |
| 171 DCHECK_NE(result, net::ERR_IO_PENDING); | |
| 172 | |
| 173 if (result < net::OK) { | |
| 174 DVLOG(1) << "Failed to refresh socket: " << result; | |
| 175 CloseStream(static_cast<net::Error>(result), callback); | |
| 176 return; | |
| 177 } | |
| 178 | |
| 179 DCHECK_GT(result, 0); | |
| 180 last_error_ = net::OK; | |
| 181 read_buffer_->DidConsume(result); | |
| 182 | |
| 183 DVLOG(1) << "Refresh complete with " << result << " new bytes. " | |
| 184 << "Current position " << next_pos_ | |
| 185 << " of " << read_buffer_->BytesConsumed() << "."; | |
| 186 | |
| 187 if (!callback.is_null()) | |
| 188 callback.Run(); | |
| 189 } | |
| 190 | |
| 191 void SocketInputStream::ResetInternal() { | |
| 192 read_buffer_->SetOffset(0); | |
| 193 next_pos_ = 0; | |
| 194 last_error_ = net::OK; | |
| 195 weak_ptr_factory_.InvalidateWeakPtrs(); // Invalidate any callbacks. | |
| 196 } | |
| 197 | |
| 198 void SocketInputStream::CloseStream(net::Error error, | |
| 199 const base::Closure& callback) { | |
| 200 DCHECK_LT(error, net::ERR_IO_PENDING); | |
| 201 ResetInternal(); | |
| 202 last_error_ = error; | |
| 203 LOG(ERROR) << "Closing stream with result " << error; | |
| 204 if (!callback.is_null()) | |
| 205 callback.Run(); | |
| 206 } | |
| 207 | |
| 208 SocketOutputStream::SocketOutputStream(net::StreamSocket* socket) | |
| 209 : socket_(socket), | |
| 210 io_buffer_(new net::IOBuffer(kDefaultBufferSize)), | |
| 211 write_buffer_(new net::DrainableIOBuffer(io_buffer_.get(), | |
| 212 kDefaultBufferSize)), | |
| 213 next_pos_(0), | |
| 214 last_error_(net::OK), | |
| 215 weak_ptr_factory_(this) { | |
| 216 DCHECK(socket->IsConnected()); | |
| 217 } | |
| 218 | |
| 219 SocketOutputStream::~SocketOutputStream() { | |
| 220 } | |
| 221 | |
| 222 bool SocketOutputStream::Next(void** data, int* size) { | |
| 223 DCHECK_NE(GetState(), CLOSED); | |
| 224 DCHECK_NE(GetState(), FLUSHING); | |
| 225 if (next_pos_ == write_buffer_->size()) | |
| 226 return false; | |
| 227 | |
| 228 *data = write_buffer_->data() + next_pos_; | |
| 229 *size = write_buffer_->size() - next_pos_; | |
| 230 next_pos_ = write_buffer_->size(); | |
| 231 return true; | |
| 232 } | |
| 233 | |
| 234 void SocketOutputStream::BackUp(int count) { | |
| 235 DCHECK_GE(count, 0); | |
| 236 if (count > next_pos_) | |
| 237 next_pos_ = 0; | |
| 238 next_pos_ -= count; | |
| 239 DVLOG(1) << "Backing up " << count << " bytes in output buffer. " | |
| 240 << next_pos_ << " bytes used."; | |
| 241 } | |
| 242 | |
| 243 int64 SocketOutputStream::ByteCount() const { | |
| 244 DCHECK_NE(GetState(), CLOSED); | |
| 245 DCHECK_NE(GetState(), FLUSHING); | |
| 246 return next_pos_; | |
| 247 } | |
| 248 | |
| 249 net::Error SocketOutputStream::Flush(const base::Closure& callback) { | |
| 250 DCHECK_EQ(GetState(), READY); | |
| 251 | |
| 252 if (!socket_->IsConnected()) { | |
| 253 LOG(ERROR) << "Socket was disconnected, closing output stream"; | |
| 254 last_error_ = net::ERR_CONNECTION_CLOSED; | |
| 255 return net::OK; | |
| 256 } | |
| 257 | |
| 258 DVLOG(1) << "Flushing " << next_pos_ << " bytes into socket."; | |
| 259 int result = socket_->Write( | |
| 260 write_buffer_, | |
| 261 next_pos_, | |
| 262 base::Bind(&SocketOutputStream::FlushCompletionCallback, | |
| 263 weak_ptr_factory_.GetWeakPtr(), | |
| 264 callback)); | |
| 265 DVLOG(1) << "Write returned " << result; | |
| 266 if (result == net::ERR_IO_PENDING) { | |
| 267 last_error_ = net::ERR_IO_PENDING; | |
| 268 return net::ERR_IO_PENDING; | |
| 269 } | |
| 270 | |
| 271 FlushCompletionCallback(base::Closure(), result); | |
| 272 return net::OK; | |
| 273 } | |
| 274 | |
| 275 SocketOutputStream::State SocketOutputStream::GetState() const{ | |
| 276 if (last_error_ < net::ERR_IO_PENDING) | |
| 277 return CLOSED; | |
| 278 | |
| 279 if (last_error_ == net::ERR_IO_PENDING) | |
| 280 return FLUSHING; | |
| 281 | |
| 282 DCHECK_EQ(last_error_, net::OK); | |
| 283 if (next_pos_ == 0) | |
| 284 return EMPTY; | |
| 285 | |
| 286 return READY; | |
| 287 } | |
| 288 | |
| 289 net::Error SocketOutputStream::last_error() const { | |
| 290 return last_error_; | |
| 291 } | |
| 292 | |
| 293 void SocketOutputStream::FlushCompletionCallback( | |
| 294 const base::Closure& callback, int result) { | |
| 295 // If an error occurred before the completion callback could complete, ignore | |
| 296 // the result. | |
| 297 if (GetState() == CLOSED) | |
| 298 return; | |
| 299 | |
| 300 // Result == 0 implies EOF, which is treated as an error. | |
| 301 if (result == 0) | |
| 302 result = net::ERR_CONNECTION_CLOSED; | |
| 303 | |
| 304 DCHECK_NE(result, net::ERR_IO_PENDING); | |
| 305 | |
| 306 if (result < net::OK) { | |
| 307 LOG(ERROR) << "Failed to flush socket."; | |
| 308 last_error_ = static_cast<net::Error>(result); | |
| 309 if (!callback.is_null()) | |
| 310 callback.Run(); | |
| 311 return; | |
| 312 } | |
| 313 | |
| 314 DCHECK_GT(result, net::OK); | |
| 315 last_error_ = net::OK; | |
| 316 | |
| 317 if (write_buffer_->BytesConsumed() + result < next_pos_) { | |
| 318 DVLOG(1) << "Partial flush complete. Retrying."; | |
| 319 // Only a partial write was completed. Flush again to finish the write. | |
| 320 write_buffer_->DidConsume(result); | |
| 321 Flush(callback); | |
| 322 return; | |
| 323 } | |
| 324 | |
| 325 DVLOG(1) << "Socket flush complete."; | |
| 326 write_buffer_->SetOffset(0); | |
| 327 next_pos_ = 0; | |
| 328 if (!callback.is_null()) | |
| 329 callback.Run(); | |
| 330 } | |
| 331 | |
| 332 } // namespace gcm | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 29273008:
Revert r229533 "[GCM] Initial work to set up directory structure and introduce socket integration" (Closed)
Base URL: svn://svn.chromium.org/chrome/trunk/src