Index: net/http/http_pipelined_connection.cc |
diff --git a/net/http/http_pipelined_connection.cc b/net/http/http_pipelined_connection.cc |
new file mode 100644 |
index 0000000000000000000000000000000000000000..b59c1d37a4fe3e69734a87b7f45176501b4d4d15 |
--- /dev/null |
+++ b/net/http/http_pipelined_connection.cc |
@@ -0,0 +1,478 @@ |
+// Copyright (c) 2011 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/http/http_pipelined_connection.h" |
+ |
+#include "base/message_loop.h" |
+#include "base/stl_util-inl.h" |
+#include "net/base/io_buffer.h" |
+#include "net/http/http_pipelined_host.h" |
+#include "net/http/http_pipelined_stream.h" |
+#include "net/http/http_request_info.h" |
+#include "net/http/http_stream_parser.h" |
+#include "net/socket/client_socket_handle.h" |
+ |
+namespace net { |
+ |
+HttpPipelinedConnection::HttpPipelinedConnection( |
+ ClientSocketHandle* connection, |
+ HttpPipelinedHost* host, |
+ const SSLConfig& used_ssl_config, |
+ const ProxyInfo& used_proxy_info, |
+ const BoundNetLog& net_log, |
+ bool was_npn_negotiated) |
+ : host_(host), |
+ connection_(connection), |
+ used_ssl_config_(used_ssl_config), |
+ used_proxy_info_(used_proxy_info), |
+ net_log_(net_log), |
+ was_npn_negotiated_(was_npn_negotiated), |
+ read_buf_(new GrowableIOBuffer()), |
+ next_pipeline_id_(1), |
+ active_(false), |
+ usable_(true), |
+ ALLOW_THIS_IN_INITIALIZER_LIST(method_factory_(this)), |
+ send_next_state_(SEND_STATE_NONE), |
+ ALLOW_THIS_IN_INITIALIZER_LIST(send_io_callback_( |
+ this, &HttpPipelinedConnection::OnSendIOCallback)), |
+ send_user_callback_(NULL), |
+ read_next_state_(READ_STATE_NONE), |
+ ALLOW_THIS_IN_INITIALIZER_LIST(read_io_callback_( |
+ this, &HttpPipelinedConnection::OnReadIOCallback)), |
+ read_user_callback_(NULL) { |
+ DCHECK(connection_.get()); |
+} |
+ |
+HttpPipelinedConnection::~HttpPipelinedConnection() { |
+ DCHECK(depth() == 0); |
+ DCHECK(parser_map_.empty()); |
+ DCHECK(callback_map_.empty()); |
+ DCHECK(deferred_request_queue_.empty()); |
+ DCHECK(request_order_.empty()); |
+ DCHECK_EQ(send_next_state_, SEND_STATE_NONE); |
+ DCHECK_EQ(read_next_state_, READ_STATE_NONE); |
+ DCHECK(!send_user_callback_); |
+ DCHECK(!read_user_callback_); |
+ if (!usable_) { |
+ connection_->socket()->Disconnect(); |
+ } |
+ connection_->Reset(); |
+} |
+ |
+int HttpPipelinedConnection::AddStream() { |
+ int pipeline_id = next_pipeline_id_++; |
+ DCHECK(pipeline_id); |
+ stream_state_map_.insert(std::make_pair(pipeline_id, STREAM_CREATED)); |
+ return pipeline_id; |
+} |
+ |
+void HttpPipelinedConnection::InitializeParser(int pipeline_id, |
+ const HttpRequestInfo* request, |
+ const BoundNetLog& net_log) { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ DCHECK(!ContainsKey(parser_map_, pipeline_id)); |
+ stream_state_map_[pipeline_id] = STREAM_BOUND; |
+ HttpStreamParser* parser = new HttpStreamParser(connection_.get(), |
+ request, |
+ read_buf_.get(), |
+ net_log); |
+ parser_map_.insert(std::make_pair(pipeline_id, parser)); |
+ if (!active_) { |
+ active_ = true; |
+ MessageLoop::current()->PostTask( |
+ FROM_HERE, |
+ method_factory_.NewRunnableMethod( |
+ &HttpPipelinedConnection::FillPipeline)); |
+ } |
+} |
+ |
+void HttpPipelinedConnection::FillPipeline() { |
+ host_->OnPipelineHasCapacity(this); |
+} |
+ |
+void HttpPipelinedConnection::RemoveStream(int pipeline_id) { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ |
+ if (stream_state_map_[pipeline_id] != STREAM_CREATED) { |
+ DCHECK_EQ(stream_state_map_[pipeline_id], STREAM_CLOSED); |
+ DCHECK(ContainsKey(parser_map_, pipeline_id)); |
+ DCHECK(!ContainsKey(callback_map_, pipeline_id)); |
+ |
+ stream_state_map_.erase(pipeline_id); |
+ ParserMap::iterator it = parser_map_.find(pipeline_id); |
+ delete it->second; |
+ parser_map_.erase(it); |
+ } |
+ |
+ MessageLoop::current()->PostTask( |
+ FROM_HERE, |
+ method_factory_.NewRunnableMethod( |
+ &HttpPipelinedConnection::FillPipeline)); |
+} |
+ |
+int HttpPipelinedConnection::SendRequest(int pipeline_id, |
+ const std::string& request_line, |
+ const HttpRequestHeaders& headers, |
+ UploadDataStream* request_body, |
+ HttpResponseInfo* response, |
+ CompletionCallback* callback) { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ DCHECK_EQ(stream_state_map_[pipeline_id], STREAM_BOUND); |
+ if (!usable_) { |
+ return ERR_PIPELINE_EVICTION; |
+ } |
+ |
+ DeferredSendRequest deferred_request; |
+ deferred_request.pipeline_id = pipeline_id; |
+ deferred_request.request_line = request_line; |
+ deferred_request.headers = headers; |
+ deferred_request.request_body = request_body; |
+ deferred_request.response = response; |
+ deferred_request.callback = callback; |
+ deferred_request_queue_.push(deferred_request); |
+ |
+ if (send_next_state_ == SEND_STATE_NONE) { |
+ send_next_state_ = SEND_STATE_NEXT_REQUEST; |
+ return DoSendRequestLoop(OK); |
+ } else { |
+ return ERR_IO_PENDING; |
+ } |
+} |
+ |
+int HttpPipelinedConnection::DoSendRequestLoop(int result) { |
+ int rv = result; |
+ do { |
+ SendRequestState state = send_next_state_; |
+ send_next_state_ = SEND_STATE_NONE; |
+ switch (state) { |
+ case SEND_STATE_NEXT_REQUEST: |
+ rv = DoSendNextRequest(rv); |
+ break; |
+ case SEND_STATE_COMPLETE: |
+ rv = DoSendComplete(rv); |
+ break; |
+ case SEND_STATE_UNUSABLE: |
+ rv = DoEvictPendingSendRequests(rv); |
+ break; |
+ default: |
+ NOTREACHED() << "bad send state"; |
+ rv = ERR_FAILED; |
+ break; |
+ } |
+ } while (rv != ERR_IO_PENDING && send_next_state_ != SEND_STATE_NONE); |
+ return rv; |
+} |
+ |
+void HttpPipelinedConnection::OnSendIOCallback(int result) { |
+ DCHECK(send_user_callback_); |
+ DoSendRequestLoop(result); |
+} |
+ |
+int HttpPipelinedConnection::DoSendNextRequest(int result) { |
+ DCHECK(!deferred_request_queue_.empty()); |
+ const DeferredSendRequest& deferred_request = deferred_request_queue_.front(); |
+ DCHECK(ContainsKey(stream_state_map_, deferred_request.pipeline_id)); |
+ if (stream_state_map_[deferred_request.pipeline_id] == STREAM_CLOSED) { |
+ deferred_request_queue_.pop(); |
+ if (deferred_request_queue_.empty()) { |
+ send_next_state_ = SEND_STATE_NONE; |
+ } else { |
+ send_next_state_ = SEND_STATE_NEXT_REQUEST; |
+ } |
+ return OK; |
+ } |
+ DCHECK(ContainsKey(parser_map_, deferred_request.pipeline_id)); |
+ int rv = parser_map_[deferred_request.pipeline_id]->SendRequest( |
+ deferred_request.request_line, |
+ deferred_request.headers, |
+ deferred_request.request_body, |
+ deferred_request.response, |
+ &send_io_callback_); |
+ if (result == ERR_IO_PENDING || rv == ERR_IO_PENDING) { |
+ send_user_callback_ = deferred_request.callback; |
+ } |
+ request_order_.push(deferred_request.pipeline_id); |
+ stream_state_map_[deferred_request.pipeline_id] = STREAM_SENT; |
+ deferred_request_queue_.pop(); |
+ send_next_state_ = SEND_STATE_COMPLETE; |
+ return rv; |
+} |
+ |
+int HttpPipelinedConnection::DoSendComplete(int result) { |
+ if (send_user_callback_) { |
+ CompletionCallback* callback = send_user_callback_; |
+ send_user_callback_ = NULL; |
+ callback->Run(result); |
+ } |
+ if (result != OK) { |
+ send_next_state_ = SEND_STATE_UNUSABLE; |
+ usable_ = false; |
+ return result; |
+ } |
+ if (deferred_request_queue_.empty()) { |
+ send_next_state_ = SEND_STATE_NONE; |
+ } else { |
+ send_next_state_ = SEND_STATE_NEXT_REQUEST; |
+ MessageLoop::current()->PostTask( |
+ FROM_HERE, |
+ method_factory_.NewRunnableMethod( |
+ &HttpPipelinedConnection::DoSendRequestLoop, |
+ ERR_IO_PENDING)); |
+ } |
+ return OK; |
+} |
+ |
+int HttpPipelinedConnection::DoEvictPendingSendRequests(int result) { |
+ send_next_state_ = SEND_STATE_NONE; |
+ while (!deferred_request_queue_.empty()) { |
+ const DeferredSendRequest& evicted_send = deferred_request_queue_.front(); |
+ if (stream_state_map_[evicted_send.pipeline_id] == STREAM_CLOSED) { |
+ continue; |
+ } |
+ evicted_send.callback->Run(ERR_PIPELINE_EVICTION); |
+ deferred_request_queue_.pop(); |
+ } |
+ return result; |
+} |
+ |
+int HttpPipelinedConnection::ReadResponseHeaders(int pipeline_id, |
+ CompletionCallback* callback) { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ DCHECK_EQ(stream_state_map_[pipeline_id], STREAM_SENT); |
+ DCHECK(!ContainsKey(callback_map_, pipeline_id)); |
+ if (!usable_) { |
+ return ERR_PIPELINE_EVICTION; |
+ } |
+ callback_map_[pipeline_id] = callback; |
+ if (read_next_state_ == READ_STATE_NONE) { |
+ read_next_state_ = READ_STATE_NEXT_HEADERS; |
+ return DoReadHeadersLoop(OK); |
+ } else { |
+ return ERR_IO_PENDING; |
+ } |
+} |
+ |
+int HttpPipelinedConnection::DoReadHeadersLoop(int result) { |
+ int rv = result; |
+ do { |
+ ReadHeadersState state = read_next_state_; |
+ read_next_state_ = READ_STATE_NONE; |
+ switch (state) { |
+ case READ_STATE_NEXT_HEADERS: |
+ rv = DoReadNextHeaders(rv); |
+ break; |
+ case READ_STATE_COMPLETE: |
+ rv = DoReadHeadersComplete(rv); |
+ break; |
+ case READ_STATE_STREAM_CLOSED: |
+ rv = DoStreamClosed(); |
+ break; |
+ case READ_STATE_UNUSABLE: |
+ rv = DoEvictPendingReadHeaders(rv); |
+ break; |
+ default: |
+ NOTREACHED() << "bad read state"; |
+ rv = ERR_FAILED; |
+ break; |
+ } |
+ } while (rv != ERR_IO_PENDING && read_next_state_ != READ_STATE_NONE); |
+ return rv; |
+} |
+ |
+void HttpPipelinedConnection::OnReadIOCallback(int result) { |
+ DCHECK(read_user_callback_ != NULL); |
+ DoReadHeadersLoop(result); |
+} |
+ |
+int HttpPipelinedConnection::DoReadNextHeaders(int result) { |
+ DCHECK(!request_order_.empty()); |
+ int pipeline_id = request_order_.front(); |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ if (stream_state_map_[pipeline_id] == STREAM_CLOSED) { |
+ // Since nobody will read whatever data is on the pipeline associated with |
+ // this request, we must shut down the rest of the pipeline. |
+ read_next_state_ = READ_STATE_UNUSABLE; |
+ return OK; |
+ } |
+ CallbackMap::iterator it = callback_map_.find(pipeline_id); |
+ if (it == callback_map_.end()) { |
+ return ERR_IO_PENDING; |
+ } |
+ DCHECK(ContainsKey(parser_map_, pipeline_id)); |
+ int rv = parser_map_[pipeline_id]->ReadResponseHeaders( |
+ &read_io_callback_); |
+ if (rv == ERR_IO_PENDING) { |
+ read_next_state_ = READ_STATE_COMPLETE; |
+ read_user_callback_ = it->second; |
+ } else if (result == ERR_IO_PENDING) { |
+ read_next_state_ = READ_STATE_STREAM_CLOSED; |
+ MessageLoop::current()->PostTask( |
+ FROM_HERE, |
+ method_factory_.NewRunnableMethod( |
+ &HttpPipelinedConnection::FireUserCallback, |
+ it->second, |
+ rv)); |
+ rv = ERR_IO_PENDING; |
+ } |
+ callback_map_.erase(it); |
+ return rv; |
+} |
+ |
+int HttpPipelinedConnection::DoReadHeadersComplete(int result) { |
+ read_next_state_ = READ_STATE_STREAM_CLOSED; |
+ if (read_user_callback_) { |
+ MessageLoop::current()->PostTask( |
+ FROM_HERE, |
+ method_factory_.NewRunnableMethod( |
+ &HttpPipelinedConnection::FireUserCallback, |
+ read_user_callback_, |
+ result)); |
+ read_user_callback_ = NULL; |
+ } |
+ return ERR_IO_PENDING; |
+} |
+ |
+void HttpPipelinedConnection::FireUserCallback(CompletionCallback* callback, |
+ int result) { |
+ DCHECK(callback); |
+ callback->Run(result); |
+} |
+ |
+int HttpPipelinedConnection::DoStreamClosed() { |
+ DCHECK(!request_order_.empty()); |
+ request_order_.pop(); |
+ if (request_order_.empty()) { |
+ read_next_state_ = READ_STATE_NONE; |
+ return OK; |
+ } else { |
+ read_next_state_ = READ_STATE_NEXT_HEADERS; |
+ MessageLoop::current()->PostTask( |
+ FROM_HERE, |
+ method_factory_.NewRunnableMethod( |
+ &HttpPipelinedConnection::DoReadHeadersLoop, |
+ ERR_IO_PENDING)); |
+ return ERR_IO_PENDING; // Wait for the task to fire. |
+ } |
+} |
+ |
+int HttpPipelinedConnection::DoEvictPendingReadHeaders(int result) { |
+ while (!request_order_.empty()) { |
+ int evicted_id = request_order_.front(); |
+ request_order_.pop(); |
+ CallbackMap::iterator cb_it = callback_map_.find(evicted_id); |
+ if (cb_it == callback_map_.end()) { |
+ continue; |
+ } |
+ callback_map_.erase(cb_it); |
+ if (stream_state_map_[evicted_id] == STREAM_CLOSED) { |
+ continue; |
+ } |
+ cb_it->second->Run(ERR_PIPELINE_EVICTION); |
+ } |
+ DCHECK(callback_map_.empty()); |
+ read_next_state_ = READ_STATE_NONE; |
+ return result; |
+} |
+ |
+void HttpPipelinedConnection::Close(int pipeline_id, |
+ bool not_reusable) { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ if (stream_state_map_[pipeline_id] == STREAM_CLOSED) { |
+ // TODO(simonjam): Why is Close() sometimes called twice? |
+ return; |
+ } |
+ stream_state_map_[pipeline_id] = STREAM_CLOSED; |
+ // TODO(simonjam): If it hasn't sent yet, then we can just ignore it instead |
+ // of shutting the pipeline down. |
+ bool is_active_stream = !request_order_.empty() && |
+ pipeline_id == request_order_.front(); |
+ if (not_reusable || !is_active_stream) { |
+ usable_ = false; |
+ } |
+ if (is_active_stream) { |
+ read_next_state_ = READ_STATE_STREAM_CLOSED; |
+ read_user_callback_ = NULL; |
+ DoReadHeadersLoop(OK); |
+ } else { |
+ if (send_next_state_ == SEND_STATE_NONE) { |
+ send_next_state_ = SEND_STATE_UNUSABLE; |
+ DoSendRequestLoop(OK); |
+ } |
+ if (read_next_state_ == READ_STATE_NONE) { |
+ read_next_state_ = READ_STATE_UNUSABLE; |
+ DoReadHeadersLoop(OK); |
+ } |
+ } |
+} |
+ |
+int HttpPipelinedConnection::ReadResponseBody(int pipeline_id, |
+ IOBuffer* buf, int buf_len, |
+ CompletionCallback* callback) { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ DCHECK(!request_order_.empty()); |
+ DCHECK(pipeline_id == request_order_.front()); |
+ DCHECK(ContainsKey(parser_map_, pipeline_id)); |
+ return parser_map_[pipeline_id]->ReadResponseBody(buf, buf_len, callback); |
+} |
+ |
+uint64 HttpPipelinedConnection::GetUploadProgress(int pipeline_id) const { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ DCHECK(ContainsKey(parser_map_, pipeline_id)); |
+ return parser_map_.find(pipeline_id)->second->GetUploadProgress(); |
+} |
+ |
+HttpResponseInfo* HttpPipelinedConnection::GetResponseInfo(int pipeline_id) { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ DCHECK(ContainsKey(parser_map_, pipeline_id)); |
+ return parser_map_[pipeline_id]->GetResponseInfo(); |
+} |
+ |
+bool HttpPipelinedConnection::IsResponseBodyComplete(int pipeline_id) const { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ DCHECK(ContainsKey(parser_map_, pipeline_id)); |
+ return parser_map_.find(pipeline_id)->second->IsResponseBodyComplete(); |
+} |
+ |
+bool HttpPipelinedConnection::CanFindEndOfResponse(int pipeline_id) const { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ DCHECK(ContainsKey(parser_map_, pipeline_id)); |
+ return parser_map_.find(pipeline_id)->second->CanFindEndOfResponse(); |
+} |
+ |
+bool HttpPipelinedConnection::IsMoreDataBuffered(int pipeline_id) const { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ return read_buf_->offset(); |
+} |
+ |
+bool HttpPipelinedConnection::IsConnectionReused(int pipeline_id) const { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ if (pipeline_id > 1) { |
+ return true; |
+ } |
+ ClientSocketHandle::SocketReuseType reuse_type = connection_->reuse_type(); |
+ return connection_->is_reused() || |
+ reuse_type == ClientSocketHandle::UNUSED_IDLE; |
+} |
+ |
+void HttpPipelinedConnection::SetConnectionReused(int pipeline_id) { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ connection_->set_is_reused(true); |
+} |
+ |
+void HttpPipelinedConnection::GetSSLInfo(int pipeline_id, |
+ SSLInfo* ssl_info) { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ DCHECK(ContainsKey(parser_map_, pipeline_id)); |
+ return parser_map_[pipeline_id]->GetSSLInfo(ssl_info); |
+} |
+ |
+void HttpPipelinedConnection::GetSSLCertRequestInfo( |
+ int pipeline_id, |
+ SSLCertRequestInfo* cert_request_info) { |
+ DCHECK(ContainsKey(stream_state_map_, pipeline_id)); |
+ DCHECK(ContainsKey(parser_map_, pipeline_id)); |
+ return parser_map_[pipeline_id]->GetSSLCertRequestInfo(cert_request_info); |
+} |
+ |
+} // namespace net |