blob: 3b3a6ce55e15e79c9ac8beec738438f613e05587 (
plain)
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
|
// Copyright (c) 2012 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 "net/websockets/websocket_throttle.h"
#include <algorithm>
#include <set>
#include <string>
#include "base/memory/singleton.h"
#include "base/message_loop.h"
#include "base/string_number_conversions.h"
#include "base/string_util.h"
#include "base/stringprintf.h"
#include "net/base/io_buffer.h"
#include "net/socket_stream/socket_stream.h"
#include "net/websockets/websocket_job.h"
namespace net {
WebSocketThrottle::WebSocketThrottle() {
}
WebSocketThrottle::~WebSocketThrottle() {
DCHECK(queue_.empty());
DCHECK(addr_map_.empty());
}
// static
WebSocketThrottle* WebSocketThrottle::GetInstance() {
return Singleton<WebSocketThrottle>::get();
}
void WebSocketThrottle::PutInQueue(WebSocketJob* job) {
queue_.push_back(job);
const AddressList& address_list = job->address_list();
std::set<IPEndPoint> address_set;
for (AddressList::const_iterator addr_iter = address_list.begin();
addr_iter != address_list.end();
++addr_iter) {
const IPEndPoint& address = *addr_iter;
// If |address| is already processed, don't do it again.
if (!address_set.insert(address).second)
continue;
ConnectingAddressMap::iterator iter = addr_map_.find(address);
if (iter == addr_map_.end()) {
ConnectingQueue* queue = new ConnectingQueue();
queue->push_back(job);
addr_map_[address] = queue;
} else {
iter->second->push_back(job);
job->SetWaiting();
DVLOG(1) << "Waiting on " << address.ToString();
}
}
}
void WebSocketThrottle::RemoveFromQueue(WebSocketJob* job) {
ConnectingQueue::iterator queue_iter =
std::find(queue_.begin(), queue_.end(), job);
if (queue_iter == queue_.end())
return;
queue_.erase(queue_iter);
const AddressList& address_list = job->address_list();
std::set<IPEndPoint> address_set;
for (AddressList::const_iterator addr_iter = address_list.begin();
addr_iter != address_list.end();
++addr_iter) {
const IPEndPoint& address = *addr_iter;
// If |address| is already processed, don't do it again.
if (!address_set.insert(address).second)
continue;
ConnectingAddressMap::iterator map_iter = addr_map_.find(address);
DCHECK(map_iter != addr_map_.end());
ConnectingQueue* queue = map_iter->second;
// Job may not be front of queue if the socket is closed while waiting.
ConnectingQueue::iterator address_queue_iter =
std::find(queue->begin(), queue->end(), job);
if (address_queue_iter != queue->end())
queue->erase(address_queue_iter);
if (queue->empty()) {
delete queue;
addr_map_.erase(map_iter);
}
}
}
void WebSocketThrottle::WakeupSocketIfNecessary() {
for (ConnectingQueue::iterator iter = queue_.begin();
iter != queue_.end();
++iter) {
WebSocketJob* job = *iter;
if (!job->IsWaiting())
continue;
bool should_wakeup = true;
const AddressList& address_list = job->address_list();
for (AddressList::const_iterator addr_iter = address_list.begin();
addr_iter != address_list.end();
++addr_iter) {
const IPEndPoint& address = *addr_iter;
ConnectingAddressMap::iterator map_iter = addr_map_.find(address);
DCHECK(map_iter != addr_map_.end());
ConnectingQueue* queue = map_iter->second;
if (job != queue->front()) {
should_wakeup = false;
break;
}
}
if (should_wakeup)
job->Wakeup();
}
}
} // namespace net
|
