SocketStreamHandleImplSoup.cpp [plain text]
#include "config.h"
#include "SocketStreamHandleImpl.h"
#if USE(SOUP)
#include "Logging.h"
#include "SocketStreamError.h"
#include "SocketStreamHandleClient.h"
#include "URL.h"
#include <gio/gio.h>
#include <glib.h>
#include <wtf/Vector.h>
#include <wtf/glib/GUniquePtr.h>
#include <wtf/text/CString.h>
#define READ_BUFFER_SIZE 1024
namespace WebCore {
Ref<SocketStreamHandleImpl> SocketStreamHandleImpl::create(const URL& url, SocketStreamHandleClient& client, SessionID)
{
Ref<SocketStreamHandleImpl> socket = adoptRef(*new SocketStreamHandleImpl(url, client));
unsigned port = url.port() ? url.port().value() : (url.protocolIs("wss") ? 443 : 80);
GRefPtr<GSocketClient> socketClient = adoptGRef(g_socket_client_new());
if (url.protocolIs("wss"))
g_socket_client_set_tls(socketClient.get(), TRUE);
Ref<SocketStreamHandle> protectedSocketStreamHandle = socket.copyRef();
g_socket_client_connect_to_host_async(socketClient.get(), url.host().utf8().data(), port, socket->m_cancellable.get(),
reinterpret_cast<GAsyncReadyCallback>(connectedCallback), &protectedSocketStreamHandle.leakRef());
return socket;
}
Ref<SocketStreamHandle> SocketStreamHandleImpl::create(GSocketConnection* socketConnection, SocketStreamHandleClient& client)
{
Ref<SocketStreamHandleImpl> socket = adoptRef(*new SocketStreamHandleImpl(URL(), client));
GRefPtr<GSocketConnection> connection = socketConnection;
socket->connected(WTFMove(connection));
return WTFMove(socket);
}
SocketStreamHandleImpl::SocketStreamHandleImpl(const URL& url, SocketStreamHandleClient& client)
: SocketStreamHandle(url, client)
, m_cancellable(adoptGRef(g_cancellable_new()))
{
LOG(Network, "SocketStreamHandle %p new client %p", this, &m_client);
}
SocketStreamHandleImpl::~SocketStreamHandleImpl()
{
LOG(Network, "SocketStreamHandle %p delete", this);
}
void SocketStreamHandleImpl::connected(GRefPtr<GSocketConnection>&& socketConnection)
{
m_socketConnection = WTFMove(socketConnection);
m_outputStream = G_POLLABLE_OUTPUT_STREAM(g_io_stream_get_output_stream(G_IO_STREAM(m_socketConnection.get())));
m_inputStream = g_io_stream_get_input_stream(G_IO_STREAM(m_socketConnection.get()));
m_readBuffer = std::make_unique<char[]>(READ_BUFFER_SIZE);
RefPtr<SocketStreamHandleImpl> protectedThis(this);
g_input_stream_read_async(m_inputStream.get(), m_readBuffer.get(), READ_BUFFER_SIZE, G_PRIORITY_DEFAULT, m_cancellable.get(),
reinterpret_cast<GAsyncReadyCallback>(readReadyCallback), protectedThis.leakRef());
m_state = Open;
m_client.didOpenSocketStream(*this);
}
void SocketStreamHandleImpl::connectedCallback(GSocketClient* client, GAsyncResult* result, SocketStreamHandleImpl* handle)
{
RefPtr<SocketStreamHandle> protectedThis = adoptRef(handle);
GUniqueOutPtr<GError> error;
GRefPtr<GSocketConnection> socketConnection = adoptGRef(g_socket_client_connect_to_host_finish(client, result, &error.outPtr()));
if (g_cancellable_is_cancelled(handle->m_cancellable.get())) {
if (socketConnection)
g_io_stream_close(G_IO_STREAM(socketConnection.get()), nullptr, nullptr);
return;
}
if (error)
handle->didFail(SocketStreamError(error->code, String(), error->message));
else
handle->connected(WTFMove(socketConnection));
}
void SocketStreamHandleImpl::readBytes(gssize bytesRead)
{
if (!bytesRead) {
close();
return;
}
RefPtr<SocketStreamHandle> protectedThis(this);
std::optional<size_t> optionalLength;
if (bytesRead != -1)
optionalLength = static_cast<size_t>(bytesRead);
m_client.didReceiveSocketStreamData(*this, m_readBuffer.get(), optionalLength);
if (m_inputStream) {
g_input_stream_read_async(m_inputStream.get(), m_readBuffer.get(), READ_BUFFER_SIZE, G_PRIORITY_DEFAULT, m_cancellable.get(),
reinterpret_cast<GAsyncReadyCallback>(readReadyCallback), protectedThis.leakRef());
}
}
void SocketStreamHandleImpl::readReadyCallback(GInputStream* stream, GAsyncResult* result, SocketStreamHandleImpl* handle)
{
RefPtr<SocketStreamHandle> protectedThis = adoptRef(handle);
GUniqueOutPtr<GError> error;
gssize bytesRead = g_input_stream_read_finish(stream, result, &error.outPtr());
if (g_cancellable_is_cancelled(handle->m_cancellable.get()))
return;
if (error)
handle->didFail(SocketStreamError(error->code, String(), error->message));
else
handle->readBytes(bytesRead);
}
void SocketStreamHandleImpl::didFail(SocketStreamError&& error)
{
m_client.didFailSocketStream(*this, WTFMove(error));
}
void SocketStreamHandleImpl::writeReady()
{
if (!bufferedAmount()) {
stopWaitingForSocketWritability();
return;
}
sendPendingData();
}
std::optional<size_t> SocketStreamHandleImpl::platformSend(const char* data, size_t length)
{
LOG(Network, "SocketStreamHandle %p platformSend", this);
if (!m_outputStream || !data)
return std::nullopt;
GUniqueOutPtr<GError> error;
gssize written = g_pollable_output_stream_write_nonblocking(m_outputStream.get(), data, length, m_cancellable.get(), &error.outPtr());
if (error) {
if (g_error_matches(error.get(), G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
beginWaitingForSocketWritability();
else
didFail(SocketStreamError(error->code, String(), error->message));
return std::nullopt;
}
if (written == -1 || static_cast<size_t>(written) < length)
beginWaitingForSocketWritability();
if (written == -1)
return std::nullopt;
return static_cast<size_t>(written);
}
void SocketStreamHandleImpl::platformClose()
{
LOG(Network, "SocketStreamHandle %p platformClose", this);
g_cancellable_cancel(m_cancellable.get());
stopWaitingForSocketWritability();
if (m_socketConnection) {
GUniqueOutPtr<GError> error;
g_io_stream_close(G_IO_STREAM(m_socketConnection.get()), nullptr, &error.outPtr());
if (error)
didFail(SocketStreamError(error->code, String(), error->message));
m_socketConnection = nullptr;
}
m_outputStream = nullptr;
m_inputStream = nullptr;
m_readBuffer = nullptr;
m_client.didCloseSocketStream(*this);
}
void SocketStreamHandleImpl::beginWaitingForSocketWritability()
{
if (m_writeReadySource) return;
m_writeReadySource = adoptGRef(g_pollable_output_stream_create_source(m_outputStream.get(), m_cancellable.get()));
ref();
g_source_set_callback(m_writeReadySource.get(), reinterpret_cast<GSourceFunc>(writeReadyCallback), this, [](gpointer handle) {
static_cast<SocketStreamHandleImpl*>(handle)->deref();
});
g_source_attach(m_writeReadySource.get(), g_main_context_get_thread_default());
}
void SocketStreamHandleImpl::stopWaitingForSocketWritability()
{
if (!m_writeReadySource) return;
g_source_destroy(m_writeReadySource.get());
m_writeReadySource = nullptr;
}
gboolean SocketStreamHandleImpl::writeReadyCallback(GPollableOutputStream*, SocketStreamHandleImpl* handle)
{
if (g_cancellable_is_cancelled(handle->m_cancellable.get()))
return G_SOURCE_REMOVE;
handle->writeReady();
return G_SOURCE_CONTINUE;
}
}
#endif