[WebSocket] bufferedAmount should not decrease inside a task.

Source/modules/websockets/WebSocket.cpp

 /*
  * Copyright (C) 2011 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 210 matching lines @@
 
 const char* WebSocket::subprotocolSeperator()
 {
     return ", ";
 }
 
 WebSocket::WebSocket(ExecutionContext* context)
     : ActiveDOMObject(context)
     , m_state(CONNECTING)
     , m_bufferedAmount(0)
+    , m_consumedBufferedAmount(0)
     , m_bufferedAmountAfterClose(0)
     , m_binaryType(BinaryTypeBlob)
     , m_subprotocol("")
     , m_extensions("")
     , m_eventQueue(EventQueue::create(this))
+    , m_bufferedAmountConsumeTimer(this, &WebSocket::reflectBufferedAmountConsumption)
 {
     ScriptWrappable::init(this);
 }
 
 WebSocket::~WebSocket()
 {
     ASSERT(!m_channel);
 }
 
 void WebSocket::logError(const String& message)
@@ skipping 120 matching lines @@
 }
 
 void WebSocket::updateBufferedAmountAfterClose(unsigned long payloadSize)
 {
     m_bufferedAmountAfterClose = saturateAdd(m_bufferedAmountAfterClose, payloadSize);
     m_bufferedAmountAfterClose = saturateAdd(m_bufferedAmountAfterClose, getFramingOverhead(payloadSize));
 
     logError("WebSocket is already in CLOSING or CLOSED state.");
 }
 
+void WebSocket::reflectBufferedAmountConsumption(Timer<WebSocket>*)
+{
+    ASSERT(m_bufferedAmount >= m_consumedBufferedAmount);
+    WTF_LOG(Network, "WebSocket %p reflectBufferedAmountConsumption() %lu => %lu", this, m_bufferedAmount, m_bufferedAmount - m_consumedBufferedAmount);
+
+    m_bufferedAmount -= m_consumedBufferedAmount;
+    m_consumedBufferedAmount = 0;
+}
+
 void WebSocket::releaseChannel()
 {
     ASSERT(m_channel);
     m_channel->disconnect();
     m_channel = nullptr;
 }
 
 void WebSocket::send(const String& message, ExceptionState& exceptionState)
 {
     WTF_LOG(Network, "WebSocket %p send() Sending String '%s'", this, message.utf8().data());
     if (m_state == CONNECTING) {
         setInvalidStateErrorForSendMethod(exceptionState);
         return;
     }
     // No exception is raised if the connection was once established but has subsequently been closed.
     if (m_state == CLOSING || m_state == CLOSED) {
         updateBufferedAmountAfterClose(message.utf8().length());
         return;
     }
     ASSERT(m_channel);
+    m_bufferedAmount += message.utf8().length();
     handleSendResult(m_channel->send(message), exceptionState, WebSocketSendTypeString);
 }
 
 void WebSocket::send(ArrayBuffer* binaryData, ExceptionState& exceptionState)
 {
     WTF_LOG(Network, "WebSocket %p send() Sending ArrayBuffer %p", this, binaryData);
     ASSERT(binaryData);
     if (m_state == CONNECTING) {
         setInvalidStateErrorForSendMethod(exceptionState);
         return;
     }
     if (m_state == CLOSING || m_state == CLOSED) {
         updateBufferedAmountAfterClose(binaryData->byteLength());
         return;
     }
     ASSERT(m_channel);
+    m_bufferedAmount += binaryData->byteLength();
     handleSendResult(m_channel->send(*binaryData, 0, binaryData->byteLength()), exceptionState, WebSocketSendTypeArrayBuffer);
 }
 
 void WebSocket::send(ArrayBufferView* arrayBufferView, ExceptionState& exceptionState)
 {
     WTF_LOG(Network, "WebSocket %p send() Sending ArrayBufferView %p", this, arrayBufferView);
     ASSERT(arrayBufferView);
     if (m_state == CONNECTING) {
         setInvalidStateErrorForSendMethod(exceptionState);
         return;
     }
     if (m_state == CLOSING || m_state == CLOSED) {
         updateBufferedAmountAfterClose(arrayBufferView->byteLength());
         return;
     }
     ASSERT(m_channel);
+    m_bufferedAmount += arrayBufferView->byteLength();
     RefPtr<ArrayBuffer> arrayBuffer(arrayBufferView->buffer());
     handleSendResult(m_channel->send(*arrayBuffer, arrayBufferView->byteOffset(), arrayBufferView->byteLength()), exceptionState, WebSocketSendTypeArrayBufferView);
 }
 
 void WebSocket::send(Blob* binaryData, ExceptionState& exceptionState)
 {
     WTF_LOG(Network, "WebSocket %p send() Sending Blob '%s'", this, binaryData->uuid().utf8().data());
     ASSERT(binaryData);
     if (m_state == CONNECTING) {
         setInvalidStateErrorForSendMethod(exceptionState);
         return;
     }
     if (m_state == CLOSING || m_state == CLOSED) {
         updateBufferedAmountAfterClose(static_cast<unsigned long>(binaryData->size()));
         return;
     }
+    m_bufferedAmount += binaryData->size();
     ASSERT(m_channel);
     handleSendResult(m_channel->send(binaryData->blobDataHandle()), exceptionState, WebSocketSendTypeBlob);
 }
 
 void WebSocket::close(unsigned short code, const String& reason, ExceptionState& exceptionState)
 {
     closeInternal(code, reason, exceptionState);
 }
 
 void WebSocket::close(ExceptionState& exceptionState)
@@ skipping 179 matching lines @@
     }
 }
 
 void WebSocket::didReceiveMessageError()
 {
     WTF_LOG(Network, "WebSocket %p didReceiveMessageError()", this);
     m_state = CLOSED;
     m_eventQueue->dispatch(Event::create(EventTypeNames::error));
 }
 
-void WebSocket::didUpdateBufferedAmount(unsigned long bufferedAmount)
+void WebSocket::didConsumeBufferedAmount(unsigned long consumed)
 {
-    WTF_LOG(Network, "WebSocket %p didUpdateBufferedAmount() New bufferedAmount is %lu", this, bufferedAmount);
+    ASSERT(m_bufferedAmount >= consumed);
+    WTF_LOG(Network, "WebSocket %p didConsumeBufferedAmount(%lu)", this, consumed);
     if (m_state == CLOSED)
         return;
-    m_bufferedAmount = bufferedAmount;
+    m_consumedBufferedAmount += consumed;
+    if (!m_bufferedAmountConsumeTimer.isActive())
+        m_bufferedAmountConsumeTimer.startOneShot(0, FROM_HERE);
 }
 
 void WebSocket::didStartClosingHandshake()
 {
     WTF_LOG(Network, "WebSocket %p didStartClosingHandshake()", this);
     m_state = CLOSING;
 }
 
-void WebSocket::didClose(unsigned long unhandledBufferedAmount, ClosingHandshakeCompletionStatus closingHandshakeCompletion, unsigned short code, const String& reason)
+void WebSocket::didClose(ClosingHandshakeCompletionStatus closingHandshakeCompletion, unsigned short code, const String& reason)
 {
     WTF_LOG(Network, "WebSocket %p didClose()", this);
     if (!m_channel)
         return;
-    bool wasClean = m_state == CLOSING && !unhandledBufferedAmount && closingHandshakeCompletion == ClosingHandshakeComplete && code != WebSocketChannel::CloseEventCodeAbnormalClosure;
+    bool hasAllDataConsumed = m_bufferedAmount == m_consumedBufferedAmount;
+    bool wasClean = m_state == CLOSING && hasAllDataConsumed && closingHandshakeCompletion == ClosingHandshakeComplete && code != WebSocketChannel::CloseEventCodeAbnormalClosure;
+    m_state = CLOSED;
 
-    m_state = CLOSED;
-    m_bufferedAmount = unhandledBufferedAmount;
     m_eventQueue->dispatch(CloseEvent::create(wasClean, code, reason));
     releaseChannel();
 }
 
 size_t WebSocket::getFramingOverhead(size_t payloadSize)
 {
     static const size_t hybiBaseFramingOverhead = 2; // Every frame has at least two-byte header.
     static const size_t hybiMaskingKeyLength = 4; // Every frame from client must have masking key.
     static const size_t minimumPayloadSizeWithTwoByteExtendedPayloadLength = 126;
     static const size_t minimumPayloadSizeWithEightByteExtendedPayloadLength = 0x10000;
     size_t overhead = hybiBaseFramingOverhead + hybiMaskingKeyLength;
     if (payloadSize >= minimumPayloadSizeWithEightByteExtendedPayloadLength)
         overhead += 8;
     else if (payloadSize >= minimumPayloadSizeWithTwoByteExtendedPayloadLength)
         overhead += 2;
     return overhead;
 }
 
 void WebSocket::trace(Visitor* visitor)
 {
     visitor->trace(m_channel);
     visitor->trace(m_eventQueue);
     EventTargetWithInlineData::trace(visitor);
 }
 
 } // namespace WebCore