forked from enlightenment/efl
efl_io_queue: basic class to interact with Efl.Io interfaces.
The use of low-level interfaces such as Efl.Io.Reader and Efl.Io.Writer are not that user-friendly as they can handle partial data. Classes such as Efl.Io.Copier makes them easy to use, but they need a reader (source) or writer (destination) and in our examples we used fixed buffers or some existing streams (stdin/stdout/stderr, networking...). However, if interactively we need to produce some data to be sent, such as implementing some networking protocols, we'd have to write our own Efl.Io.Reader and Efl.Io.Writer classes to handle the buffering. Not anymore! With Efl.Io.Queue you can write stuff to it and it will buffer to memory. Once stuff is read, it will automatically remove those bytes from buffer.
This commit is contained in:
Gustavo Sverzut Barbieri
parent
a1526169e7
commit
86e87b2fd9
|
|
@ -52,6 +52,7 @@ efl_eolian_files = \
|
|||
lib/efl/interfaces/efl_io_sizer.eo \
|
||||
lib/efl/interfaces/efl_io_writer.eo \
|
||||
lib/efl/interfaces/efl_io_buffer.eo \
|
||||
lib/efl/interfaces/efl_io_queue.eo \
|
||||
$(efl_eolian_legacy_files) \
|
||||
$(NULL)
|
||||
|
||||
|
|
@ -100,6 +101,7 @@ lib/efl/interfaces/efl_io_reader.c \
|
|||
lib/efl/interfaces/efl_io_sizer.c \
|
||||
lib/efl/interfaces/efl_io_writer.c \
|
||||
lib/efl/interfaces/efl_io_buffer.c \
|
||||
lib/efl/interfaces/efl_io_queue.c \
|
||||
$(NULL)
|
||||
|
||||
lib_efl_libefl_la_CPPFLAGS = -I$(top_builddir)/src/lib/efl -I$(top_srcdir)/src/lib/efl @EFL_CFLAGS@ -DEFL_GFX_FILTER_BETA
|
||||
|
|
|
|||
|
|
@ -48,5 +48,6 @@
|
|||
/ecore_buffer_provider_example
|
||||
/efl_io_copier_example
|
||||
/efl_io_copier_simple_example
|
||||
/efl_io_queue_example
|
||||
/efl_net_server_example
|
||||
/efl_net_dialer_http_example
|
||||
|
|
|
|||
|
|
@ -79,6 +79,7 @@ ecore_con_eet_client_example \
|
|||
ecore_con_eet_server_example \
|
||||
efl_io_copier_example \
|
||||
efl_io_copier_simple_example \
|
||||
efl_io_queue_example \
|
||||
efl_net_server_example \
|
||||
efl_net_dialer_http_example
|
||||
|
||||
|
|
@ -287,6 +288,9 @@ efl_io_copier_example_LDADD = $(ECORE_CON_COMMON_LDADD)
|
|||
efl_io_copier_simple_example_SOURCES = efl_io_copier_simple_example.c
|
||||
efl_io_copier_simple_example_LDADD = $(ECORE_COMMON_LDADD)
|
||||
|
||||
efl_io_queue_example_SOURCES = efl_io_queue_example.c
|
||||
efl_io_queue_example_LDADD = $(ECORE_CON_COMMON_LDADD)
|
||||
|
||||
efl_net_server_example_SOURCES = efl_net_server_example.c
|
||||
efl_net_server_example_LDADD = $(ECORE_CON_COMMON_LDADD)
|
||||
|
||||
|
|
@ -341,6 +345,7 @@ ecore_con_eet_server_example.c \
|
|||
ecore_con_eet_descriptor_example.c \
|
||||
efl_io_copier_example.c \
|
||||
efl_io_copier_simple_example.c \
|
||||
efl_io_queue_example.c \
|
||||
efl_net_server_example.c \
|
||||
efl_net_dialer_http_example.c
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,374 @@
|
|||
#define EFL_BETA_API_SUPPORT 1
|
||||
#define EFL_EO_API_SUPPORT 1
|
||||
#include <Ecore.h>
|
||||
#include <Ecore_Getopt.h>
|
||||
#include <Ecore_Con.h>
|
||||
|
||||
static int retval = EXIT_SUCCESS;
|
||||
static Eina_List *waiting = NULL;
|
||||
static Eina_List *commands = NULL;
|
||||
static Eina_Slice line_delimiter;
|
||||
static Eo *send_queue, *receive_queue;
|
||||
|
||||
static void
|
||||
_command_next(void)
|
||||
{
|
||||
Eina_Slice slice;
|
||||
char *cmd;
|
||||
|
||||
if (!commands)
|
||||
{
|
||||
efl_io_queue_eos_mark(send_queue);
|
||||
return;
|
||||
}
|
||||
|
||||
cmd = commands->data;
|
||||
commands = eina_list_remove_list(commands, commands);
|
||||
|
||||
slice = (Eina_Slice)EINA_SLICE_STR(cmd);
|
||||
efl_io_writer_write(send_queue, &slice, NULL);
|
||||
fprintf(stderr, "INFO: sent '" EINA_SLICE_STR_FMT "'\n",
|
||||
EINA_SLICE_STR_PRINT(slice));
|
||||
|
||||
/* don't use line_delimiter directly, 'len' may be changed! */
|
||||
slice = line_delimiter;
|
||||
efl_io_writer_write(send_queue, &slice, NULL);
|
||||
free(cmd);
|
||||
}
|
||||
|
||||
static void
|
||||
_receiver_data(void *data EINA_UNUSED, const Eo_Event *event)
|
||||
{
|
||||
Eina_Slice slice;
|
||||
|
||||
if (!efl_io_queue_slice_get(event->object, &slice)) return;
|
||||
|
||||
/* this will happen when we're called when we issue our own
|
||||
* efl_io_queue_clear() below.
|
||||
*/
|
||||
if (slice.len == 0) return;
|
||||
|
||||
if (slice.len < line_delimiter.len)
|
||||
{
|
||||
fprintf(stderr, "ERROR: received short line '" EINA_SLICE_STR_FMT "'\n",
|
||||
EINA_SLICE_STR_PRINT(slice));
|
||||
}
|
||||
else if (memcmp(slice.bytes + slice.len - line_delimiter.len,
|
||||
line_delimiter.bytes, line_delimiter.len) != 0)
|
||||
{
|
||||
fprintf(stderr, "WARNING: received without line-delimiter '"
|
||||
EINA_SLICE_STR_FMT "'\n",
|
||||
EINA_SLICE_STR_PRINT(slice));
|
||||
}
|
||||
else
|
||||
{
|
||||
slice.len -= line_delimiter.len;
|
||||
fprintf(stderr, "INFO: received '" EINA_SLICE_STR_FMT "'\n",
|
||||
EINA_SLICE_STR_PRINT(slice));
|
||||
}
|
||||
|
||||
efl_io_queue_clear(event->object);
|
||||
_command_next();
|
||||
}
|
||||
|
||||
static void
|
||||
_dialer_connected(void *data EINA_UNUSED, const Eo_Event *event)
|
||||
{
|
||||
fprintf(stderr, "INFO: connected to %s (%s)\n",
|
||||
efl_net_dialer_address_dial_get(event->object),
|
||||
efl_net_socket_address_remote_get(event->object));
|
||||
|
||||
_command_next();
|
||||
}
|
||||
|
||||
static void
|
||||
_copier_done(void *data EINA_UNUSED, const Eo_Event *event)
|
||||
{
|
||||
fprintf(stderr, "INFO: %s done\n", efl_name_get(event->object));
|
||||
|
||||
waiting = eina_list_remove(waiting, event->object);
|
||||
if (!waiting)
|
||||
ecore_main_loop_quit();
|
||||
}
|
||||
|
||||
static void
|
||||
_copier_error(void *data EINA_UNUSED, const Eo_Event *event)
|
||||
{
|
||||
const Eina_Error *perr = event->info;
|
||||
fprintf(stderr, "INFO: %s error: #%d '%s'\n",
|
||||
efl_name_get(event->object), *perr, eina_error_msg_get(*perr));
|
||||
retval = EXIT_FAILURE;
|
||||
ecore_main_loop_quit();
|
||||
}
|
||||
|
||||
EFL_CALLBACKS_ARRAY_DEFINE(copier_cbs,
|
||||
{ EFL_IO_COPIER_EVENT_DONE, _copier_done },
|
||||
{ EFL_IO_COPIER_EVENT_ERROR, _copier_error });
|
||||
|
||||
static char *
|
||||
_unescape(const char *str)
|
||||
{
|
||||
char *ret = strdup(str);
|
||||
char *c, *w;
|
||||
Eina_Bool escaped = EINA_FALSE;
|
||||
|
||||
for (c = ret, w = ret; *c != '\0'; c++)
|
||||
{
|
||||
if (escaped)
|
||||
{
|
||||
escaped = EINA_FALSE;
|
||||
switch (*c)
|
||||
{
|
||||
case 'n': *w = '\n'; break;
|
||||
case 'r': *w = '\r'; break;
|
||||
case 't': *w = '\t'; break;
|
||||
default: w++; /* no change */
|
||||
}
|
||||
w++;
|
||||
}
|
||||
else
|
||||
{
|
||||
if (*c == '\\')
|
||||
escaped = EINA_TRUE;
|
||||
else
|
||||
w++;
|
||||
}
|
||||
}
|
||||
*w = '\0';
|
||||
return ret;
|
||||
}
|
||||
|
||||
static const Ecore_Getopt options = {
|
||||
"efl_io_queue_example", /* program name */
|
||||
NULL, /* usage line */
|
||||
"1", /* version */
|
||||
"(C) 2016 Enlightenment Project", /* copyright */
|
||||
"BSD 2-Clause", /* license */
|
||||
/* long description, may be multiline and contain \n */
|
||||
"Example of Efl_Io_Queue usage.\n"
|
||||
"\n"
|
||||
"This uses Efl_Io_Queue to easily interface with Efl_Io_Copier in order to "
|
||||
"send commands to a TCP server.",
|
||||
EINA_FALSE,
|
||||
{
|
||||
ECORE_GETOPT_STORE_STR('d', "line-delimiter",
|
||||
"Changes the line delimiter to be used in both send and receive. Defaults to \\r\\n"),
|
||||
ECORE_GETOPT_STORE_ULONG('l', "buffer-limit",
|
||||
"If set will limit buffer size to this limit of bytes. If used alongside with --line-delimiter and that delimiter was not found but bffer limit was reached, the line event will be triggered without the delimiter at the end."),
|
||||
ECORE_GETOPT_VERSION('V', "version"),
|
||||
ECORE_GETOPT_COPYRIGHT('C', "copyright"),
|
||||
ECORE_GETOPT_LICENSE('L', "license"),
|
||||
ECORE_GETOPT_HELP('h', "help"),
|
||||
|
||||
ECORE_GETOPT_STORE_METAVAR_STR(0, NULL,
|
||||
"The server address as\n"
|
||||
"IP:PORT to connect using TCP and an IPv4 (A.B.C.D:PORT) or IPv6 ([A:B:C:D::E]:PORT).\n",
|
||||
"server_address"),
|
||||
ECORE_GETOPT_APPEND_METAVAR(0, NULL,
|
||||
"Commands to send",
|
||||
"commands",
|
||||
ECORE_GETOPT_TYPE_STR),
|
||||
ECORE_GETOPT_SENTINEL
|
||||
}
|
||||
};
|
||||
|
||||
int
|
||||
main(int argc, char **argv)
|
||||
{
|
||||
char *address = NULL;
|
||||
char *line_delimiter_str = NULL;
|
||||
char *cmd;
|
||||
unsigned long buffer_limit = 0;
|
||||
Eina_Bool quit_option = EINA_FALSE;
|
||||
Ecore_Getopt_Value values[] = {
|
||||
ECORE_GETOPT_VALUE_STR(line_delimiter_str),
|
||||
ECORE_GETOPT_VALUE_ULONG(buffer_limit),
|
||||
|
||||
/* standard block to provide version, copyright, license and help */
|
||||
ECORE_GETOPT_VALUE_BOOL(quit_option), /* -V/--version quits */
|
||||
ECORE_GETOPT_VALUE_BOOL(quit_option), /* -C/--copyright quits */
|
||||
ECORE_GETOPT_VALUE_BOOL(quit_option), /* -L/--license quits */
|
||||
ECORE_GETOPT_VALUE_BOOL(quit_option), /* -h/--help quits */
|
||||
|
||||
/* positional argument */
|
||||
ECORE_GETOPT_VALUE_STR(address),
|
||||
ECORE_GETOPT_VALUE_LIST(commands),
|
||||
|
||||
ECORE_GETOPT_VALUE_NONE /* sentinel */
|
||||
};
|
||||
Eina_Error err;
|
||||
int args;
|
||||
Eo *dialer, *sender, *receiver, *loop;
|
||||
|
||||
ecore_init();
|
||||
ecore_con_init();
|
||||
|
||||
args = ecore_getopt_parse(&options, values, argc, argv);
|
||||
if (args < 0)
|
||||
{
|
||||
fputs("ERROR: Could not parse command line options.\n", stderr);
|
||||
retval = EXIT_FAILURE;
|
||||
goto end;
|
||||
}
|
||||
|
||||
if (quit_option) goto end;
|
||||
|
||||
args = ecore_getopt_parse_positional(&options, values, argc, argv, args);
|
||||
if (args < 0)
|
||||
{
|
||||
fputs("ERROR: Could not parse positional arguments.\n", stderr);
|
||||
retval = EXIT_FAILURE;
|
||||
goto end;
|
||||
}
|
||||
|
||||
line_delimiter_str = _unescape(line_delimiter_str ? line_delimiter_str : "\\r\\n");
|
||||
|
||||
if (!commands)
|
||||
{
|
||||
fputs("ERROR: missing commands to send.\n", stderr);
|
||||
retval = EXIT_FAILURE;
|
||||
goto end;
|
||||
}
|
||||
|
||||
line_delimiter = (Eina_Slice)EINA_SLICE_STR(line_delimiter_str);
|
||||
|
||||
/*
|
||||
* Without a send_queue we'd have to manually implement an
|
||||
* Efl_Io_Reader object that would provide partial data when
|
||||
* Efl_Io_Reader.read() is called by Efl_Io_Copier. This is
|
||||
* cumbersome... we just want to write a full command and have the
|
||||
* queue to handle that for us.
|
||||
*
|
||||
* Our example's usage is to write each command at once followed by
|
||||
* the line_delimiter, then wait for a reply from the server, then
|
||||
* write another.
|
||||
*/
|
||||
send_queue = efl_add(EFL_IO_QUEUE_CLASS, NULL,
|
||||
efl_name_set(efl_self, "send_queue"),
|
||||
efl_io_queue_limit_set(efl_self, buffer_limit));
|
||||
if (!send_queue)
|
||||
{
|
||||
fprintf(stderr, "ERROR: could not create Efl_Io_Queue (send)\n");
|
||||
retval = EXIT_FAILURE;
|
||||
goto end;
|
||||
}
|
||||
|
||||
/*
|
||||
* Without a receive_queue we'd have to manually implement an
|
||||
* Efl_Io_Writer object that would handle write of partial data
|
||||
* with Efl_Io_Writer.write() is called by Efl_Io_Copier.
|
||||
*
|
||||
* For output we could have another solution as well: use NULL
|
||||
* destination and handle "line" or "data" events manually,
|
||||
* stealing the buffer so it doesn't grow.
|
||||
*
|
||||
* Our example's usage is to peek its data with slice_get() then
|
||||
* clear().
|
||||
*/
|
||||
receive_queue = efl_add(EFL_IO_QUEUE_CLASS, NULL,
|
||||
efl_name_set(efl_self, "receive_queue"),
|
||||
efl_io_queue_limit_set(efl_self, buffer_limit),
|
||||
efl_event_callback_add(efl_self, EFL_IO_QUEUE_EVENT_SLICE_CHANGED, _receiver_data, NULL));
|
||||
if (!receive_queue)
|
||||
{
|
||||
fprintf(stderr, "ERROR: could not create Efl_Io_Queue (receive)\n");
|
||||
retval = EXIT_FAILURE;
|
||||
goto error_receive_queue;
|
||||
}
|
||||
|
||||
/*
|
||||
* From here on it's mostly the same all Efl_Io_Copier would do,
|
||||
* check efl_io_copier_simple_example.c and efl_io_copier_example.c
|
||||
*/
|
||||
|
||||
/*
|
||||
* some objects such as the Efl.Io.Copier and Efl.Net.Dialer.Tcp
|
||||
* depend on main loop, thus their parent must be a loop
|
||||
* provider. We use the loop itself.
|
||||
*/
|
||||
loop = ecore_main_loop_get();
|
||||
|
||||
/* The TCP client to use to send/receive network data */
|
||||
dialer = efl_add(EFL_NET_DIALER_TCP_CLASS, loop,
|
||||
efl_name_set(efl_self, "dialer"),
|
||||
efl_event_callback_add(efl_self, EFL_NET_DIALER_EVENT_CONNECTED, _dialer_connected, NULL));
|
||||
if (!dialer)
|
||||
{
|
||||
fprintf(stderr, "ERROR: could not create Efl_Net_Dialer_Tcp\n");
|
||||
retval = EXIT_FAILURE;
|
||||
goto error_dialer;
|
||||
}
|
||||
|
||||
/* sender: send_queue->network */
|
||||
sender = efl_add(EFL_IO_COPIER_CLASS, loop,
|
||||
efl_name_set(efl_self, "sender"),
|
||||
efl_io_copier_line_delimiter_set(efl_self, &line_delimiter),
|
||||
efl_io_copier_source_set(efl_self, send_queue),
|
||||
efl_io_copier_destination_set(efl_self, dialer),
|
||||
efl_event_callback_array_add(efl_self, copier_cbs(), NULL));
|
||||
if (!sender)
|
||||
{
|
||||
fprintf(stderr, "ERROR: could not create Efl_Io_Copier (sender)\n");
|
||||
retval = EXIT_FAILURE;
|
||||
goto error_sender;
|
||||
}
|
||||
|
||||
/* receiver: network->receive_queue */
|
||||
receiver = efl_add(EFL_IO_COPIER_CLASS, loop,
|
||||
efl_name_set(efl_self, "receiver"),
|
||||
efl_io_copier_line_delimiter_set(efl_self, &line_delimiter),
|
||||
efl_io_copier_source_set(efl_self, dialer),
|
||||
efl_io_copier_destination_set(efl_self, receive_queue),
|
||||
efl_event_callback_array_add(efl_self, copier_cbs(), NULL));
|
||||
if (!receiver)
|
||||
{
|
||||
fprintf(stderr, "ERROR: could not create Efl_Io_Copier (receiver)\n");
|
||||
retval = EXIT_FAILURE;
|
||||
goto error_receiver;
|
||||
}
|
||||
|
||||
err = efl_net_dialer_dial(dialer, address);
|
||||
if (err)
|
||||
{
|
||||
fprintf(stderr, "ERROR: could not dial %s: %s\n",
|
||||
address, eina_error_msg_get(err));
|
||||
goto error_dialing;
|
||||
}
|
||||
|
||||
waiting = eina_list_append(waiting, sender);
|
||||
waiting = eina_list_append(waiting, receiver);
|
||||
|
||||
ecore_main_loop_begin();
|
||||
|
||||
if (waiting)
|
||||
{
|
||||
fprintf(stderr, "ERROR: %d operations were waiting!\n",
|
||||
eina_list_count(waiting));
|
||||
eina_list_free(waiting);
|
||||
waiting = NULL;
|
||||
}
|
||||
|
||||
error_dialing:
|
||||
efl_io_closer_close(receiver);
|
||||
efl_del(receiver);
|
||||
error_receiver:
|
||||
efl_io_closer_close(sender);
|
||||
efl_del(sender);
|
||||
error_sender:
|
||||
efl_del(dialer);
|
||||
error_dialer:
|
||||
efl_del(receive_queue);
|
||||
error_receive_queue:
|
||||
efl_del(send_queue);
|
||||
end:
|
||||
EINA_LIST_FREE(commands, cmd)
|
||||
{
|
||||
fprintf(stderr, "ERROR: unsent command: %s\n", cmd);
|
||||
free(cmd);
|
||||
}
|
||||
|
||||
ecore_con_shutdown();
|
||||
ecore_shutdown();
|
||||
|
||||
return retval;
|
||||
}
|
||||
|
|
@ -137,6 +137,7 @@ EAPI extern const Efl_Event_Description _EFL_GFX_PATH_CHANGED;
|
|||
#include "interfaces/efl_io_positioner.eo.h"
|
||||
|
||||
#include "interfaces/efl_io_buffer.eo.h"
|
||||
#include "interfaces/efl_io_queue.eo.h"
|
||||
|
||||
#else
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,430 @@
|
|||
#define EFL_IO_READER_PROTECTED 1
|
||||
#define EFL_IO_WRITER_PROTECTED 1
|
||||
|
||||
#include "config.h"
|
||||
#include "Efl.h"
|
||||
|
||||
#define MY_CLASS EFL_IO_QUEUE_CLASS
|
||||
|
||||
/*
|
||||
* This queue is simple and based on a single buffer that is
|
||||
* reallocated as needed up to some limit, keeping some pre-allocated
|
||||
* amount of bytes.
|
||||
*
|
||||
* Writes appends to the buffer. Reads consume and remove data from
|
||||
* buffer head.
|
||||
*
|
||||
* To avoid too much memmove(), reads won't immediately remove data,
|
||||
* instead will only increment position_read and allow some
|
||||
* slack. When the slack limit is reached or the buffer needs more
|
||||
* memory for write, then the memmove() happens.
|
||||
*
|
||||
* A more complex and possibly efficient version of this would be to
|
||||
* keep a list of internal buffers of fixed size. Writing would result
|
||||
* into segment and write into these chunks, creating new if
|
||||
* needed. Reading would consume from multiple chunks and if they're
|
||||
* all used, would be freed.
|
||||
*/
|
||||
|
||||
typedef struct _Efl_Io_Queue_Data
|
||||
{
|
||||
uint8_t *bytes;
|
||||
size_t allocated;
|
||||
size_t preallocated;
|
||||
size_t limit;
|
||||
size_t position_read; /* to avoid memmove(), allows some slack */
|
||||
size_t position_write;
|
||||
Eina_Bool pending_eos;
|
||||
Eina_Bool eos;
|
||||
Eina_Bool closed;
|
||||
Eina_Bool can_read;
|
||||
Eina_Bool can_write;
|
||||
} Efl_Io_Queue_Data;
|
||||
|
||||
static Eina_Bool
|
||||
_efl_io_queue_realloc(Eo *o, Efl_Io_Queue_Data *pd, size_t size)
|
||||
{
|
||||
void *tmp;
|
||||
size_t limit = efl_io_queue_limit_get(o);
|
||||
|
||||
if ((limit > 0) && (size > limit))
|
||||
size = limit;
|
||||
|
||||
if (pd->allocated == size) return EINA_FALSE;
|
||||
|
||||
if (size == 0)
|
||||
{
|
||||
free(pd->bytes);
|
||||
tmp = NULL;
|
||||
}
|
||||
else
|
||||
{
|
||||
tmp = realloc(pd->bytes, size);
|
||||
EINA_SAFETY_ON_NULL_RETURN_VAL(tmp, EINA_FALSE);
|
||||
}
|
||||
|
||||
pd->bytes = tmp;
|
||||
pd->allocated = size;
|
||||
return EINA_TRUE;
|
||||
}
|
||||
|
||||
static size_t
|
||||
_efl_io_queue_slack_get(const Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
const size_t used = pd->position_write - pd->position_read;
|
||||
|
||||
if (used >= 4096) return 4096;
|
||||
else if (used >= 1024) return 1024;
|
||||
else if (used >= 128) return 128;
|
||||
else return 32;
|
||||
}
|
||||
|
||||
static Eina_Bool
|
||||
_efl_io_queue_realloc_rounded(Eo *o, Efl_Io_Queue_Data *pd, size_t size)
|
||||
{
|
||||
if ((size > 0) && (size < 128))
|
||||
size = ((size / 32) + 1) * 32;
|
||||
else if (size < 1024)
|
||||
size = ((size / 128) + 1) * 128;
|
||||
else if (size < 8192)
|
||||
size = ((size / 1024) + 1) * 1024;
|
||||
else
|
||||
size = ((size / 4096) + 1) * 4096;
|
||||
|
||||
return _efl_io_queue_realloc(o, pd, size);
|
||||
}
|
||||
|
||||
/* reset position_read to zero, allowing all memory for write */
|
||||
static void
|
||||
_efl_io_queue_adjust(Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
size_t used = pd->position_write - pd->position_read;
|
||||
memmove(pd->bytes, pd->bytes + pd->position_read, used);
|
||||
pd->position_write = used;
|
||||
pd->position_read = 0;
|
||||
}
|
||||
|
||||
static void
|
||||
_efl_io_queue_adjust_and_realloc_if_needed(Eo *o, Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
const size_t slack = _efl_io_queue_slack_get(pd);
|
||||
size_t spare;
|
||||
|
||||
if (pd->limit > 0)
|
||||
{
|
||||
if (pd->position_write + slack >= pd->limit)
|
||||
_efl_io_queue_adjust(pd);
|
||||
}
|
||||
else if (pd->position_read > slack)
|
||||
_efl_io_queue_adjust(pd);
|
||||
|
||||
spare = pd->allocated - pd->position_write;
|
||||
if (spare > slack)
|
||||
{
|
||||
size_t new_size = pd->position_write + slack;
|
||||
|
||||
/*
|
||||
* this may result in going over slack again, no
|
||||
* problems with that.
|
||||
*/
|
||||
if (new_size < pd->preallocated)
|
||||
new_size = pd->preallocated;
|
||||
|
||||
/* use rounded so we avoid too many reallocs */
|
||||
_efl_io_queue_realloc_rounded(o, pd, new_size);
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
_efl_io_queue_update_cans(Eo *o, Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
size_t used = pd->position_write - pd->position_read;
|
||||
size_t limit;
|
||||
|
||||
efl_io_reader_can_read_set(o, used > 0);
|
||||
|
||||
limit = efl_io_queue_limit_get(o);
|
||||
if (pd->pending_eos)
|
||||
efl_io_writer_can_write_set(o, EINA_FALSE);
|
||||
else
|
||||
efl_io_writer_can_write_set(o, (limit == 0) || (used < limit));
|
||||
}
|
||||
|
||||
EOLIAN static void
|
||||
_efl_io_queue_preallocate(Eo *o, Efl_Io_Queue_Data *pd, size_t size)
|
||||
{
|
||||
EINA_SAFETY_ON_TRUE_RETURN(efl_io_closer_closed_get(o));
|
||||
if (pd->allocated < size)
|
||||
_efl_io_queue_realloc_rounded(o, pd, size);
|
||||
pd->preallocated = size;
|
||||
}
|
||||
|
||||
EOLIAN static void
|
||||
_efl_io_queue_limit_set(Eo *o, Efl_Io_Queue_Data *pd, size_t limit)
|
||||
{
|
||||
EINA_SAFETY_ON_TRUE_RETURN(efl_io_closer_closed_get(o));
|
||||
|
||||
if (pd->limit == limit) return;
|
||||
pd->limit = limit;
|
||||
if (pd->limit == 0) goto end;
|
||||
|
||||
_efl_io_queue_adjust(pd);
|
||||
|
||||
if (pd->allocated > limit)
|
||||
_efl_io_queue_realloc(o, pd, limit);
|
||||
|
||||
if (pd->position_write > limit)
|
||||
{
|
||||
pd->position_write = limit;
|
||||
if (pd->position_read > limit) pd->position_read = limit;
|
||||
}
|
||||
|
||||
_efl_io_queue_adjust_and_realloc_if_needed(o, pd);
|
||||
efl_event_callback_call(o, EFL_IO_QUEUE_EVENT_SLICE_CHANGED, NULL);
|
||||
|
||||
end:
|
||||
_efl_io_queue_update_cans(o, pd);
|
||||
}
|
||||
|
||||
EOLIAN static size_t
|
||||
_efl_io_queue_limit_get(Eo *o EINA_UNUSED, Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
return pd->limit;
|
||||
}
|
||||
|
||||
EOLIAN static size_t
|
||||
_efl_io_queue_usage_get(Eo *o EINA_UNUSED, Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
return pd->position_write - pd->position_read;
|
||||
}
|
||||
|
||||
EOLIAN static Eina_Bool
|
||||
_efl_io_queue_slice_get(Eo *o, Efl_Io_Queue_Data *pd, Eina_Slice *slice)
|
||||
{
|
||||
if (slice)
|
||||
{
|
||||
slice->mem = pd->bytes + pd->position_read;
|
||||
slice->len = efl_io_queue_usage_get(o);
|
||||
}
|
||||
EINA_SAFETY_ON_TRUE_RETURN_VAL(efl_io_closer_closed_get(o), EINA_FALSE);
|
||||
return EINA_TRUE;
|
||||
}
|
||||
|
||||
EOLIAN static void
|
||||
_efl_io_queue_clear(Eo *o, Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
pd->position_read = 0;
|
||||
pd->position_write = 0;
|
||||
efl_io_reader_can_read_set(o, EINA_FALSE);
|
||||
efl_event_callback_call(o, EFL_IO_QUEUE_EVENT_SLICE_CHANGED, NULL);
|
||||
if (pd->pending_eos)
|
||||
efl_io_reader_eos_set(o, EINA_TRUE);
|
||||
}
|
||||
|
||||
EOLIAN static void
|
||||
_efl_io_queue_eos_mark(Eo *o, Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
if (pd->eos) return;
|
||||
|
||||
if (efl_io_queue_usage_get(o) > 0)
|
||||
pd->pending_eos = EINA_TRUE;
|
||||
else
|
||||
efl_io_reader_eos_set(o, EINA_TRUE);
|
||||
}
|
||||
|
||||
EOLIAN static Efl_Object *
|
||||
_efl_io_queue_efl_object_finalize(Eo *o, Efl_Io_Queue_Data *pd EINA_UNUSED)
|
||||
{
|
||||
o = efl_finalize(efl_super(o, MY_CLASS));
|
||||
if (!o) return NULL;
|
||||
|
||||
_efl_io_queue_update_cans(o, pd);
|
||||
|
||||
return o;
|
||||
}
|
||||
|
||||
EOLIAN static void
|
||||
_efl_io_queue_efl_object_destructor(Eo *o, Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
if (!efl_io_closer_closed_get(o))
|
||||
efl_io_closer_close(o);
|
||||
|
||||
efl_destructor(efl_super(o, MY_CLASS));
|
||||
|
||||
if (pd->bytes)
|
||||
{
|
||||
free(pd->bytes);
|
||||
pd->bytes = NULL;
|
||||
pd->allocated = 0;
|
||||
pd->position_read = 0;
|
||||
pd->position_write = 0;
|
||||
}
|
||||
}
|
||||
|
||||
EOLIAN static Eina_Error
|
||||
_efl_io_queue_efl_io_reader_read(Eo *o, Efl_Io_Queue_Data *pd, Eina_Rw_Slice *rw_slice)
|
||||
{
|
||||
Eina_Slice ro_slice;
|
||||
size_t available;
|
||||
|
||||
EINA_SAFETY_ON_NULL_RETURN_VAL(rw_slice, EINVAL);
|
||||
EINA_SAFETY_ON_TRUE_GOTO(efl_io_closer_closed_get(o), error);
|
||||
|
||||
available = pd->position_write - pd->position_read;
|
||||
if (rw_slice->len > available)
|
||||
{
|
||||
rw_slice->len = available;
|
||||
if (rw_slice->len == 0)
|
||||
return EAGAIN;
|
||||
}
|
||||
|
||||
ro_slice.len = rw_slice->len;
|
||||
ro_slice.mem = pd->bytes + pd->position_read;
|
||||
|
||||
*rw_slice = eina_rw_slice_copy(*rw_slice, ro_slice);
|
||||
pd->position_read += ro_slice.len;
|
||||
|
||||
efl_io_reader_can_read_set(o, pd->position_read < pd->position_write);
|
||||
efl_event_callback_call(o, EFL_IO_QUEUE_EVENT_SLICE_CHANGED, NULL);
|
||||
|
||||
if ((pd->pending_eos) && (efl_io_queue_usage_get(o) == 0))
|
||||
efl_io_reader_eos_set(o, EINA_TRUE);
|
||||
|
||||
return 0;
|
||||
|
||||
error:
|
||||
rw_slice->len = 0;
|
||||
return EINVAL;
|
||||
}
|
||||
|
||||
EOLIAN static Eina_Bool
|
||||
_efl_io_queue_efl_io_reader_can_read_get(Eo *o EINA_UNUSED, Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
return pd->can_read;
|
||||
}
|
||||
|
||||
EOLIAN static void
|
||||
_efl_io_queue_efl_io_reader_can_read_set(Eo *o, Efl_Io_Queue_Data *pd, Eina_Bool can_read)
|
||||
{
|
||||
EINA_SAFETY_ON_TRUE_RETURN(efl_io_closer_closed_get(o));
|
||||
if (pd->can_read == can_read) return;
|
||||
pd->can_read = can_read;
|
||||
efl_event_callback_call(o, EFL_IO_READER_EVENT_CAN_READ_CHANGED, NULL);
|
||||
}
|
||||
|
||||
EOLIAN static Eina_Bool
|
||||
_efl_io_queue_efl_io_reader_eos_get(Eo *o EINA_UNUSED, Efl_Io_Queue_Data *pd EINA_UNUSED)
|
||||
{
|
||||
return pd->eos;
|
||||
}
|
||||
|
||||
EOLIAN static void
|
||||
_efl_io_queue_efl_io_reader_eos_set(Eo *o, Efl_Io_Queue_Data *pd EINA_UNUSED, Eina_Bool is_eos)
|
||||
{
|
||||
EINA_SAFETY_ON_TRUE_RETURN(efl_io_closer_closed_get(o));
|
||||
if (pd->eos == is_eos) return;
|
||||
pd->eos = is_eos;
|
||||
if (is_eos)
|
||||
{
|
||||
pd->pending_eos = EINA_FALSE;
|
||||
efl_event_callback_call(o, EFL_IO_READER_EVENT_EOS, NULL);
|
||||
}
|
||||
}
|
||||
|
||||
EOLIAN static Eina_Error
|
||||
_efl_io_queue_efl_io_writer_write(Eo *o, Efl_Io_Queue_Data *pd, Eina_Slice *slice, Eina_Slice *remaining)
|
||||
{
|
||||
size_t available_write, available_total, todo, limit;
|
||||
int err = EINVAL;
|
||||
|
||||
EINA_SAFETY_ON_NULL_RETURN_VAL(slice, EINVAL);
|
||||
EINA_SAFETY_ON_TRUE_GOTO(efl_io_closer_closed_get(o), error);
|
||||
|
||||
err = EBADF;
|
||||
EINA_SAFETY_ON_TRUE_GOTO(pd->pending_eos, error);
|
||||
|
||||
available_write = pd->allocated - pd->position_write;
|
||||
available_total = available_write + pd->position_read;
|
||||
limit = efl_io_queue_limit_get(o);
|
||||
|
||||
err = ENOSPC;
|
||||
if (available_write >= slice->len)
|
||||
{
|
||||
todo = slice->len;
|
||||
}
|
||||
else if (available_total >= slice->len)
|
||||
{
|
||||
_efl_io_queue_adjust(pd);
|
||||
todo = slice->len;
|
||||
}
|
||||
else if ((limit > 0) && (pd->allocated == limit)) goto error;
|
||||
else
|
||||
{
|
||||
_efl_io_queue_adjust(pd);
|
||||
_efl_io_queue_realloc_rounded(o, pd, pd->position_write + slice->len);
|
||||
if (pd->allocated >= pd->position_write + slice->len)
|
||||
todo = slice->len;
|
||||
else
|
||||
todo = pd->allocated - pd->position_write;
|
||||
|
||||
if (todo == 0) goto error;
|
||||
}
|
||||
|
||||
memcpy(pd->bytes + pd->position_write, slice->mem, todo);
|
||||
if (remaining)
|
||||
{
|
||||
remaining->len = slice->len - todo;
|
||||
if (remaining->len)
|
||||
remaining->mem = slice->bytes + todo;
|
||||
else
|
||||
remaining->mem = NULL;
|
||||
}
|
||||
slice->len = todo;
|
||||
|
||||
pd->position_write += todo;
|
||||
|
||||
_efl_io_queue_adjust_and_realloc_if_needed(o, pd);
|
||||
efl_event_callback_call(o, EFL_IO_QUEUE_EVENT_SLICE_CHANGED, NULL);
|
||||
_efl_io_queue_update_cans(o, pd);
|
||||
|
||||
return 0;
|
||||
|
||||
error:
|
||||
if (remaining) *remaining = *slice;
|
||||
slice->len = 0;
|
||||
return err;
|
||||
}
|
||||
|
||||
EOLIAN static Eina_Bool
|
||||
_efl_io_queue_efl_io_writer_can_write_get(Eo *o EINA_UNUSED, Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
return pd->can_write;
|
||||
}
|
||||
|
||||
EOLIAN static void
|
||||
_efl_io_queue_efl_io_writer_can_write_set(Eo *o, Efl_Io_Queue_Data *pd, Eina_Bool can_write)
|
||||
{
|
||||
EINA_SAFETY_ON_TRUE_RETURN(efl_io_closer_closed_get(o));
|
||||
if (pd->can_write == can_write) return;
|
||||
pd->can_write = can_write;
|
||||
efl_event_callback_call(o, EFL_IO_WRITER_EVENT_CAN_WRITE_CHANGED, NULL);
|
||||
}
|
||||
|
||||
EOLIAN static Eina_Error
|
||||
_efl_io_queue_efl_io_closer_close(Eo *o, Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
EINA_SAFETY_ON_TRUE_RETURN_VAL(efl_io_closer_closed_get(o), EINVAL);
|
||||
efl_io_queue_eos_mark(o);
|
||||
efl_io_queue_clear(o);
|
||||
pd->closed = EINA_TRUE;
|
||||
efl_event_callback_call(o, EFL_IO_CLOSER_EVENT_CLOSED, NULL);
|
||||
return 0;
|
||||
}
|
||||
|
||||
EOLIAN static Eina_Bool
|
||||
_efl_io_queue_efl_io_closer_closed_get(Eo *o EINA_UNUSED, Efl_Io_Queue_Data *pd)
|
||||
{
|
||||
return pd->closed;
|
||||
}
|
||||
|
||||
#include "interfaces/efl_io_queue.eo.c"
|
||||
|
|
@ -0,0 +1,94 @@
|
|||
class Efl.Io.Queue (Efl.Object, Efl.Io.Reader, Efl.Io.Writer, Efl.Io.Closer) {
|
||||
[[Generic In-memory queue of data to be used as I/O.
|
||||
|
||||
This class is to be used to receive temporary data using
|
||||
@Efl.Io.Writer.write and hold it until someone calls
|
||||
@Efl.Io.Reader.read to consume it.
|
||||
|
||||
A fixed sized queue can be implemented by setting @.limit
|
||||
followed by @.preallocate
|
||||
|
||||
@since 1.19
|
||||
]]
|
||||
|
||||
methods {
|
||||
preallocate {
|
||||
[[Immediately pre-allocate a buffer of at least a given size.]]
|
||||
params {
|
||||
@in size: size; [[amount of bytes to pre-allocate.]]
|
||||
}
|
||||
}
|
||||
|
||||
@property limit {
|
||||
[[Limit how big the buffer can grow.
|
||||
|
||||
This affects both @.preallocate and how buffer grows
|
||||
when @Efl.Io.Writer.write is called.
|
||||
|
||||
If you want a buffer of an exact size, always set the
|
||||
limit before any further calls that can grow it.
|
||||
]]
|
||||
get { }
|
||||
set {
|
||||
[[Constructor-only property to set buffer limit. 0 is unlimited]]
|
||||
}
|
||||
values {
|
||||
size: size; [[Defines a maximum buffer size, or 0 to allow unlimited amount of bytes]]
|
||||
}
|
||||
}
|
||||
|
||||
@property usage {
|
||||
[[How many bytes are available for read]]
|
||||
get { }
|
||||
values {
|
||||
usage: size;
|
||||
}
|
||||
}
|
||||
|
||||
slice_get { // TODO: property and return of Eina.Slice (not pointer)
|
||||
[[Get a temporary access to queue's internal read memory.
|
||||
|
||||
The memory pointed by slice may be changed by other
|
||||
methods of this class. The event "slice,changed" will be
|
||||
called in those situations.
|
||||
]]
|
||||
params {
|
||||
@out slice: Eina.Slice; [[slice of the current buffer, may be invalidated if @Efl.Io.Writer.write, @Efl.Io.Closer.close or @Efl.Io.Reader.read are called. It is the full slice available for reading.]]
|
||||
}
|
||||
return: bool (false);
|
||||
}
|
||||
|
||||
clear {
|
||||
[[Clear the queue. Same as reading all data]]
|
||||
}
|
||||
|
||||
eos_mark {
|
||||
[[Mark this end-of-stream.
|
||||
|
||||
That will set @Efl.Io.Reader.eos to $true and forbid any
|
||||
further writes.
|
||||
|
||||
Unlike @Efl.Io.Closer.close, this won't clear anything.
|
||||
]]
|
||||
}
|
||||
}
|
||||
|
||||
events {
|
||||
slice,changed; [[The read-slice returned by @.slice_get may have changed.]]
|
||||
}
|
||||
|
||||
implements {
|
||||
Efl.Object.finalize;
|
||||
Efl.Object.destructor;
|
||||
Efl.Io.Reader.read;
|
||||
Efl.Io.Reader.can_read.get;
|
||||
Efl.Io.Reader.can_read.set;
|
||||
Efl.Io.Reader.eos.get;
|
||||
Efl.Io.Reader.eos.set;
|
||||
Efl.Io.Writer.write;
|
||||
Efl.Io.Writer.can_write.get;
|
||||
Efl.Io.Writer.can_write.set;
|
||||
Efl.Io.Closer.close;
|
||||
Efl.Io.Closer.closed.get;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue