1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
// Copyright 2015 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "remoting/base/buffered_socket_writer.h"
#include "base/bind.h"
#include "base/stl_util.h"
#include "net/base/io_buffer.h"
#include "net/base/net_errors.h"
#include "net/socket/socket.h"
namespace remoting {
namespace {
int WriteNetSocket(net::Socket* socket,
const scoped_refptr<net::IOBuffer>& buf,
int buf_len,
const net::CompletionCallback& callback) {
return socket->Write(buf.get(), buf_len, callback);
}
} // namespace
struct BufferedSocketWriter::PendingPacket {
PendingPacket(scoped_refptr<net::DrainableIOBuffer> data,
const base::Closure& done_task)
: data(data),
done_task(done_task) {
}
scoped_refptr<net::DrainableIOBuffer> data;
base::Closure done_task;
};
// static
scoped_ptr<BufferedSocketWriter> BufferedSocketWriter::CreateForSocket(
net::Socket* socket,
const WriteFailedCallback& write_failed_callback) {
scoped_ptr<BufferedSocketWriter> result(new BufferedSocketWriter());
result->Init(base::Bind(&WriteNetSocket, socket), write_failed_callback);
return result.Pass();
}
BufferedSocketWriter::BufferedSocketWriter() : weak_factory_(this) {
}
BufferedSocketWriter::~BufferedSocketWriter() {
STLDeleteElements(&queue_);
}
void BufferedSocketWriter::Init(
const WriteCallback& write_callback,
const WriteFailedCallback& write_failed_callback) {
write_callback_ = write_callback;
write_failed_callback_ = write_failed_callback;
}
bool BufferedSocketWriter::Write(
const scoped_refptr<net::IOBufferWithSize>& data,
const base::Closure& done_task) {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK(data.get());
// Don't write after error.
if (is_closed())
return false;
queue_.push_back(new PendingPacket(
new net::DrainableIOBuffer(data.get(), data->size()), done_task));
DoWrite();
return !is_closed();
}
bool BufferedSocketWriter::is_closed() {
return write_callback_.is_null();
}
void BufferedSocketWriter::DoWrite() {
DCHECK(thread_checker_.CalledOnValidThread());
base::WeakPtr<BufferedSocketWriter> self = weak_factory_.GetWeakPtr();
while (self && !write_pending_ && !is_closed() && !queue_.empty()) {
int result = write_callback_.Run(
queue_.front()->data.get(), queue_.front()->data->BytesRemaining(),
base::Bind(&BufferedSocketWriter::OnWritten,
weak_factory_.GetWeakPtr()));
HandleWriteResult(result);
}
}
void BufferedSocketWriter::HandleWriteResult(int result) {
if (result < 0) {
if (result == net::ERR_IO_PENDING) {
write_pending_ = true;
} else {
write_callback_.Reset();
if (!write_failed_callback_.is_null()) {
WriteFailedCallback callback = write_failed_callback_;
callback.Run(result);
}
}
return;
}
DCHECK(!queue_.empty());
queue_.front()->data->DidConsume(result);
if (queue_.front()->data->BytesRemaining() == 0) {
base::Closure done_task = queue_.front()->done_task;
delete queue_.front();
queue_.pop_front();
if (!done_task.is_null())
done_task.Run();
}
}
void BufferedSocketWriter::OnWritten(int result) {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK(write_pending_);
write_pending_ = false;
base::WeakPtr<BufferedSocketWriter> self = weak_factory_.GetWeakPtr();
HandleWriteResult(result);
if (self)
DoWrite();
}
} // namespace remoting
|
