Skip to content

Cleanup socket API #229

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mrabine merged 4 commits into from
Aug 27, 2026
Merged

Cleanup socket API #229

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
eee98c1
Cleanup socket API
mrabine Aug 27, 2026
1702d8e
fix empty datagram
mrabine Aug 27, 2026
ac2d127
ret value
mrabine Aug 27, 2026
79dc858
format
mrabine Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 0 additions & 75 deletions core/include/join/async_datagram_socket.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -276,81 +276,6 @@ namespace join
{
return this->_socket.ttl ();
}

protected:
/**
* @brief method called when an operation completes.
* @param op completed operation.
* @param result number of bytes transferred, or operation specific value.
*/
void onComplete (IoOperation* op, int result) override
{
dispatch (op, (result < 0) ? std::error_code (-result, std::generic_category ()) : std::error_code (),
(result > 0) ? static_cast<size_t> (result) : 0);
}

/**
* @brief method called when an operation is cancelled.
* @param op cancelled operation.
* @param result negative errno.
*/
void onCancel (IoOperation* op, [[maybe_unused]] int result) override
{
dispatch (op, make_error_code (std::errc::operation_canceled), 0);
}

/**
* @brief invoke the handler owning the given operation slot.
* @param op completed or cancelled operation.
* @param code error code to report.
* @param size number of bytes transferred.
*/
void dispatch (IoOperation* op, const std::error_code& code, size_t size) noexcept
{
if (op == &this->_ops->read.op)
{
this->_ops->read.state.store (AsyncOperation::Dispatching, std::memory_order_release);

ReadHandler handler = std::move (this->_onRead);
std::error_code result = code;

if (JOIN_LIKELY (!result))
{
if (JOIN_UNLIKELY (size == 0))
{
result = make_error_code (Errc::ConnectionClosed);
this->_socket._state = Socket::Disconnected;
}
else if (JOIN_UNLIKELY (this->_ops->readMsg.msg_flags & MSG_TRUNC))
{
result = make_error_code (Errc::MessageTooLong);
}
}

if (JOIN_LIKELY (handler))
{
handler (result, size);
}

AsyncOperation::State expected = AsyncOperation::Dispatching;
this->_ops->read.state.compare_exchange_strong (expected, AsyncOperation::Idle,
std::memory_order_release, std::memory_order_relaxed);
return;
}

this->_ops->write.state.store (AsyncOperation::Dispatching, std::memory_order_release);

WriteHandler handler = std::move (this->_onWrite);

if (JOIN_LIKELY (handler))
{
handler (code, size);
}

AsyncOperation::State expected = AsyncOperation::Dispatching;
this->_ops->write.state.compare_exchange_strong (expected, AsyncOperation::Idle, std::memory_order_release,
std::memory_order_relaxed);
}
};
}

Expand Down
11 changes: 2 additions & 9 deletions core/include/join/async_socket.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -547,16 +547,9 @@ namespace join
ReadHandler handler = std::move (_onRead);
std::error_code result = code;

if (JOIN_LIKELY (!result))
if (JOIN_UNLIKELY (!result && (_ops->readMsg.msg_flags & MSG_TRUNC)))
{
if (JOIN_UNLIKELY (size == 0))
{
result = make_error_code (Errc::ConnectionClosed); // LCOV_EXCL_LINE
}
else if (JOIN_UNLIKELY (_ops->readMsg.msg_flags & MSG_TRUNC))
{
result = make_error_code (Errc::MessageTooLong);
}
result = make_error_code (Errc::MessageTooLong);
}

if (JOIN_LIKELY (handler))
Expand Down
35 changes: 14 additions & 21 deletions core/include/join/datagram_socket.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,10 +39,6 @@ namespace join
template <class Protocol>
class BasicDatagramSocket final : public BasicSocket<Protocol>
{
/// friendship with basic asynchronous datagram socket
template <class P, class E>
friend class BasicAsyncDatagramSocket;

public:
using Ptr = std::unique_ptr<BasicDatagramSocket<Protocol>>;
using Mode = typename BasicSocket<Protocol>::Mode;
Expand Down Expand Up @@ -145,9 +141,11 @@ namespace join
if ((protocol.family () == AF_INET6) &&
(::setsockopt (this->_handle, IPPROTO_IPV6, IPV6_V6ONLY, &off, sizeof (off)) == -1))
{
// LCOV_EXCL_START
lastError = std::error_code (errno, std::generic_category ());
close ();
return -1;
// LCOV_EXCL_STOP
}
}

Expand All @@ -156,9 +154,11 @@ namespace join
if ((protocol.family () == AF_INET) &&
(::setsockopt (this->_handle, IPPROTO_IP, IP_HDRINCL, &off, sizeof (off)) == -1))
{
// LCOV_EXCL_START
lastError = std::error_code (errno, std::generic_category ());
close ();
return -1;
// LCOV_EXCL_STOP
}

this->setOption (Option::MulticastTtl, _ttl);
Expand All @@ -185,7 +185,7 @@ namespace join

if ((this->_state == State::Closed) && (open (endpoint.protocol ()) == -1))
{
return -1;
return -1; // LCOV_EXCL_LINE
}

if (::connect (this->_handle, endpoint.addr (), endpoint.length ()) == -1)
Expand Down Expand Up @@ -218,11 +218,13 @@ namespace join
sizeof (struct sockaddr_storage));
if (result == -1)
{
// LCOV_EXCL_START
if (errno != EAFNOSUPPORT)
{
lastError = std::error_code (errno, std::generic_category ());
return -1;
}
// LCOV_EXCL_STOP
}

this->_state = State::Disconnected;
Expand All @@ -248,7 +250,7 @@ namespace join
* @param endpoint endpoint from where data are coming (optional).
* @return The number of bytes received, -1 on failure.
*/
int readFrom (char* data, unsigned long maxSize, Endpoint* endpoint = nullptr) noexcept
ssize_t readFrom (char* data, size_t maxSize, Endpoint* endpoint = nullptr) noexcept
{
struct sockaddr_storage sa;

Expand All @@ -264,19 +266,10 @@ namespace join
message.msg_control = nullptr;
message.msg_controllen = 0;

int size = ::recvmsg (this->_handle, &message, 0);
if (size < 1)
ssize_t size = ::recvmsg (this->_handle, &message, 0);
if (size == -1)
{
if (size == -1)
{
lastError = std::error_code (errno, std::generic_category ());
}
else
{
lastError = make_error_code (Errc::ConnectionClosed);
this->_state = State::Disconnected;
}

lastError = std::error_code (errno, std::generic_category ());
return -1;
}

Expand All @@ -301,14 +294,14 @@ namespace join
* @param endpoint endpoint where to write the data.
* @return the number of bytes written, -1 on failure.
*/
int writeTo (const char* data, unsigned long maxSize, const Endpoint& endpoint) noexcept
ssize_t writeTo (const char* data, size_t maxSize, const Endpoint& endpoint) noexcept
{
if ((this->_state == State::Closed) && (open (endpoint.protocol ()) == -1))
{
return -1;
return -1; // LCOV_EXCL_LINE
}

int result = ::sendto (this->_handle, data, maxSize, 0, endpoint.addr (), endpoint.length ());
ssize_t result = ::sendto (this->_handle, data, maxSize, 0, endpoint.addr (), endpoint.length ());
if (result < 0)
{
lastError = std::error_code (errno, std::generic_category ());
Expand Down
54 changes: 7 additions & 47 deletions core/include/join/socket.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,7 +302,7 @@ namespace join
* @brief get the number of readable bytes.
* @return the number of readable bytes, -1 on failure.
*/
int canRead () const noexcept
ssize_t canRead () const noexcept
{
int available = 0;

Expand Down Expand Up @@ -332,7 +332,7 @@ namespace join
* @param maxSize maximum number of bytes to read.
* @return the number of bytes received, -1 on failure.
*/
int read (char* data, unsigned long maxSize) noexcept
ssize_t read (char* data, size_t maxSize) noexcept
{
struct iovec iov;
iov.iov_base = data;
Expand All @@ -346,18 +346,10 @@ namespace join
message.msg_control = nullptr;
message.msg_controllen = 0;

int size = ::recvmsg (_handle, &message, 0);
if (size < 1)
ssize_t size = ::recvmsg (_handle, &message, 0);
if (size == -1)
{
if (size == -1)
{
lastError = std::error_code (errno, std::generic_category ());
}
else
{
lastError = make_error_code (Errc::ConnectionClosed);
}

lastError = std::error_code (errno, std::generic_category ());
return -1;
}

Expand Down Expand Up @@ -386,7 +378,7 @@ namespace join
* @param maxSize maximum number of bytes to write.
* @return the number of bytes written, -1 on failure.
*/
int write (const char* data, unsigned long maxSize) noexcept
ssize_t write (const char* data, size_t maxSize) noexcept
{
struct iovec iov;
iov.iov_base = const_cast<char*> (data);
Expand All @@ -400,7 +392,7 @@ namespace join
message.msg_control = nullptr;
message.msg_controllen = 0;

int result = ::sendmsg (_handle, &message, 0);
ssize_t result = ::sendmsg (_handle, &message, 0);
if (result == -1)
{
lastError = std::error_code (errno, std::generic_category ());
Expand Down Expand Up @@ -629,38 +621,6 @@ namespace join
return _handle;
}

/**
* @brief get standard 1s complement checksum.
* @param data data pointer.
* @param len data len.
* @param current Current sum.
* @return checksum.
*/
static uint16_t checksum (const uint16_t* data, size_t len, uint16_t current = 0)
{
uint32_t sum = current;

while (len > 1)
{
sum += *data++;
len -= 2;
}

if (len == 1)
{
#if __BYTE_ORDER == __LITTLE_ENDIAN
sum += *reinterpret_cast<const uint8_t*> (data);
#else
sum += *reinterpret_cast<const uint8_t*> (data) << 8;
#endif
}

sum = (sum >> 16) + (sum & 0xffff);
sum += (sum >> 16);

return static_cast<uint16_t> (~sum);
}

/**
* @brief wait for the socket handle to become ready.
* @param wantRead set to true if want read
Expand Down
2 changes: 1 addition & 1 deletion core/include/join/socket_stream.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,7 +241,7 @@ namespace join
{
for (;;)
{
int nread = _socket.read (eback (), _bufsize);
ssize_t nread = _socket.read (eback (), _bufsize);
if (nread == -1)
{
if (lastError == Errc::TemporaryError)
Expand Down
30 changes: 24 additions & 6 deletions core/include/join/stream_socket.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -288,20 +288,38 @@ namespace join
_remote = {};
}

/**
* @brief read data.
* @param data buffer used to store the data received.
* @param maxSize maximum number of bytes to read.
* @return the number of bytes received, -1 on failure.
*/
ssize_t read (char* data, size_t maxSize) noexcept
{
ssize_t size = BasicSocket<Protocol>::read (data, maxSize);
if (size == 0)
{
lastError = make_error_code (Errc::ConnectionClosed);
return -1;
}

return size;
}

/**
* @brief read data until size is reached or an error occurred.
* @param data buffer used to store the data received.
* @param size number of bytes to read.
* @param timeout timeout in milliseconds.
* @return 0 on success, -1 on failure.
*/
int readExactly (char* data, unsigned long size, int timeout = 0) noexcept
int readExactly (char* data, size_t size, int timeout = 0) noexcept
{
unsigned long numRead = 0;
size_t numRead = 0;

while (numRead < size)
{
int result = this->read (data + numRead, size - numRead);
ssize_t result = this->read (data + numRead, size - numRead);
if (result == -1)
{
if (lastError == Errc::TemporaryError)
Expand All @@ -328,13 +346,13 @@ namespace join
* @param timeout timeout in milliseconds.
* @return 0 on success, -1 on failure.
*/
int writeExactly (const char* data, unsigned long size, int timeout = 0) noexcept
int writeExactly (const char* data, size_t size, int timeout = 0) noexcept
{
unsigned long numWrite = 0;
size_t numWrite = 0;

while (numWrite < size)
{
int result = this->write (data + numWrite, size - numWrite);
ssize_t result = this->write (data + numWrite, size - numWrite);
if (result == -1)
{
if (lastError == Errc::TemporaryError)
Expand Down
Loading
Loading