/* See LICENSE file for copyright and license details. */
#include "common.h"
/**
* Wait for the next message to be received
*
* @param ctx The state of the library, must be connected
* @param pending Information for each pending request
* @param n The number of elements in `pending`
* @param selected The index of the element in `pending` which corresponds
* to the first inbound message, note that this only means
* that the message is not for any of the other request,
* if the message is corrupt any of the listed requests can
* be selected even if it is not for any of the requests.
* Functions that parse the message will detect such corruption.
* @return Zero on success, -1 on error. If the the message is ignored,
* which happens if corresponding `libcoopgamma_async_context_t`
* is not listed, -1 is returned and `errno` is set to 0. If -1
* is returned, `errno` is set to `ENOTRECOVERABLE` you have
* received a corrupt message and the context has been tainted
* beyond recover.
*/
int
libcoopgamma_synchronise(libcoopgamma_context_t *restrict ctx, libcoopgamma_async_context_t *restrict pending,
size_t n, size_t *restrict selected)
{
char temp[3 * sizeof(size_t) + 1];
ssize_t got;
size_t i;
char *p;
char *line;
char *value;
struct pollfd pollfd;
size_t new_size;
void *new;
if (ctx->inbound_head == ctx->inbound_tail) {
ctx->inbound_head = ctx->inbound_tail = ctx->curline = 0;
} else if (ctx->inbound_tail > 0) {
memmove(ctx->inbound, ctx->inbound + ctx->inbound_tail, ctx->inbound_head -= ctx->inbound_tail);
ctx->curline -= ctx->inbound_tail;
ctx->inbound_tail = 0;
}
pollfd.fd = ctx->fd;
pollfd.events = POLLIN | POLLRDNORM | POLLRDBAND | POLLPRI;
if (ctx->inbound_head)
goto skip_recv;
for (;;) {
if (ctx->inbound_head == ctx->inbound_size) {
new_size = ctx->inbound_size ? (ctx->inbound_size << 1) : 1024U;
new = realloc(ctx->inbound, new_size);
if (!new)
return -1;
ctx->inbound = new;
ctx->inbound_size = new_size;
}
if (ctx->blocking) {
pollfd.revents = 0;
if (poll(&pollfd, (nfds_t)1, -1) < 0)
return -1;
}
got = recv(ctx->fd, &ctx->inbound[ctx->inbound_head], ctx->inbound_size - ctx->inbound_head, 0);
if (got <= 0) {
if (got == 0)
errno = ECONNRESET;
return -1;
}
#ifdef DEBUG_MODE
fprintf(stderr, "\033[32m");
fwrite(&ctx->inbound[ctx->inbound_head], (size_t)got, 1U, stderr);
fprintf(stderr, "\033[m");
fflush(stderr);
#endif
ctx->inbound_head += (size_t)got;
skip_recv:
while (!ctx->have_all_headers) {
line = &ctx->inbound[ctx->curline];
p = memchr(line, '\n', ctx->inbound_head - ctx->curline);
if (!p)
break;
if (memchr(line, '\0', (size_t)(p - line)) != NULL)
ctx->bad_message = 1;
*p++ = '\0';
ctx->curline = (size_t)(p - ctx->inbound);
if (!*line) {
ctx->have_all_headers = 1;
} else if (strstr(line, "In response to: ") == line) {
value = &line[sizeof("In response to: ") - 1U];
ctx->in_response_to = (uint32_t)atol(value);
} else if (strstr(line, "Length: ") == line) {
value = &line[sizeof("Length: ") - 1U];
ctx->length = (size_t)atol(value);
sprintf(temp, "%zu", ctx->length);
if (strcmp(value, temp))
goto fatal;
}
}
if (ctx->have_all_headers && ctx->inbound_head >= ctx->curline + ctx->length) {
ctx->curline += ctx->length;
if (ctx->bad_message) {
ctx->bad_message = 0;
ctx->have_all_headers = 0;
ctx->length = 0;
ctx->inbound_tail = ctx->curline;
errno = EBADMSG;
return -1;
}
for (i = 0; i < n; i++) {
if (pending[i].message_id == ctx->in_response_to) {
*selected = i;
return 0;
}
}
*selected = 0;
ctx->bad_message = 0;
ctx->have_all_headers = 0;
ctx->length = 0;
ctx->inbound_tail = ctx->curline;
errno = 0;
return -1;
}
}
fatal:
errno = ENOTRECOVERABLE;
return -1;
}
