1 // Copyright 2014 The Chromium Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
5 #include "mojo/services/network/tcp_connected_socket_impl.h"
7 #include "base/message_loop/message_loop.h"
8 #include "mojo/services/network/net_adapters.h"
9 #include "net/base/net_errors.h"
13 TCPConnectedSocketImpl::TCPConnectedSocketImpl(
14 scoped_ptr<net::TCPSocket> socket,
15 ScopedDataPipeConsumerHandle send_stream,
16 ScopedDataPipeProducerHandle receive_stream)
17 : socket_(socket.Pass()),
18 send_stream_(send_stream.Pass()),
19 receive_stream_(receive_stream.Pass()),
20 weak_ptr_factory_(this) {
21 // Queue up async communication.
26 TCPConnectedSocketImpl::~TCPConnectedSocketImpl() {
29 void TCPConnectedSocketImpl::ReceiveMore() {
30 DCHECK(!pending_receive_.get());
33 MojoResult result = NetToMojoPendingBuffer::BeginWrite(
34 &receive_stream_, &pending_receive_, &num_bytes);
35 if (result == MOJO_RESULT_SHOULD_WAIT) {
36 // The pipe is full. We need to wait for it to have more space.
37 receive_handle_watcher_.Start(
38 receive_stream_.get(),
39 MOJO_HANDLE_SIGNAL_WRITABLE, MOJO_DEADLINE_INDEFINITE,
40 base::Bind(&TCPConnectedSocketImpl::OnReceiveStreamReady,
41 weak_ptr_factory_.GetWeakPtr()));
43 } else if (result != MOJO_RESULT_OK) {
44 // The receive stream is in a bad state.
45 // TODO(darin): How should this be communicated to our client?
50 // Mojo is ready for the receive.
51 CHECK_GT(static_cast<uint32_t>(std::numeric_limits<int>::max()), num_bytes);
52 scoped_refptr<net::IOBuffer> buf(
53 new NetToMojoIOBuffer(pending_receive_.get()));
54 int read_result = socket_->Read(
55 buf.get(), static_cast<int>(num_bytes),
56 base::Bind(&TCPConnectedSocketImpl::DidReceive, base::Unretained(this),
58 if (read_result == net::ERR_IO_PENDING) {
59 // Pending I/O, wait for result in DidReceive().
60 } else if (read_result >= 0) {
61 // Synchronous data ready.
62 DidReceive(true, read_result);
64 // Some kind of error.
65 // TODO(brettw) notify caller of error.
70 void TCPConnectedSocketImpl::OnReceiveStreamReady(MojoResult result) {
71 // TODO(darin): Handle a bad |result| value.
75 void TCPConnectedSocketImpl::DidReceive(bool completed_synchronously,
79 pending_receive_ = NULL; // Closes the pipe (owned by the pending write).
80 // TODO(brettw) notify the caller of an error?
85 receive_stream_ = pending_receive_->Complete(result);
86 pending_receive_ = NULL;
88 // Schedule more reading.
89 if (completed_synchronously) {
90 // Don't recursively call ReceiveMore if this is a sync read.
91 base::MessageLoop::current()->PostTask(
93 base::Bind(&TCPConnectedSocketImpl::ReceiveMore,
94 weak_ptr_factory_.GetWeakPtr()));
100 void TCPConnectedSocketImpl::SendMore() {
101 uint32_t num_bytes = 0;
102 MojoResult result = MojoToNetPendingBuffer::BeginRead(
103 &send_stream_, &pending_send_, &num_bytes);
104 if (result == MOJO_RESULT_SHOULD_WAIT) {
105 // Data not ready, wait for it.
106 send_handle_watcher_.Start(
108 MOJO_HANDLE_SIGNAL_READABLE, MOJO_DEADLINE_INDEFINITE,
109 base::Bind(&TCPConnectedSocketImpl::OnSendStreamReady,
110 weak_ptr_factory_.GetWeakPtr()));
112 } else if (result != MOJO_RESULT_OK) {
113 // TODO(brettw) notify caller of error.
118 // Got a buffer from Mojo, give it to the socket. Note that the sockets may
119 // do partial writes.
120 scoped_refptr<net::IOBuffer> buf(new MojoToNetIOBuffer(pending_send_.get()));
121 int write_result = socket_->Write(
122 buf.get(), static_cast<int>(num_bytes),
123 base::Bind(&TCPConnectedSocketImpl::DidSend, base::Unretained(this),
125 if (write_result == net::ERR_IO_PENDING) {
126 // Pending I/O, wait for result in DidSend().
127 } else if (write_result >= 0) {
128 // Synchronous data consumed.
129 DidSend(true, write_result);
133 void TCPConnectedSocketImpl::OnSendStreamReady(MojoResult result) {
134 // TODO(brettw): Handle a bad |result| value.
138 void TCPConnectedSocketImpl::DidSend(bool completed_synchronously,
141 // TODO(brettw) report error.
142 pending_send_ = NULL;
147 // Take back ownership of the stream and free the IOBuffer.
148 send_stream_ = pending_send_->Complete(result);
149 pending_send_ = NULL;
151 // Schedule more writing.
152 if (completed_synchronously) {
153 // Don't recursively call SendMore if this is a sync read.
154 base::MessageLoop::current()->PostTask(
156 base::Bind(&TCPConnectedSocketImpl::SendMore,
157 weak_ptr_factory_.GetWeakPtr()));