base::Bind: Convert Socket::Read.

net/socket/web_socket_server_socket.cc

 // 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/socket/web_socket_server_socket.h"
 
 #include <algorithm>
 #include <deque>
 #include <limits>
 #include <map>
@@ skipping 117 matching lines @@
     DCHECK(transport_socket);
     DCHECK(delegate);
   }
 
   virtual ~WebSocketServerSocketImpl() {
     std::deque<PendingReq>::iterator it = GetPendingReq(PendingReq::TYPE_READ);
     if (it != pending_reqs_.end() &&
         it->type == PendingReq::TYPE_READ &&
         it->io_buf != NULL &&
         it->io_buf->data() != NULL &&
-        it->callback != 0) {
-      it->callback->Run(0);  // Report EOF.
+        (it->old_callback || !it->callback.is_null())) {
+      if (it->old_callback)
+        it->old_callback->Run(0);  // Report EOF.
+      else
+        it->callback.Run(0);
     }
   }
 
  private:
   enum Phase {
     // Before Accept() is called.
     PHASE_NYMPH,
 
     // After Accept() is called and until handshake success/fail.
     PHASE_HANDSHAKE,
@@ skipping 18 matching lines @@
       TYPE_WRITE          = 1 << 2,
 
       TYPE_READ_METADATA  = TYPE_READ | TYPE_METADATA,
       TYPE_WRITE_METADATA = TYPE_WRITE | TYPE_METADATA
     };
 
     PendingReq(Type type, net::DrainableIOBuffer* io_buf,
                net::OldCompletionCallback* callback)
         : type(type),
           io_buf(io_buf),
+          old_callback(callback) {
+      switch (type) {
+        case PendingReq::TYPE_READ:
+        case PendingReq::TYPE_WRITE:
+        case PendingReq::TYPE_READ_METADATA:
+        case PendingReq::TYPE_WRITE_METADATA: {
+          DCHECK(io_buf);
+          break;
+        }
+        default: {
+          NOTREACHED();
+          break;
+        }
+      }
+    }
+    PendingReq(Type type, net::DrainableIOBuffer* io_buf,
+               const net::CompletionCallback& callback)
+        : type(type),
+          io_buf(io_buf),
+          old_callback(NULL),
           callback(callback) {
       switch (type) {
         case PendingReq::TYPE_READ:
         case PendingReq::TYPE_WRITE:
         case PendingReq::TYPE_READ_METADATA:
         case PendingReq::TYPE_WRITE_METADATA: {
           DCHECK(io_buf);
           break;
         }
         default: {
           NOTREACHED();
           break;
         }
       }
     }
 
     Type type;
     scoped_refptr<net::DrainableIOBuffer> io_buf;
-    net::OldCompletionCallback* callback;
+    net::OldCompletionCallback* old_callback;
+    net::CompletionCallback callback;
   };
 
   // Socket implementation.
   virtual int Read(net::IOBuffer* buf, int buf_len,
                    net::OldCompletionCallback* callback) OVERRIDE {
     if (buf_len == 0)
       return 0;
     if (buf == NULL || buf_len < 0) {
       NOTREACHED();
       return net::ERR_INVALID_ARGUMENT;
@@ skipping 47 matching lines @@
       }
       case PHASE_NYMPH:
       case PHASE_HANDSHAKE:
       default: {
         NOTREACHED();
         return net::ERR_UNEXPECTED;
       }
     }
     return net::ERR_IO_PENDING;
   }
+  virtual int Read(net::IOBuffer* buf, int buf_len,
+                   const net::CompletionCallback& callback) OVERRIDE {
+    if (buf_len == 0)
+      return 0;
+    if (buf == NULL || buf_len < 0) {
+      NOTREACHED();
+      return net::ERR_INVALID_ARGUMENT;
+    }
+    while (int bytes_remaining = fill_handshake_buf_->BytesConsumed() -
+        process_handshake_buf_->BytesConsumed()) {
+      DCHECK(!is_transport_read_pending_);
+      DCHECK(GetPendingReq(PendingReq::TYPE_READ) == pending_reqs_.end());
+      switch (phase_) {
+        case PHASE_FRAME_OUTSIDE:
+        case PHASE_FRAME_INSIDE:
+        case PHASE_FRAME_LENGTH:
+        case PHASE_FRAME_SKIP: {
+          int n = std::min(bytes_remaining, buf_len);
+          int rv = ProcessDataFrames(
+              process_handshake_buf_->data(), n, buf->data(), buf_len);
+          process_handshake_buf_->DidConsume(n);
+          if (rv == 0) {
+            // ProcessDataFrames may return zero for non-empty buffer if it
+            // contains only frame delimiters without real data.  In this case:
+            // try again and do not just return zero (zero stands for EOF).
+            continue;
+          }
+          return rv;
+        }
+        case PHASE_SHUT: {
+          return 0;
+        }
+        case PHASE_NYMPH:
+        case PHASE_HANDSHAKE:
+        default: {
+          NOTREACHED();
+          return net::ERR_UNEXPECTED;
+        }
+      }
+    }
+    switch (phase_) {
+      case PHASE_FRAME_OUTSIDE:
+      case PHASE_FRAME_INSIDE:
+      case PHASE_FRAME_LENGTH:
+      case PHASE_FRAME_SKIP: {
+        pending_reqs_.push_back(PendingReq(
+            PendingReq::TYPE_READ,
+            new net::DrainableIOBuffer(buf, buf_len),
+            callback));
+        ConsiderTransportRead();
+        break;
+      }
+      case PHASE_SHUT: {
+        return 0;
+      }
+      case PHASE_NYMPH:
+      case PHASE_HANDSHAKE:
+      default: {
+        NOTREACHED();
+        return net::ERR_UNEXPECTED;
+      }
+    }
+    return net::ERR_IO_PENDING;
+  }
 
   virtual int Write(net::IOBuffer* buf, int buf_len,
                     net::OldCompletionCallback* callback) OVERRIDE {
     if (buf_len == 0)
       return 0;
     if (buf == NULL || buf_len < 0) {
       NOTREACHED();
       return net::ERR_INVALID_ARGUMENT;
     }
     DCHECK_EQ(std::find(buf->data(), buf->data() + buf_len, '\xff'),
@@ skipping 116 matching lines @@
               &WebSocketServerSocketImpl::OnWrite, rv));
     }
   }
 
   void Shut(int result) {
     if (result > 0 || result == net::ERR_IO_PENDING)
       result = net::ERR_UNEXPECTED;
     if (result != 0) {
       while (!pending_reqs_.empty()) {
         PendingReq& req = pending_reqs_.front();
-        if (req.callback)
-          req.callback->Run(result);
+        if (req.old_callback)
+          req.old_callback->Run(result);
+        else if (!req.callback.is_null())
+          req.callback.Run(result);
         pending_reqs_.pop_front();
       }
       transport_socket_.reset();  // terminate underlying connection.
     }
     phase_ = PHASE_SHUT;
   }
 
   // Callbacks for transport socket.
   void OnRead(int result) {
     if (!is_transport_read_pending_) {
@@ skipping 28 matching lines @@
           NOTREACHED();
           Shut(net::ERR_UNEXPECTED);
           return;
         }
         fill_handshake_buf_->DidConsume(result);
         // ProcessHandshake invalidates iterators for |pending_reqs_|
         int rv = ProcessHandshake();
         if (rv > 0) {
           process_handshake_buf_->DidConsume(rv);
           phase_ = PHASE_FRAME_OUTSIDE;
-          net::OldCompletionCallback* cb = pending_reqs_.front().callback;
+          net::OldCompletionCallback* old_cb =
+              pending_reqs_.front().old_callback;
+          net::CompletionCallback cb = pending_reqs_.front().callback;
           pending_reqs_.pop_front();
           ConsiderTransportWrite();  // Schedule answer handshake.
-          if (cb)
-            cb->Run(0);
+          if (old_cb)
+            old_cb->Run(0);
+          else if (!cb.is_null())
+            cb.Run(0);
         } else if (rv == net::ERR_IO_PENDING) {
           if (fill_handshake_buf_->BytesRemaining() < 1)
             Shut(net::ERR_LIMIT_VIOLATION);
         } else if (rv < 0) {
           Shut(rv);
         } else {
           Shut(net::ERR_UNEXPECTED);
         }
         break;
       }
       case PHASE_FRAME_OUTSIDE:
       case PHASE_FRAME_INSIDE:
       case PHASE_FRAME_LENGTH:
       case PHASE_FRAME_SKIP: {
         int rv = ProcessDataFrames(
             it->io_buf->data(), result,
             it->io_buf->data(), it->io_buf->BytesRemaining());
         if (rv < 0) {
           Shut(rv);
           return;
         }
         if (rv > 0 || phase_ == PHASE_SHUT) {
-          net::OldCompletionCallback* cb = it->callback;
+          net::OldCompletionCallback* old_cb = it->old_callback;
+          net::CompletionCallback cb = it->callback;
           pending_reqs_.erase(it);
-          if (cb)
-            cb->Run(rv);
+          if (old_cb)
+            old_cb->Run(rv);
+          else if (!cb.is_null())
+            cb.Run(rv);
         }
         break;
       }
       case PHASE_NYMPH:
       default: {
         NOTREACHED();
         Shut(net::ERR_UNEXPECTED);
         break;
       }
     }
@@ skipping 17 matching lines @@
     if (it == pending_reqs_.end() ||
         it->io_buf == NULL ||
         it->io_buf->data() == NULL) {
       NOTREACHED();
       Shut(net::ERR_UNEXPECTED);
       return;
     }
     DCHECK_LE(result, it->io_buf->BytesRemaining());
     it->io_buf->DidConsume(result);
     if (it->io_buf->BytesRemaining() == 0) {
-      net::OldCompletionCallback* cb = it->callback;
+      net::OldCompletionCallback* old_cb = it->old_callback;
+      net::CompletionCallback cb = it->callback;
       int bytes_written = it->io_buf->BytesConsumed();
       DCHECK_GT(bytes_written, 0);
       pending_reqs_.erase(it);
-      if (cb)
-        cb->Run(bytes_written);
+      if (old_cb)
+        old_cb->Run(bytes_written);
+      else if (!cb.is_null())
+        cb.Run(bytes_written);
     }
     ConsiderTransportWrite();
   }
 
   // Returns (positive) number of consumed bytes on success.
   // Returns ERR_IO_PENDING in case of incomplete input.
   // Returns ERR_WS_PROTOCOL_ERROR or ERR_LIMIT_VIOLATION in case of failure to
   // reasonably parse input.
   int ProcessHandshake() {
     static const char kGetPrefix[] = "GET ";
@@ skipping 362 matching lines @@
 
 WebSocketServerSocket* CreateWebSocketServerSocket(
     Socket* transport_socket, WebSocketServerSocket::Delegate* delegate) {
   return new WebSocketServerSocketImpl(transport_socket, delegate);
 }
 
 WebSocketServerSocket::~WebSocketServerSocket() {
 }
 
 }  // namespace net;