#include "config.h"
#include "CurlStream.h"
#include "CurlStreamScheduler.h"
#include "SocketStreamError.h"
#if USE(CURL)
namespace WebCore {
CurlStream::CurlStream(CurlStreamScheduler& scheduler, CurlStreamID streamID, URL&& url)
: m_scheduler(scheduler)
, m_streamID(streamID)
{
ASSERT(!isMainThread());
m_curlHandle = WTF::makeUnique<CurlHandle>();
url.setProtocol(url.protocolIs("wss") ? "https" : "http");
m_curlHandle->setUrl(WTFMove(url));
m_curlHandle->enableConnectionOnly();
auto errorCode = m_curlHandle->perform();
if (errorCode != CURLE_OK) {
notifyFailure(errorCode);
return;
}
m_scheduler.callClientOnMainThread(m_streamID, [streamID = m_streamID](Client& client) {
client.didOpen(streamID);
});
}
CurlStream::~CurlStream()
{
ASSERT(!isMainThread());
destroyHandle();
}
void CurlStream::destroyHandle()
{
if (!m_curlHandle)
return;
m_curlHandle = nullptr;
}
void CurlStream::send(UniqueArray<uint8_t>&& buffer, size_t length)
{
ASSERT(!isMainThread());
if (!m_curlHandle)
return;
m_sendBuffers.append(std::make_pair(WTFMove(buffer), length));
}
void CurlStream::appendMonitoringFd(fd_set& readfds, fd_set& writefds, fd_set& exceptfds, int& maxfd)
{
ASSERT(!isMainThread());
if (!m_curlHandle)
return;
auto socket = m_curlHandle->getActiveSocket();
if (!socket.has_value()) {
notifyFailure(socket.error());
return;
}
FD_SET(*socket, &readfds);
FD_SET(*socket, &exceptfds);
if (m_sendBuffers.size())
FD_SET(*socket, &writefds);
if (maxfd < *socket)
maxfd = *socket;
}
void CurlStream::tryToTransfer(const fd_set& readfds, const fd_set& writefds, const fd_set& exceptfds)
{
ASSERT(!isMainThread());
if (!m_curlHandle)
return;
auto socket = m_curlHandle->getActiveSocket();
if (!socket.has_value()) {
notifyFailure(socket.error());
return;
}
if (FD_ISSET(*socket, &readfds) || FD_ISSET(*socket, &exceptfds))
tryToReceive();
if (FD_ISSET(*socket, &writefds))
tryToSend();
}
void CurlStream::tryToReceive()
{
if (!m_curlHandle)
return;
auto receiveBuffer = makeUniqueArray<uint8_t>(kReceiveBufferSize);
size_t bytesReceived = 0;
auto errorCode = m_curlHandle->receive(receiveBuffer.get(), kReceiveBufferSize, bytesReceived);
if (errorCode != CURLE_OK) {
if (errorCode != CURLE_AGAIN)
notifyFailure(errorCode);
return;
}
if (!bytesReceived)
destroyHandle();
m_scheduler.callClientOnMainThread(m_streamID, [streamID = m_streamID, buffer = WTFMove(receiveBuffer), length = bytesReceived](Client& client) mutable {
client.didReceiveData(streamID, reinterpret_cast<const char*>(buffer.get()), length);
});
}
void CurlStream::tryToSend()
{
if (!m_curlHandle || !m_sendBuffers.size())
return;
auto& [buffer, length] = m_sendBuffers.first();
size_t bytesSent = 0;
auto errorCode = m_curlHandle->send(buffer.get() + m_sendBufferOffset, length - m_sendBufferOffset, bytesSent);
if (errorCode != CURLE_OK) {
if (errorCode != CURLE_AGAIN)
notifyFailure(errorCode);
return;
}
m_sendBufferOffset += bytesSent;
if (m_sendBufferOffset >= length) {
m_sendBuffers.remove(0);
m_sendBufferOffset = 0;
}
m_scheduler.callClientOnMainThread(m_streamID, [streamID = m_streamID, length = bytesSent](Client& client) {
client.didSendData(streamID, length);
});
}
void CurlStream::notifyFailure(CURLcode errorCode)
{
destroyHandle();
m_scheduler.callClientOnMainThread(m_streamID, [streamID = m_streamID, errorCode](Client& client) mutable {
client.didFail(streamID, errorCode);
});
}
}
#endif