[WebSocket] bufferedAmount should not decrease inside a task.

Source/modules/websockets/NewWebSocketChannelImpl.cpp

 /*
  * Copyright (C) 2013 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 97 matching lines @@
     // |this| is deleted here.
 }
 
 NewWebSocketChannelImpl::NewWebSocketChannelImpl(ExecutionContext* context, WebSocketChannelClient* client, const String& sourceURL, unsigned lineNumber)
     : ContextLifecycleObserver(context)
     , m_handle(adoptPtr(blink::Platform::current()->createWebSocketHandle()))
     , m_client(client)
     , m_identifier(0)
     , m_sendingQuota(0)
     , m_receivedDataSizeForFlowControl(receivedDataSizeForFlowControlHighWaterMark * 2) // initial quota
-    , m_bufferedAmount(0)
     , m_sentSizeOfTopMessage(0)
     , m_sourceURLAtConstruction(sourceURL)
     , m_lineNumberAtConstruction(lineNumber)
 {
     if (context->isDocument() && toDocument(context)->page())
         m_identifier = createUniqueIdentifier();
 }
 
 NewWebSocketChannelImpl::~NewWebSocketChannelImpl()
 {
@@ skipping 93 matching lines @@
     if (m_identifier) {
         // FIXME: Change the inspector API to show the entire message instead
         // of individual frames.
         InspectorInstrumentation::didSendWebSocketFrame(document(), m_identifier, WebSocketFrame::OpCodeBinary, true, data->data(), data->size());
     }
     m_messages.append(adoptPtr(new Message(data)));
     sendInternal();
     return SendSuccess;
 }
 
-unsigned long NewWebSocketChannelImpl::bufferedAmount() const
-{
-    WTF_LOG(Network, "NewWebSocketChannelImpl %p bufferedAmount()", this);
-    return m_bufferedAmount;
-}
-
 void NewWebSocketChannelImpl::close(int code, const String& reason)
 {
     WTF_LOG(Network, "NewWebSocketChannelImpl %p close(%d, %s)", this, code, reason.utf8().data());
     ASSERT(m_handle);
     unsigned short codeToSend = static_cast<unsigned short>(code == CloseEventCodeNotSpecified ? CloseEventCodeNoStatusRcvd : code);
     m_handle->close(codeToSend, reason);
 }
 
 void NewWebSocketChannelImpl::fail(const String& reason, MessageLevel level, const String& sourceURL, unsigned lineNumber)
 {
@@ skipping 50 matching lines @@
     : type(MessageTypeArrayBuffer)
     , arrayBuffer(arrayBuffer) { }
 
 NewWebSocketChannelImpl::Message::Message(PassOwnPtr<Vector<char> > vectorData)
     : type(MessageTypeVector)
     , vectorData(vectorData) { }
 
 void NewWebSocketChannelImpl::sendInternal()
 {
     ASSERT(m_handle);
-    unsigned long bufferedAmount = m_bufferedAmount;
+    unsigned long consumedBufferedAmount = 0;
     while (!m_messages.isEmpty() && m_sendingQuota > 0 && !m_blobLoader) {
         bool final = false;
         Message* message = m_messages.first().get();
         switch (message->type) {
         case MessageTypeText: {
             WebSocketHandle::MessageType type =
                 m_sentSizeOfTopMessage ? WebSocketHandle::MessageTypeContinuation : WebSocketHandle::MessageTypeText;
             size_t size = std::min(static_cast<size_t>(m_sendingQuota), message->text.length() - m_sentSizeOfTopMessage);
             final = (m_sentSizeOfTopMessage + size == message->text.length());
             m_handle->send(final, type, message->text.data() + m_sentSizeOfTopMessage, size);
             m_sentSizeOfTopMessage += size;
             m_sendingQuota -= size;
+            consumedBufferedAmount += size;
             break;
         }
         case MessageTypeBlob:
             ASSERT(!m_blobLoader);
             m_blobLoader = adoptPtrWillBeNoop(new BlobLoader(message->blobDataHandle, this));
             break;
         case MessageTypeArrayBuffer: {
             WebSocketHandle::MessageType type =
                 m_sentSizeOfTopMessage ? WebSocketHandle::MessageTypeContinuation : WebSocketHandle::MessageTypeBinary;
             size_t size = std::min(static_cast<size_t>(m_sendingQuota), message->arrayBuffer->byteLength() - m_sentSizeOfTopMessage);
             final = (m_sentSizeOfTopMessage + size == message->arrayBuffer->byteLength());
             m_handle->send(final, type, static_cast<const char*>(message->arrayBuffer->data()) + m_sentSizeOfTopMessage, size);
             m_sentSizeOfTopMessage += size;
             m_sendingQuota -= size;
+            consumedBufferedAmount += size;
             break;
         }
         case MessageTypeVector: {
             WebSocketHandle::MessageType type =
                 m_sentSizeOfTopMessage ? WebSocketHandle::MessageTypeContinuation : WebSocketHandle::MessageTypeBinary;
             size_t size = std::min(static_cast<size_t>(m_sendingQuota), message->vectorData->size() - m_sentSizeOfTopMessage);
             final = (m_sentSizeOfTopMessage + size == message->vectorData->size());
             m_handle->send(final, type, message->vectorData->data() + m_sentSizeOfTopMessage, size);
             m_sentSizeOfTopMessage += size;
             m_sendingQuota -= size;
+            consumedBufferedAmount += size;
             break;
         }
         }
         if (final) {
             m_messages.removeFirst();
             m_sentSizeOfTopMessage = 0;
         }
     }
-    if (m_client && m_bufferedAmount != bufferedAmount) {
-        m_client->didUpdateBufferedAmount(m_bufferedAmount);
-    }
+    if (m_client && consumedBufferedAmount > 0)
+        m_client->didConsumeBufferedAmount(consumedBufferedAmount);
 }
 
 void NewWebSocketChannelImpl::flowControlIfNecessary()
 {
     if (!m_handle || m_receivedDataSizeForFlowControl < receivedDataSizeForFlowControlHighWaterMark) {
         return;
     }
     m_handle->flowControl(m_receivedDataSizeForFlowControl);
     m_receivedDataSizeForFlowControl = 0;
 }
@@ skipping 10 matching lines @@
 {
     m_handle.clear();
     abortAsyncOperations();
     if (!m_client) {
         return;
     }
     WebSocketChannelClient* client = m_client;
     m_client = 0;
     WebSocketChannelClient::ClosingHandshakeCompletionStatus status =
         wasClean ? WebSocketChannelClient::ClosingHandshakeComplete : WebSocketChannelClient::ClosingHandshakeIncomplete;
-    client->didClose(m_bufferedAmount, status, code, reason);
+    client->didClose(status, code, reason);
     // client->didClose may delete this object.
 }
 
 Document* NewWebSocketChannelImpl::document()
 {
     ASSERT(m_identifier);
     ExecutionContext* context = executionContext();
     ASSERT(context->isDocument());
     return toDocument(context);
 }
@@ skipping 151 matching lines @@
     // |this| can be deleted here.
 }
 
 void NewWebSocketChannelImpl::trace(Visitor* visitor)
 {
     visitor->trace(m_blobLoader);
     WebSocketChannel::trace(visitor);
 }
 
 } // namespace WebCore