OLD | NEW |
1 // Copyright (c) 2010 The Chromium Authors. All rights reserved. | 1 // Copyright (c) 2010 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 "remoting/protocol/buffered_socket_writer.h" | 5 #include "remoting/protocol/buffered_socket_writer.h" |
6 | 6 |
7 #include "base/message_loop.h" | 7 #include "base/message_loop.h" |
8 #include "base/stl_util-inl.h" | 8 #include "base/stl_util-inl.h" |
9 #include "net/base/net_errors.h" | 9 #include "net/base/net_errors.h" |
10 | 10 |
(...skipping 30 matching lines...) Expand all Loading... |
41 ALLOW_THIS_IN_INITIALIZER_LIST( | 41 ALLOW_THIS_IN_INITIALIZER_LIST( |
42 written_callback_(this, &BufferedSocketWriterBase::OnWritten)), | 42 written_callback_(this, &BufferedSocketWriterBase::OnWritten)), |
43 closed_(false) { | 43 closed_(false) { |
44 } | 44 } |
45 | 45 |
46 BufferedSocketWriterBase::~BufferedSocketWriterBase() { } | 46 BufferedSocketWriterBase::~BufferedSocketWriterBase() { } |
47 | 47 |
48 void BufferedSocketWriterBase::Init(net::Socket* socket, | 48 void BufferedSocketWriterBase::Init(net::Socket* socket, |
49 WriteFailedCallback* callback) { | 49 WriteFailedCallback* callback) { |
50 // TODO(garykac) Save copy of WriteFailedCallback. | 50 // TODO(garykac) Save copy of WriteFailedCallback. |
51 AutoLock auto_lock(lock_); | 51 base::AutoLock auto_lock(lock_); |
52 message_loop_ = MessageLoop::current(); | 52 message_loop_ = MessageLoop::current(); |
53 socket_ = socket; | 53 socket_ = socket; |
54 DCHECK(socket_); | 54 DCHECK(socket_); |
55 } | 55 } |
56 | 56 |
57 bool BufferedSocketWriterBase::Write( | 57 bool BufferedSocketWriterBase::Write( |
58 scoped_refptr<net::IOBufferWithSize> data, Task* done_task) { | 58 scoped_refptr<net::IOBufferWithSize> data, Task* done_task) { |
59 AutoLock auto_lock(lock_); | 59 base::AutoLock auto_lock(lock_); |
60 if (!socket_) | 60 if (!socket_) |
61 return false; | 61 return false; |
62 queue_.push_back(new PendingPacket(data, done_task)); | 62 queue_.push_back(new PendingPacket(data, done_task)); |
63 buffer_size_ += data->size(); | 63 buffer_size_ += data->size(); |
64 message_loop_->PostTask( | 64 message_loop_->PostTask( |
65 FROM_HERE, NewRunnableMethod(this, &BufferedSocketWriterBase::DoWrite)); | 65 FROM_HERE, NewRunnableMethod(this, &BufferedSocketWriterBase::DoWrite)); |
66 return true; | 66 return true; |
67 } | 67 } |
68 | 68 |
69 void BufferedSocketWriterBase::DoWrite() { | 69 void BufferedSocketWriterBase::DoWrite() { |
70 DCHECK_EQ(message_loop_, MessageLoop::current()); | 70 DCHECK_EQ(message_loop_, MessageLoop::current()); |
71 DCHECK(socket_); | 71 DCHECK(socket_); |
72 | 72 |
73 // Don't try to write if there is another write pending. | 73 // Don't try to write if there is another write pending. |
74 if (write_pending_) | 74 if (write_pending_) |
75 return; | 75 return; |
76 | 76 |
77 // Don't write after Close(). | 77 // Don't write after Close(). |
78 { | 78 { |
79 AutoLock auto_lock(lock_); | 79 base::AutoLock auto_lock(lock_); |
80 if (closed_) | 80 if (closed_) |
81 return; | 81 return; |
82 } | 82 } |
83 | 83 |
84 while (true) { | 84 while (true) { |
85 net::IOBuffer* current_packet; | 85 net::IOBuffer* current_packet; |
86 int current_packet_size; | 86 int current_packet_size; |
87 { | 87 { |
88 AutoLock auto_lock(lock_); | 88 base::AutoLock auto_lock(lock_); |
89 GetNextPacket_Locked(¤t_packet, ¤t_packet_size); | 89 GetNextPacket_Locked(¤t_packet, ¤t_packet_size); |
90 } | 90 } |
91 | 91 |
92 // Return if the queue is empty. | 92 // Return if the queue is empty. |
93 if (!current_packet) | 93 if (!current_packet) |
94 return; | 94 return; |
95 | 95 |
96 int result = socket_->Write(current_packet, current_packet_size, | 96 int result = socket_->Write(current_packet, current_packet_size, |
97 &written_callback_); | 97 &written_callback_); |
98 if (result >= 0) { | 98 if (result >= 0) { |
99 AutoLock auto_lock(lock_); | 99 base::AutoLock auto_lock(lock_); |
100 AdvanceBufferPosition_Locked(result); | 100 AdvanceBufferPosition_Locked(result); |
101 } else { | 101 } else { |
102 if (result == net::ERR_IO_PENDING) { | 102 if (result == net::ERR_IO_PENDING) { |
103 write_pending_ = true; | 103 write_pending_ = true; |
104 } else { | 104 } else { |
105 HandleError(result); | 105 HandleError(result); |
106 if (write_failed_callback_.get()) | 106 if (write_failed_callback_.get()) |
107 write_failed_callback_->Run(result); | 107 write_failed_callback_->Run(result); |
108 } | 108 } |
109 return; | 109 return; |
110 } | 110 } |
111 } | 111 } |
112 } | 112 } |
113 | 113 |
114 void BufferedSocketWriterBase::OnWritten(int result) { | 114 void BufferedSocketWriterBase::OnWritten(int result) { |
115 DCHECK_EQ(message_loop_, MessageLoop::current()); | 115 DCHECK_EQ(message_loop_, MessageLoop::current()); |
116 write_pending_ = false; | 116 write_pending_ = false; |
117 | 117 |
118 if (result < 0) { | 118 if (result < 0) { |
119 HandleError(result); | 119 HandleError(result); |
120 if (write_failed_callback_.get()) | 120 if (write_failed_callback_.get()) |
121 write_failed_callback_->Run(result); | 121 write_failed_callback_->Run(result); |
122 return; | 122 return; |
123 } | 123 } |
124 | 124 |
125 { | 125 { |
126 AutoLock auto_lock(lock_); | 126 base::AutoLock auto_lock(lock_); |
127 AdvanceBufferPosition_Locked(result); | 127 AdvanceBufferPosition_Locked(result); |
128 } | 128 } |
129 | 129 |
130 // Schedule next write. | 130 // Schedule next write. |
131 message_loop_->PostTask( | 131 message_loop_->PostTask( |
132 FROM_HERE, NewRunnableMethod(this, &BufferedSocketWriterBase::DoWrite)); | 132 FROM_HERE, NewRunnableMethod(this, &BufferedSocketWriterBase::DoWrite)); |
133 } | 133 } |
134 | 134 |
135 void BufferedSocketWriterBase::HandleError(int result) { | 135 void BufferedSocketWriterBase::HandleError(int result) { |
136 AutoLock auto_lock(lock_); | 136 base::AutoLock auto_lock(lock_); |
137 closed_ = true; | 137 closed_ = true; |
138 STLDeleteElements(&queue_); | 138 STLDeleteElements(&queue_); |
139 | 139 |
140 // Notify subclass that an error is received. | 140 // Notify subclass that an error is received. |
141 OnError_Locked(result); | 141 OnError_Locked(result); |
142 } | 142 } |
143 | 143 |
144 int BufferedSocketWriterBase::GetBufferSize() { | 144 int BufferedSocketWriterBase::GetBufferSize() { |
145 AutoLock auto_lock(lock_); | 145 base::AutoLock auto_lock(lock_); |
146 return buffer_size_; | 146 return buffer_size_; |
147 } | 147 } |
148 | 148 |
149 int BufferedSocketWriterBase::GetBufferChunks() { | 149 int BufferedSocketWriterBase::GetBufferChunks() { |
150 AutoLock auto_lock(lock_); | 150 base::AutoLock auto_lock(lock_); |
151 return queue_.size(); | 151 return queue_.size(); |
152 } | 152 } |
153 | 153 |
154 void BufferedSocketWriterBase::Close() { | 154 void BufferedSocketWriterBase::Close() { |
155 AutoLock auto_lock(lock_); | 155 base::AutoLock auto_lock(lock_); |
156 closed_ = true; | 156 closed_ = true; |
157 } | 157 } |
158 | 158 |
159 void BufferedSocketWriterBase::PopQueue() { | 159 void BufferedSocketWriterBase::PopQueue() { |
160 // This also calls |done_task|. | 160 // This also calls |done_task|. |
161 delete queue_.front(); | 161 delete queue_.front(); |
162 queue_.pop_front(); | 162 queue_.pop_front(); |
163 } | 163 } |
164 | 164 |
165 BufferedSocketWriter::BufferedSocketWriter() { } | 165 BufferedSocketWriter::BufferedSocketWriter() { } |
(...skipping 49 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
215 buffer_size_ -= queue_.front()->data()->size(); | 215 buffer_size_ -= queue_.front()->data()->size(); |
216 PopQueue(); | 216 PopQueue(); |
217 } | 217 } |
218 | 218 |
219 void BufferedDatagramWriter::OnError_Locked(int result) { | 219 void BufferedDatagramWriter::OnError_Locked(int result) { |
220 // Nothing to do here. | 220 // Nothing to do here. |
221 } | 221 } |
222 | 222 |
223 } // namespace protocol | 223 } // namespace protocol |
224 } // namespace remoting | 224 } // namespace remoting |
OLD | NEW |