[WebSocket] bufferedAmount should not decrease inside a task.

Source/modules/websockets/MainThreadWebSocketChannel.cpp

 /*
  * Copyright (C) 2011, 2012 Google Inc.  All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are
  * met:
  *
  *     * Redistributions of source code must retain the above copyright
  * notice, this list of conditions and the following disclaimer.
  *     * Redistributions in binary form must reproduce the above
@@ skipping 54 matching lines @@
     : m_document(document)
     , m_client(client)
     , m_resumeTimer(this, &MainThreadWebSocketChannel::resumeTimerFired)
     , m_suspended(false)
     , m_didFailOfClientAlreadyRun(false)
     , m_hasCalledDisconnectOnHandle(false)
     , m_receivedClosingHandshake(false)
     , m_closingTimer(this, &MainThreadWebSocketChannel::closingTimerFired)
     , m_state(ChannelIdle)
     , m_shouldDiscardReceivedData(false)
-    , m_unhandledBufferedAmount(0)
     , m_identifier(0)
     , m_hasContinuousFrame(false)
     , m_closeEventCode(CloseEventCodeAbnormalClosure)
     , m_outgoingFrameQueueStatus(OutgoingFrameQueueOpen)
+    , m_numConsumedBytesInCurrentFrame(0)
     , m_blobLoaderStatus(BlobLoaderNotStarted)
     , m_sourceURLAtConstruction(sourceURL)
     , m_lineNumberAtConstruction(lineNumber)
 {
     if (m_document->page())
         m_identifier = createUniqueIdentifier();
 }
 
 MainThreadWebSocketChannel::~MainThreadWebSocketChannel()
 {
@@ skipping 63 matching lines @@
 }
 
 WebSocketChannel::SendResult MainThreadWebSocketChannel::send(PassOwnPtr<Vector<char> > data)
 {
     WTF_LOG(Network, "MainThreadWebSocketChannel %p send() Sending Vector %p", this, data.get());
     enqueueVector(WebSocketFrame::OpCodeBinary, data);
     processOutgoingFrameQueue();
     return WebSocketChannel::SendSuccess;
 }
 
-unsigned long MainThreadWebSocketChannel::bufferedAmount() const
-{
-    WTF_LOG(Network, "MainThreadWebSocketChannel %p bufferedAmount()", this);
-    ASSERT(m_handle);
-    ASSERT(!m_suspended);
-    return m_handle->bufferedAmount();
-}
-
 void MainThreadWebSocketChannel::close(int code, const String& reason)
 {
     WTF_LOG(Network, "MainThreadWebSocketChannel %p close() code=%d reason='%s'", this, code, reason.utf8().data());
     ASSERT(!m_suspended);
     if (!m_handle)
         return;
     startClosingHandshake(code, reason);
     if (!m_closingTimer.isActive())
         m_closingTimer.startOneShot(2 * TCPMaximumSegmentLifetime, FROM_HERE);
 }
@@ skipping 108 matching lines @@
         const String message = "WebSocket connection to '" + m_handshake->url().elidedString() + "' failed: Connection closed before receiving a handshake response";
         m_document->addConsoleMessage(JSMessageSource, ErrorMessageLevel, message, m_sourceURLAtConstruction, m_lineNumberAtConstruction);
     }
 
     m_state = ChannelClosed;
     if (m_closingTimer.isActive())
         m_closingTimer.stop();
     if (m_outgoingFrameQueueStatus != OutgoingFrameQueueClosed)
         abortOutgoingFrameQueue();
     if (m_handle) {
-        m_unhandledBufferedAmount = m_handle->bufferedAmount();
         WebSocketChannelClient* client = m_client;
         m_client = 0;
         clearDocument();
         m_handle = nullptr;
         if (client)
-            client->didClose(m_unhandledBufferedAmount, m_receivedClosingHandshake ? WebSocketChannelClient::ClosingHandshakeComplete : WebSocketChannelClient::ClosingHandshakeIncomplete, m_closeEventCode, m_closeEventReason);
+            client->didClose(m_receivedClosingHandshake ? WebSocketChannelClient::ClosingHandshakeComplete : WebSocketChannelClient::ClosingHandshakeIncomplete, m_closeEventCode, m_closeEventReason);
     }
     deref();
 }
 
 void MainThreadWebSocketChannel::didReceiveSocketStreamData(SocketStreamHandle* handle, const char* data, int len)
 {
     WTF_LOG(Network, "MainThreadWebSocketChannel %p didReceiveSocketStreamData() Received %d bytes", this, len);
     RefPtrWillBeRawPtr<MainThreadWebSocketChannel> protect(this); // The client can close the channel, potentially removing the last reference.
     ASSERT(handle == m_handle);
     if (!m_document)
@@ skipping 10 matching lines @@
     if (m_shouldDiscardReceivedData)
         return;
     if (!appendToBuffer(data, len)) {
         m_shouldDiscardReceivedData = true;
         failAsError("Ran out of memory while receiving WebSocket data.");
         return;
     }
     processBuffer();
 }
 
-void MainThreadWebSocketChannel::didUpdateBufferedAmount(SocketStreamHandle*, size_t bufferedAmount)
+void MainThreadWebSocketChannel::didConsumeBufferedAmount(SocketStreamHandle*, size_t consumed)
 {
-    if (m_client)
-        m_client->didUpdateBufferedAmount(bufferedAmount);
+    if (m_framingOverheadQueue.isEmpty()) {
+        // Ignore the handshake consumption.
+        return;
+    }
+    if (!m_client || m_state == ChannelClosed)
+        return;
+    size_t remain = consumed;
+    while (remain > 0) {
+        ASSERT(!m_framingOverheadQueue.isEmpty());
+        const FramingOverhead& frame = m_framingOverheadQueue.first();
+
+        ASSERT(m_numConsumedBytesInCurrentFrame <= frame.frameDataSize());
+        size_t consumedInThisFrame = std::min(remain, frame.frameDataSize() - m_numConsumedBytesInCurrentFrame);
+        remain -= consumedInThisFrame;
+        m_numConsumedBytesInCurrentFrame += consumedInThisFrame;
+
+        if (m_numConsumedBytesInCurrentFrame == frame.frameDataSize()) {
+            if (m_client && WebSocketFrame::isNonControlOpCode(frame.opcode())) {
+                // FIXME: As |consumed| is the number of possibly compressed
+                // bytes, we can't determine the number of consumed original
+                // bytes in the middle of a frame.
+                m_client->didConsumeBufferedAmount(frame.originalPayloadLength());
+            }
+            m_framingOverheadQueue.takeFirst();
+            m_numConsumedBytesInCurrentFrame = 0;
+        }
+    }
 }
 
 void MainThreadWebSocketChannel::didFailSocketStream(SocketStreamHandle* handle, const SocketStreamError& error)
 {
     WTF_LOG(Network, "MainThreadWebSocketChannel %p didFailSocketStream()", this);
     ASSERT_UNUSED(handle, handle == m_handle || !m_handle);
     m_shouldDiscardReceivedData = true;
     String message;
     if (error.isNull())
         message = "WebSocket network error";
@@ skipping 488 matching lines @@
         return false;
     }
 
     if (!m_perMessageDeflate.deflate(frame)) {
         failAsError(m_perMessageDeflate.failureReason());
         return false;
     }
 
     Vector<char> frameData;
     frame.makeFrameData(frameData);
+    m_framingOverheadQueue.append(FramingOverhead(opCode, frameData.size(), dataLength));
 
     m_perMessageDeflate.resetDeflateBuffer();
     return m_handle->send(frameData.data(), frameData.size());
 }
 
 } // namespace WebCore