Implement suspend / resume in NewWebSocketChannelImpl.

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 26 matching lines @@
 #include "core/platform/NotImplemented.h"
 #include "modules/websockets/WebSocketChannel.h"
 #include "modules/websockets/WebSocketChannelClient.h"
 #include "public/platform/Platform.h"
 #include "public/platform/WebSocketHandle.h"
 #include "public/platform/WebString.h"
 #include "public/platform/WebURL.h"
 #include "public/platform/WebVector.h"
 #include "weborigin/SecurityOrigin.h"
 #include "wtf/ArrayBuffer.h"
+#include "wtf/Vector.h"
 #include "wtf/text/WTFString.h"
 
 // FIXME: We should implement Inspector notification.
 // FIXME: We should implement send(Blob).
-// FIXME: We should implement suspend / resume.
 // FIXME: We should add log messages.
 
 using WebKit::WebSocketHandle;
 
 namespace WebCore {
 
 namespace {
 
 bool isClean(int code)
 {
     return code == WebSocketChannel::CloseEventCodeNormalClosure
         || (WebSocketChannel::CloseEventCodeMinimumUserDefined <= code
         && code <= WebSocketChannel::CloseEventCodeMaximumUserDefined);
 }
 
 } // namespace
 
+class NewWebSocketChannelImpl::Resumer {
+public:
+    enum State {
+        Active,

[tyoshino (SeeGerritForStatus), 2013/09/10 09:39:29]

could you consider moving state to the channel?

it's good that you created this inner class, but it looks it might be cleaner to
let the channel have the state.

[yhirano, 2013/09/10 12:25:17]

I divided the state to two booleans: NewWebSocketChannelImpl::m_isSuspended and
Resumer::m_isAborted.

+        Suspended,
+        Resuming,
+        Aborted,
+    };
+
+    struct PendingEvent {
+        enum Type {
+            DidConnectComplete,
+            DidReceiveTextMessage,
+            DidReceiveBinaryMessage,
+            DidReceiveError,
+            DidClose,
+        };
+        Type type;
+        Vector<char> message; // for DidReceiveTextMessage / DidReceiveBinaryMessage
+        int closingCode; // for DidClose
+        String closingReason; // for DidClose
+
+        PendingEvent(Type type) : type(type), closingCode(0) { }
+        PendingEvent(Type, Vector<char>*);
+        PendingEvent(int code, const String& reason) : type(DidClose), closingCode(code), closingReason(reason) { }
+    };
+
+    Resumer(NewWebSocketChannelImpl*);
+    ~Resumer();
+
+    State state() const { return m_state; }
+    void append(const PendingEvent&);
+    void suspend();
+    void resumeLater();
+    void abort();
+
+private:
+    class PendingEventProcessor : public RefCounted<PendingEventProcessor> {
+    public:
+        PendingEventProcessor() : m_isDetached(false) { }
+        virtual ~PendingEventProcessor() { }
+        void detach() { m_isDetached = true; }
+        void append(const PendingEvent& e) { m_events.append(e); }
+        void process(NewWebSocketChannelImpl*);
+
+    private:
+        bool m_isDetached;
+        Vector<PendingEvent> m_events;
+    };
+
+    void resumeNow(Timer<Resumer>*);
+
+    State m_state;
+    NewWebSocketChannelImpl* m_channel;
+    RefPtr<PendingEventProcessor> m_pendingEventProcessor;
+    Timer<Resumer> m_timer;
+};
+
+NewWebSocketChannelImpl::Resumer::PendingEvent::PendingEvent(Type type, Vector<char>* data) : type(type), closingCode(0)
+{
+    ASSERT(type == DidReceiveTextMessage || type == DidReceiveBinaryMessage);
+    message.swap(*data);
+}
+
+NewWebSocketChannelImpl::Resumer::Resumer(NewWebSocketChannelImpl* channel)
+    : m_state(Active)
+    , m_channel(channel)
+    , m_pendingEventProcessor(adoptRef(new PendingEventProcessor))
+    , m_timer(this, &Resumer::resumeNow) { }
+
+NewWebSocketChannelImpl::Resumer::~Resumer()
+{
+    abort();
+}
+
+void NewWebSocketChannelImpl::Resumer::append(const PendingEvent& event)
+{
+    if (m_state == Aborted)
+        return;
+    m_pendingEventProcessor->append(event);
+}
+
+void NewWebSocketChannelImpl::Resumer::suspend()
+{
+    if (m_state == Aborted)
+        return;
+    m_timer.stop();
+    m_state = Suspended;
+}
+
+void NewWebSocketChannelImpl::Resumer::resumeLater()
+{
+    if (m_state == Aborted)
+        return;
+    m_state = Resuming;
+    if (!m_timer.isActive())
+        m_timer.startOneShot(0);
+}
+
+void NewWebSocketChannelImpl::Resumer::abort()
+{
+    if (m_state == Aborted)
+        return;
+    m_state = Aborted;
+    m_timer.stop();
+    m_pendingEventProcessor->detach();
+    m_pendingEventProcessor = 0;
+}
+
+void NewWebSocketChannelImpl::Resumer::resumeNow(Timer<Resumer>*)
+{
+    ASSERT(m_state == Resuming);
+    m_state = Active;
+
+    ASSERT(m_channel->m_client);
+    if (m_channel->m_handle) {
+        m_channel->sendInternal();
+        m_channel->flowControlIfNecessary();
+    }
+    m_pendingEventProcessor->process(m_channel);
+    // |this| can be aborted here.
+    // |this| can be deleted here.
+}
+
+void NewWebSocketChannelImpl::Resumer::PendingEventProcessor::process(NewWebSocketChannelImpl* channel)
+{
+    RefPtr<PendingEventProcessor> protect(this);
+    for (size_t i = 0; i < m_events.size() && !m_isDetached; ++i) {
+        PendingEvent& event = m_events[i];
+        switch (event.type) {
+        case PendingEvent::DidConnectComplete:
+            channel->handleDidConnect();
+            // |this| can be detached here.
+            break;
+        case PendingEvent::DidReceiveTextMessage:
+            channel->handleTextMessage(&event.message);
+            // |this| can be detached here.
+            break;
+        case PendingEvent::DidReceiveBinaryMessage:
+            channel->handleBinaryMessage(&event.message);
+            // |this| can be detached here.
+            break;
+        case PendingEvent::DidReceiveError:
+            channel->handleDidReceiveMessageError();
+            // |this| can be detached here.
+            break;
+        case PendingEvent::DidClose:
+            channel->handleDidClose(event.closingCode, event.closingReason);
+            // |this| can be detached here.
+            break;
+        }
+    }
+    m_events.clear();
+}
+
 NewWebSocketChannelImpl::NewWebSocketChannelImpl(ScriptExecutionContext* context, WebSocketChannelClient* client, const String& sourceURL, unsigned lineNumber)
     : ContextLifecycleObserver(context)
     , m_handle(adoptPtr(WebKit::Platform::current()->createWebSocketHandle()))
     , m_client(client)
+    , m_resumer(adoptPtr(new Resumer(this)))
     , m_sendingQuota(0)
     , m_receivedDataSizeForFlowControl(receivedDataSizeForFlowControlHighWaterMark * 2) // initial quota
     , m_bufferedAmount(0)
     , m_sentSizeOfTopMessage(0)
 {
 }
 
+NewWebSocketChannelImpl::~NewWebSocketChannelImpl()
+{
+}
+
 void NewWebSocketChannelImpl::connect(const KURL& url, const String& protocol)
 {
     if (!m_handle)
         return;
     m_url = url;
     Vector<String> protocols;
     // Since protocol is already verified and escaped, we can simply split it.
     protocol.split(", ", true, protocols);
     WebKit::WebVector<WebKit::WebString> webProtocols(protocols.size());
     for (size_t i = 0; i < protocols.size(); ++i) {
@@ skipping 45 matching lines @@
 {
     ASSERT(m_handle);
     m_handle->close(static_cast<unsigned short>(code), reason);
 }
 
 void NewWebSocketChannelImpl::fail(const String& reason, MessageLevel level, const String& sourceURL, unsigned lineNumber)
 {
     // m_handle and m_client can be null here.
     if (m_client)
         m_client->didReceiveMessageError();
-    handleDidClose(CloseEventCodeAbnormalClosure, reason);
-    // handleDidClose may delete this object.
+    if (m_resumer->state() == Resumer::Active) {
+        handleDidClose(CloseEventCodeAbnormalClosure, reason);
+        // handleDidClose may delete this object.
+    } else {
+        m_resumer->append(Resumer::PendingEvent(CloseEventCodeAbnormalClosure, reason));
+    }
 }
 
 void NewWebSocketChannelImpl::disconnect()
 {
     if (m_handle)
         m_handle->close(CloseEventCodeAbnormalClosure, "");
     m_handle.clear();
     m_client = 0;
+    m_resumer->abort();
 }
 
 void NewWebSocketChannelImpl::suspend()
 {
-    notImplemented();
+    m_resumer->suspend();
 }
 
 void NewWebSocketChannelImpl::resume()
 {
-    notImplemented();
+    m_resumer->resumeLater();
 }
 
 NewWebSocketChannelImpl::Message::Message(const String& text)
     : type(MessageTypeText)
     , text(text.utf8(String::StrictConversionReplacingUnpairedSurrogatesWithFFFD)) { }
 
 NewWebSocketChannelImpl::Message::Message(const Blob& blob)
     : type(MessageTypeBlob)
     , blob(Blob::create(blob.url(), blob.type(), blob.size())) { }
 
@@ skipping 50 matching lines @@
 {
     if (!m_handle || m_receivedDataSizeForFlowControl < receivedDataSizeForFlowControlHighWaterMark) {
         return;
     }
     m_handle->flowControl(m_receivedDataSizeForFlowControl);
     m_receivedDataSizeForFlowControl = 0;
 }
 
 void NewWebSocketChannelImpl::didConnect(WebSocketHandle* handle, bool succeed, const WebKit::WebString& selectedProtocol, const WebKit::WebString& extensions)
 {
-    if (!m_handle) {
-        return;
-    }
+    ASSERT(m_handle);
     ASSERT(handle == m_handle);
     ASSERT(m_client);
     if (!succeed) {
         failAsError("Cannot connect to " + m_url.string() + ".");
         // failAsError may delete this object.
         return;
     }
     m_subprotocol = selectedProtocol;
     m_extensions = extensions;
-    m_client->didConnect();
+    if (m_resumer->state() == Resumer::Active)
+        m_client->didConnect();
+    else
+        m_resumer->append(Resumer::PendingEvent(Resumer::PendingEvent::DidConnectComplete));
 }
 
 void NewWebSocketChannelImpl::didReceiveData(WebSocketHandle* handle, WebSocketHandle::MessageType type, const char* data, size_t size, bool fin)
 {
-    if (!m_handle) {
-        return;
-    }
+    ASSERT(m_handle);
     ASSERT(handle == m_handle);
     ASSERT(m_client);
     // Non-final frames cannot be empty.
     ASSERT(fin || size);
     switch (type) {
     case WebSocketHandle::MessageTypeText:
         ASSERT(m_receivingMessageData.isEmpty());
         m_receivingMessageTypeIsText = true;
         break;
     case WebSocketHandle::MessageTypeBinary:
         ASSERT(m_receivingMessageData.isEmpty());
         m_receivingMessageTypeIsText = false;
         break;
     case WebSocketHandle::MessageTypeContinuation:
         ASSERT(!m_receivingMessageData.isEmpty());
         break;
     }
     m_receivingMessageData.append(data, size);
     m_receivedDataSizeForFlowControl += size;
     flowControlIfNecessary();
     if (!fin) {
         return;
     }
+    if (m_resumer->state() != Resumer::Active) {
+        Resumer::PendingEvent::Type type =
+            m_receivingMessageTypeIsText ? Resumer::PendingEvent::DidReceiveTextMessage : Resumer::PendingEvent::DidReceiveBinaryMessage;
+        m_resumer->append(Resumer::PendingEvent(type, &m_receivingMessageData));
+        return;
+    }
     Vector<char> messageData;
     messageData.swap(m_receivingMessageData);
     if (m_receivingMessageTypeIsText) {
         handleTextMessage(&messageData);
         // handleTextMessage may delete this object.
     } else {
         handleBinaryMessage(&messageData);
     }
 }
 
 
 void NewWebSocketChannelImpl::didClose(WebSocketHandle* handle, unsigned short code, const WebKit::WebString& reason)
 {
+    ASSERT(m_handle);
+    m_handle.clear();
     // FIXME: Maybe we should notify an error to m_client for some didClose messages.
-    handleDidClose(code, reason);
-    // handleDidClose may delete this object.
+    if (m_resumer->state() == Resumer::Active) {
+        handleDidClose(code, reason);
+        // handleDidClose may delete this object.
+    } else {
+        m_resumer->append(Resumer::PendingEvent(code, reason));
+    }
+}
+
+void NewWebSocketChannelImpl::handleDidConnect()
+{
+    ASSERT(m_client);
+    ASSERT(m_resumer->state() == Resumer::Active);
+    m_client->didConnect();
 }
 
 void NewWebSocketChannelImpl::handleTextMessage(Vector<char>* messageData)
 {
     ASSERT(m_client);
+    ASSERT(m_resumer->state() == Resumer::Active);
     ASSERT(messageData);
     String message = "";
     if (m_receivingMessageData.size() > 0) {
         message = String::fromUTF8(messageData->data(), messageData->size());
     }
     // For consistency with handleBinaryMessage, we clear |messageData|.
     messageData->clear();
     if (message.isNull()) {
         failAsError("Could not decode a text frame as UTF-8.");
         // failAsError may delete this object.
     } else {
         m_client->didReceiveMessage(message);
     }
 }
 
 void NewWebSocketChannelImpl::handleBinaryMessage(Vector<char>* messageData)
 {
     ASSERT(m_client);
+    ASSERT(m_resumer->state() == Resumer::Active);
     ASSERT(messageData);
     OwnPtr<Vector<char> > binaryData = adoptPtr(new Vector<char>);
     messageData->swap(*binaryData);
     m_client->didReceiveBinaryData(binaryData.release());
 }
 
+void NewWebSocketChannelImpl::handleDidReceiveMessageError()
+{
+    ASSERT(m_client);
+    ASSERT(m_resumer->state() == Resumer::Active);
+    m_client->didReceiveMessageError();
+}
+
 void NewWebSocketChannelImpl::handleDidClose(unsigned short code, const String& reason)
 {
+    ASSERT(m_resumer->state() == Resumer::Active);
     m_handle.clear();
+    m_resumer->abort();
     if (!m_client) {
         return;
     }
     WebSocketChannelClient* client = m_client;
     m_client = 0;
     WebSocketChannelClient::ClosingHandshakeCompletionStatus status =
         isClean(code) ? WebSocketChannelClient::ClosingHandshakeComplete : WebSocketChannelClient::ClosingHandshakeIncomplete;
     client->didClose(m_bufferedAmount, status, code, reason);
     // client->didClose may delete this object.
 }
 
 } // namespace WebCore