WebSocket implementation in WebKit

WebCore/platform/network/soup/SocketStreamHandleSoup.cpp

+/*
+ * Copyright (C) 2009 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:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY APPLE COMPUTER, INC. ``AS IS'' AND ANY
+ * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL APPLE COMPUTER, INC. OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
+ * OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include "config.h"
+#include "SocketStreamHandle.h"
+
+#include <errno.h>
+#include <glib.h>
+#include <netdb.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+
+#include <wtf/PassOwnPtr.h>
+
+#include "CString.h"
+#include "Logging.h"
+#include "KURL.h"
+#include "NotImplemented.h"
+#include "SharedBuffer.h"
+#include "SocketStreamError.h"
+#include "SocketStreamHandleClient.h"
+
+#undef LOG
+#define LOG(channel, ...) do { fprintf(stderr, __VA_ARGS__); fprintf(stderr, "\n"); } while (0);
+
+namespace WebCore {
+
+const int kSocketStreamHandleBufSize = 1024;
+
+class SocketStreamHandleInternal {
+public:
+    static PassOwnPtr<SocketStreamHandleInternal> create(SocketStreamHandle* handle) { return new SocketStreamHandleInternal(handle); }
+    ~SocketStreamHandleInternal();
+
+    void connect(const KURL&);
+    int send(const char*, int);
+    void close();
+
+    static gboolean handleRead(GIOChannel* source, GIOCondition condition, gpointer data);
+    static gboolean handleWrite(GIOChannel* source, GIOCondition condition, gpointer data);
+    static gboolean handleError(GIOChannel* source, GIOCondition condition, gpointer data);
+    static gboolean handleClose(gpointer data);
+
+private:
+    explicit SocketStreamHandleInternal(SocketStreamHandle*);
+
+    gboolean canReceive();
+    gboolean canSend();
+    gboolean didError(GIOCondition condition);
+    gboolean didClose();
+
+    SocketStreamHandle* m_handle;
+    struct addrinfo* m_addrinfo;
+    int m_sock;
+    GIOChannel* m_g_io;
+    guint m_in_watch;
+    guint m_out_watch;
+    guint m_err_watch;
+    Vector<char> m_send_buffer;
+    gchar m_read_buf[kSocketStreamHandleBufSize];
+};
+
+/* static */
+gboolean SocketStreamHandleInternal::handleRead(GIOChannel* source, GIOCondition cdondition, gpointer data)
+{
+    LOG(Network, "SocketStreamHandle handleRead");
+    SocketStreamHandleInternal* internal = reinterpret_cast<SocketStreamHandleInternal*>(data);
+    return internal->canReceive();
+}
+
+/* static */
+gboolean SocketStreamHandleInternal::handleWrite(GIOChannel* source, GIOCondition condition, gpointer data)
+{
+    LOG(Network, "SocketStreamHandle handleWrite");
+    SocketStreamHandleInternal* internal = reinterpret_cast<SocketStreamHandleInternal*>(data);
+    return internal->canSend();
+}
+
+/* static */
+gboolean SocketStreamHandleInternal::handleError(GIOChannel* source, GIOCondition condition, gpointer data)
+{
+    LOG(Network, "SocketStreamHandle handleError");
+    SocketStreamHandleInternal* internal = reinterpret_cast<SocketStreamHandleInternal*>(data);
+    return internal->didError(condition);
+}
+
+/* static */
+gboolean SocketStreamHandleInternal::handleClose(gpointer data)
+{
+    LOG(Network, "SocketStreamHandle handleClose");
+    SocketStreamHandleInternal* internal = reinterpret_cast<SocketStreamHandleInternal*>(data);
+    internal->didClose();
+    return FALSE;
+}
+
+SocketStreamHandleInternal::SocketStreamHandleInternal(SocketStreamHandle* handle)
+    : m_handle(handle)
+    , m_addrinfo(NULL)
+    , m_sock(-1)
+    , m_g_io(NULL)
+    , m_in_watch(0)
+    , m_out_watch(0)
+    , m_err_watch(0)
+{
+    LOG(Network, "new SocketStreamHandleInternal %p", this);
+}
+
+SocketStreamHandleInternal::~SocketStreamHandleInternal()
+{
+    LOG(Network, "delete SocketStreamHandleInternal %p", this);
+    m_handle = NULL;
+    if (m_g_io) {
+        GError* gerr = NULL;
+        g_io_channel_shutdown(m_g_io, false, &gerr);
+        if (gerr) {
+            LOG(Network, "shutdown failed: %s", gerr->message);
+            g_error_free(gerr);
+        }
+        g_io_channel_unref(m_g_io);
+        m_g_io = NULL;
+    }
+    if (m_sock >= 0)
+        ::close(m_sock);
+    if (m_addrinfo)
+        freeaddrinfo(m_addrinfo);
+}
+
+void SocketStreamHandleInternal::connect(const KURL& url)
+{
+    LOG(Network, "SocketStreamHandleInternal %p connect %s", this, url.prettyURL().utf8().data());
+
+    struct addrinfo hints;
+    memset(&hints, 0, sizeof(hints));
+    hints.ai_family = AF_UNSPEC;
+    hints.ai_flags = AI_ADDRCONFIG;
+    hints.ai_socktype = SOCK_STREAM;
+
+    LOG(Network, "SocketStreamHandleInternal %p connect %s:%d", this, url.host().utf8().data(), url.port());
+    int err = getaddrinfo(url.host().utf8().data(), String::number(url.port()).utf8().data(), &hints, &m_addrinfo);
+    if (err) {
+        LOG(Network, "SocketStreamHandleInternal %p resolve error %s:%d", this, url.host().utf8().data(), url.port());
+        close();
+        return;
+    }
+    ASSERT(m_addrinfo);
+
+    if (m_handle->m_client)
+        m_handle->m_client->willOpenStream(m_handle, url);
+
+    struct addrinfo* rp;
+    for (rp = m_addrinfo; rp != NULL; rp = rp->ai_next) {
+        m_sock = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
+        if (m_sock < 0) {
+            LOG(Network, "SocketStreamHandleInternal %p socket error %d", this, errno);
+            continue;
+        }
+        m_g_io = g_io_channel_unix_new(m_sock);
+        ASSERT(m_g_io);
+
+        GError* gerr = NULL;
+        GIOStatus status = g_io_channel_set_encoding(m_g_io, NULL, &gerr);
+        ASSERT(status == G_IO_STATUS_NORMAL);
+        ASSERT(!gerr);
+
+        g_io_channel_set_buffered(m_g_io, FALSE);
+
+        m_out_watch = g_io_add_watch(m_g_io, G_IO_OUT, &SocketStreamHandleInternal::handleWrite, this);
+        m_err_watch = g_io_add_watch(m_g_io, static_cast<GIOCondition>(G_IO_ERR|G_IO_HUP), &SocketStreamHandleInternal::handleError, this);
+
+        status = g_io_channel_set_flags(m_g_io, G_IO_FLAG_NONBLOCK, &gerr);
+        ASSERT(status == G_IO_STATUS_NORMAL);
+        ASSERT(!gerr);
+
+        LOG(Network, "SocketStreamHandleInternal %p CONNECTING", this);
+        int err = ::connect(m_sock, rp->ai_addr, rp->ai_addrlen);
+        if (err == 0) {
+            LOG(Network, "SocketStreamHandleInternal %p connect ok", this);
+            m_handle->m_client->didOpen(m_handle);
+            return;
+        } else {
+            if (errno == EINPROGRESS)
+                return;
+            LOG(Network, "SocketStreamHandleInternal %p connect failed %d", this, errno);
+            m_out_watch = m_err_watch = 0;
+            g_io_channel_unref(m_g_io);
+            m_g_io = NULL;
+            ::close(m_sock);
+            m_sock = -1;
+        }
+    }
+    LOG(Network, "SocketStreamHandleInternal %p connect all failed", this);
+    close();
+}
+
+int SocketStreamHandleInternal::send(const char* buf, int size)
+{
+    LOG(Network, "SocketStreamHandleInternal %p send %d bytes", this, size);
+    if (!m_g_io) {
+        LOG_ERROR("SocketStreamHandleInternal %p already closed", this);
+        return -1;
+    }
+    if (!m_send_buffer.isEmpty()) {
+        LOG(Network, "SocketStreamHandleInternal %p send has buffer %d", this, m_send_buffer.size());
+        return 0;
+    }
+
+    m_send_buffer.append(buf, size);
+    if (m_out_watch == 0)
+        m_out_watch = g_io_add_watch(m_g_io, G_IO_OUT, &SocketStreamHandleInternal::handleWrite, this);
+    return size;
+}
+
+void SocketStreamHandleInternal::close()
+{
+    LOG(Network, "SocketStreamHandleInternal %p close", this);
+    if (m_handle->m_state != Closed)
+        g_idle_add(&SocketStreamHandleInternal::handleClose, this);
+}
+
+gboolean SocketStreamHandleInternal::canReceive()
+{
+    LOG(Network, "SocketStreamHandleInternal %p canReceive", this);
+    if (!m_g_io) {
+        LOG_ERROR("SocketStreamHandleInternal %p already closed", this);
+        return FALSE;
+    }
+    ASSERT(m_g_io);
+    gsize num_read;
+    GError* gerr = NULL;
+    GIOStatus status = g_io_channel_read_chars(m_g_io, m_read_buf, sizeof(m_read_buf), &num_read, &gerr);
+    if (status == G_IO_STATUS_AGAIN) {
+        LOG(Network, "SocketStreamHandleInternal %p read resource temporarily unavailable", this);
+        return TRUE;
+    }
+    if (status != G_IO_STATUS_NORMAL) {
+        LOG(Network, "SocketStreamHandleInternal %p read error %d %s", this, status, gerr ? gerr->message : "");
+        if (gerr)
+            g_error_free(gerr);
+        return didError(G_IO_ERR);
+    }
+    ASSERT(!gerr);
+    ASSERT(num_read < kSocketStreamHandleBufSize);
+
+    gchar tmp_buf[kSocketStreamHandleBufSize + 1];
+    memcpy(tmp_buf, m_read_buf, num_read);
+    tmp_buf[num_read] = '\0';
+
+    LOG(Network, "SocketStreamHandleInternal %p read %s (%lu)", this, tmp_buf, num_read);
+    if (m_handle && m_handle->m_client)
+        m_handle->m_client->didReceivedData(m_handle, m_read_buf, num_read);
+    return TRUE;
+}
+
+gboolean SocketStreamHandleInternal::canSend()
+{
+    LOG(Network, "SocketStreamHandleInternal %p canSend", this);
+    if (!m_g_io) {
+        LOG_ERROR("SocketStreamHandleInternal %p already closed", this);
+        return FALSE;
+    }
+    ASSERT(m_g_io);
+    if (m_handle->m_state == Connecting) {
+        int optval;
+        socklen_t optlen = sizeof(optval);
+        int err = getsockopt(m_sock, SOL_SOCKET, SO_ERROR, (void*)(&optval), &optlen);
+        if (err < 0 || optval != 0) {
+            // TODO(ukai): errno = EINPROGRESS or EALREADY?
+            LOG(Network, "SocketStreamHandleInternal %p async connect failed %d so_error=%d", this, errno, optval);
+            return didError(G_IO_ERR);
+        }
+        LOG(Network, "SocketStreamHandleInternal %p CONNECTED", this);
+        m_in_watch = g_io_add_watch(m_g_io, G_IO_IN, &SocketStreamHandleInternal::handleRead, this);
+        m_handle->m_state = Open;
+        // TODO(ukai): proxy, ssl.
+        LOG(Network, "SocketStreamHandleInternal %p didOpen", m_handle->m_client);
+        if (m_handle->m_client)
+            m_handle->m_client->didOpen(m_handle);
+    }
+
+    if (m_handle->m_client)
+        m_handle->m_client->willSendData(m_handle, m_send_buffer.data(), m_send_buffer.size());
+    gsize bytes_written;
+    GError* gerr = NULL;
+    GIOStatus status = g_io_channel_write_chars(m_g_io, m_send_buffer.data(), m_send_buffer.size(), &bytes_written, &gerr);
+    if (status == G_IO_STATUS_AGAIN) {
+        LOG(Network, "SocketStreamHandleInternal %p write resource temporarily unavailable", this);
+        return TRUE;
+    }
+    if (status != G_IO_STATUS_NORMAL) {
+        LOG(Network, "SocketStreamHandleInternal %p write error %d %s", this, status, gerr ? gerr->message : "");
+        if (gerr)
+            g_error_free(gerr);
+        return didError(G_IO_ERR);
+    }
+    LOG(Network, "SocketStreamHandleInternal %p write %lu bytes", this, bytes_written);
+    ASSERT(!gerr);
+
+    if (m_send_buffer.size() > bytes_written) {
+        LOG(Network, "SocketStreamHandleInternal %p send buf size %lu", this, m_send_buffer.size() - bytes_written);
+        Vector<char> buf;
+        buf.append(m_send_buffer.data() + bytes_written,
+                   m_send_buffer.size() - bytes_written);
+        m_send_buffer.swap(buf);
+        return TRUE;
+    }
+    // All bytes in m_send_buffer were sent.
+    LOG(Network, "SocketStreamHandleInternal %p no send buf", this);
+    m_send_buffer.clear();
+    m_out_watch = 0;
+    return FALSE;
+}
+
+gboolean SocketStreamHandleInternal::didError(GIOCondition condition)
+{
+    LOG(Network, "SocketStreamHandleInternal %p error", this);
+    m_handle->m_state = Closed;
+    SocketStreamError err(condition);
+    if (m_handle->m_client)
+        m_handle->m_client->didFail(m_handle, err);
+    m_handle->m_client = NULL;
+    return FALSE;
+}
+
+gboolean SocketStreamHandleInternal::didClose()
+{
+    LOG(Network, "SocketStreamHandleInternal %p close", this);
+    m_handle->m_state = Closed;
+    if (m_handle->m_client)
+        m_handle->m_client->didClose(m_handle);
+    m_handle->m_client = NULL;
+    return FALSE;
+}
+
+SocketStreamHandle::SocketStreamHandle(const KURL& url, SocketStreamHandleClient* client)
+        : m_url(url)
+{
+    setClient(client);
+    LOG(Network, "SocketStreamHandle %p new client %p", this, m_client);
+    internal = SocketStreamHandleInternal::create(this);
+    internal->connect(m_url);
+}
+
+SocketStreamHandle::~SocketStreamHandle()
+{
+    LOG(Network, "SocketStreamHandle %p delete", this);
+    setClient(NULL);
+    internal.clear();
+}
+
+int SocketStreamHandle::platformSend(const char *buf, int len)
+{
+    LOG(Network, "SocketStreamHandle %p platformSend %d", this, len);
+    if (internal.get())
+        return internal->send(buf, len);
+    return 0;
+}
+
+void SocketStreamHandle::platformClose()
+{
+    LOG(Network, "SocketStreamHandle %p platformClose", this);
+    if (internal.get())
+        internal->close();
+}
+
+void SocketStreamHandle::didReceivedAuthenticationChallenge(const AuthenticationChallenge& challenge)
+{
+    if (m_client)
+        m_client->didReceivedAuthenticationChallenge(this, challenge);
+}
+
+void SocketStreamHandle::receivedCredential(const AuthenticationChallenge& challenge, const Credential& credential)
+{
+    notImplemented();
+}
+
+void SocketStreamHandle::receivedRequestToContinueWithoutCredential(const AuthenticationChallenge& challenge)
+{
+    notImplemented();
+}
+
+void SocketStreamHandle::receivedCancellation(const AuthenticationChallenge& challenge)
+{
+    if (m_client)
+        m_client->receivedCancellation(this, challenge);
+}
+
+}  // namespace WebCore