3 * Copyright 2004--2005, Google Inc.
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions are met:
8 * 1. Redistributions of source code must retain the above copyright notice,
9 * this list of conditions and the following disclaimer.
10 * 2. Redistributions in binary form must reproduce the above copyright notice,
11 * this list of conditions and the following disclaimer in the documentation
12 * and/or other materials provided with the distribution.
13 * 3. The name of the author may not be used to endorse or promote products
14 * derived from this software without specific prior written permission.
16 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
17 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
18 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
19 * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
20 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
21 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
22 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
23 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
24 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
25 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
30 #include "talk/base/asyncsocket.h"
31 #include "talk/base/logging.h"
32 #include "talk/base/socketfactory.h"
33 #include "talk/base/socketpool.h"
34 #include "talk/base/socketstream.h"
35 #include "talk/base/thread.h"
39 ///////////////////////////////////////////////////////////////////////////////
40 // StreamCache - Caches a set of open streams, defers creation to a separate
42 ///////////////////////////////////////////////////////////////////////////////
44 StreamCache::StreamCache(StreamPool* pool) : pool_(pool) {
47 StreamCache::~StreamCache() {
48 for (ConnectedList::iterator it = active_.begin(); it != active_.end();
52 for (ConnectedList::iterator it = cached_.begin(); it != cached_.end();
58 StreamInterface* StreamCache::RequestConnectedStream(
59 const SocketAddress& remote, int* err) {
60 LOG_F(LS_VERBOSE) << "(" << remote << ")";
61 for (ConnectedList::iterator it = cached_.begin(); it != cached_.end();
63 if (remote == it->first) {
64 it->second->SignalEvent.disconnect(this);
65 // Move from cached_ to active_
66 active_.push_front(*it);
70 LOG_F(LS_VERBOSE) << "Providing cached stream";
71 return active_.front().second;
74 if (StreamInterface* stream = pool_->RequestConnectedStream(remote, err)) {
75 // We track active streams so that we can remember their address
76 active_.push_front(ConnectedStream(remote, stream));
77 LOG_F(LS_VERBOSE) << "Providing new stream";
78 return active_.front().second;
83 void StreamCache::ReturnConnectedStream(StreamInterface* stream) {
84 for (ConnectedList::iterator it = active_.begin(); it != active_.end();
86 if (stream == it->second) {
87 LOG_F(LS_VERBOSE) << "(" << it->first << ")";
88 if (stream->GetState() == SS_CLOSED) {
89 // Return closed streams
90 LOG_F(LS_VERBOSE) << "Returning closed stream";
91 pool_->ReturnConnectedStream(it->second);
93 // Monitor open streams
94 stream->SignalEvent.connect(this, &StreamCache::OnStreamEvent);
95 LOG_F(LS_VERBOSE) << "Caching stream";
96 cached_.push_front(*it);
105 void StreamCache::OnStreamEvent(StreamInterface* stream, int events, int err) {
106 if ((events & SE_CLOSE) == 0) {
107 LOG_F(LS_WARNING) << "(" << events << ", " << err
108 << ") received non-close event";
111 for (ConnectedList::iterator it = cached_.begin(); it != cached_.end();
113 if (stream == it->second) {
114 LOG_F(LS_VERBOSE) << "(" << it->first << ")";
115 // We don't cache closed streams, so return it.
116 it->second->SignalEvent.disconnect(this);
117 LOG_F(LS_VERBOSE) << "Returning closed stream";
118 pool_->ReturnConnectedStream(it->second);
126 //////////////////////////////////////////////////////////////////////
128 //////////////////////////////////////////////////////////////////////
130 NewSocketPool::NewSocketPool(SocketFactory* factory) : factory_(factory) {
133 NewSocketPool::~NewSocketPool() {
137 NewSocketPool::RequestConnectedStream(const SocketAddress& remote, int* err) {
138 AsyncSocket* socket =
139 factory_->CreateAsyncSocket(remote.family(), SOCK_STREAM);
145 if ((socket->Connect(remote) != 0) && !socket->IsBlocking()) {
147 *err = socket->GetError();
153 return new SocketStream(socket);
157 NewSocketPool::ReturnConnectedStream(StreamInterface* stream) {
158 Thread::Current()->Dispose(stream);
161 //////////////////////////////////////////////////////////////////////
163 //////////////////////////////////////////////////////////////////////
165 ReuseSocketPool::ReuseSocketPool(SocketFactory* factory)
166 : factory_(factory), stream_(NULL), checked_out_(false) {
169 ReuseSocketPool::~ReuseSocketPool() {
170 ASSERT(!checked_out_);
175 ReuseSocketPool::RequestConnectedStream(const SocketAddress& remote, int* err) {
176 // Only one socket can be used from this "pool" at a time
177 ASSERT(!checked_out_);
179 LOG_F(LS_VERBOSE) << "Creating new socket";
180 int family = remote.family();
181 // TODO: Deal with this when we/I clean up DNS resolution.
182 if (remote.IsUnresolvedIP()) {
185 AsyncSocket* socket =
186 factory_->CreateAsyncSocket(family, SOCK_STREAM);
192 stream_ = new SocketStream(socket);
194 if ((stream_->GetState() == SS_OPEN) && (remote == remote_)) {
195 LOG_F(LS_VERBOSE) << "Reusing connection to: " << remote_;
199 if ((stream_->GetSocket()->Connect(remote_) != 0)
200 && !stream_->GetSocket()->IsBlocking()) {
202 *err = stream_->GetSocket()->GetError();
205 LOG_F(LS_VERBOSE) << "Opening connection to: " << remote_;
208 stream_->SignalEvent.disconnect(this);
216 ReuseSocketPool::ReturnConnectedStream(StreamInterface* stream) {
217 ASSERT(stream == stream_);
218 ASSERT(checked_out_);
219 checked_out_ = false;
220 // Until the socket is reused, monitor it to determine if it closes.
221 stream_->SignalEvent.connect(this, &ReuseSocketPool::OnStreamEvent);
225 ReuseSocketPool::OnStreamEvent(StreamInterface* stream, int events, int err) {
226 ASSERT(stream == stream_);
227 ASSERT(!checked_out_);
229 // If the stream was written to and then immediately returned to us then
230 // we may get a writable notification for it, which we should ignore.
231 if (events == SE_WRITE) {
232 LOG_F(LS_VERBOSE) << "Pooled Socket unexpectedly writable: ignoring";
236 // If the peer sent data, we can't process it, so drop the connection.
237 // If the socket has closed, clean it up.
238 // In either case, we'll reconnect it the next time it is used.
239 ASSERT(0 != (events & (SE_READ|SE_CLOSE)));
240 if (0 != (events & SE_CLOSE)) {
241 LOG_F(LS_VERBOSE) << "Connection closed with error: " << err;
243 LOG_F(LS_VERBOSE) << "Pooled Socket unexpectedly readable: closing";
248 ///////////////////////////////////////////////////////////////////////////////
249 // LoggingPoolAdapter - Adapts a StreamPool to supply streams with attached
251 ///////////////////////////////////////////////////////////////////////////////
253 LoggingPoolAdapter::LoggingPoolAdapter(
254 StreamPool* pool, LoggingSeverity level, const std::string& label,
256 : pool_(pool), level_(level), label_(label), binary_mode_(binary_mode) {
259 LoggingPoolAdapter::~LoggingPoolAdapter() {
260 for (StreamList::iterator it = recycle_bin_.begin();
261 it != recycle_bin_.end(); ++it) {
266 StreamInterface* LoggingPoolAdapter::RequestConnectedStream(
267 const SocketAddress& remote, int* err) {
268 if (StreamInterface* stream = pool_->RequestConnectedStream(remote, err)) {
269 ASSERT(SS_CLOSED != stream->GetState());
270 std::stringstream ss;
271 ss << label_ << "(0x" << std::setfill('0') << std::hex << std::setw(8)
273 LOG_V(level_) << ss.str()
274 << ((SS_OPEN == stream->GetState()) ? " Connected"
277 if (recycle_bin_.empty()) {
278 return new LoggingAdapter(stream, level_, ss.str(), binary_mode_);
280 LoggingAdapter* logging = recycle_bin_.front();
281 recycle_bin_.pop_front();
282 logging->set_label(ss.str());
283 logging->Attach(stream);
289 void LoggingPoolAdapter::ReturnConnectedStream(StreamInterface* stream) {
290 LoggingAdapter* logging = static_cast<LoggingAdapter*>(stream);
291 pool_->ReturnConnectedStream(logging->Detach());
292 recycle_bin_.push_back(logging);
295 ///////////////////////////////////////////////////////////////////////////////
297 } // namespace talk_base