efl_io_copier_flush: add may_block and ignore_line_delimiter parameters.

The may_block parameter is useful to force a flush without blocking on
read/write, sometimes particularly useful if ignore_line_delimiter is
true, then you get the data events without blocking -- as if a server
sending some content misses a trailing line delimiter, you do not want
to block on recv() but still want to flush data to user.

The ignore_line_delimiter parameter is useful if we're going to close
the copier and want to flush pending data which may exist due missing
trailing terminator. The close method will also force that if
destination can take more data.

4 files changed, 61 insertions, 7 deletions

diff --git a/src/bin/efl/efl_debug.c b/src/bin/efl/efl_debug.c
index 7b4dbb0c2e..4cb4b03c58 100644
--- a/src/bin/efl/efl_debug.c
+++ b/src/bin/efl/efl_debug.c
@@ -364,7 +364,7 @@ main(int argc, char **argv)
 
    while ((!efl_io_closer_closed_get(dialer)) &&
           efl_io_queue_usage_get(input))
-     efl_io_copier_flush(send_copier);
+     efl_io_copier_flush(send_copier, EINA_TRUE, EINA_TRUE);
 
  end:
    eina_list_free(waiting);
diff --git a/src/lib/ecore/efl_io_copier.c b/src/lib/ecore/efl_io_copier.c
index 3eee61ff59..c92a8efca0 100644
--- a/src/lib/ecore/efl_io_copier.c
+++ b/src/lib/ecore/efl_io_copier.c
@@ -27,6 +27,7 @@ typedef struct _Efl_Io_Copier_Data
    double inactivity_timeout;
    Eina_Bool closed;
    Eina_Bool done;
+   Eina_Bool force_dispatch;
    Eina_Bool close_on_exec;
    Eina_Bool close_on_destructor;
 } Efl_Io_Copier_Data;
@@ -281,7 +282,7 @@ _efl_io_copier_write(Eo *o, Efl_Io_Copier_Data *pd)
         return;
      }
 
-   if ((pd->line_delimiter.len > 0) &&
+   if ((pd->line_delimiter.len > 0) && (!pd->force_dispatch) &&
        (pd->source && !efl_io_reader_eos_get(pd->source)))
      {
         const uint8_t *p = eina_slice_find(ro_slice, pd->line_delimiter);
@@ -469,6 +470,8 @@ _efl_io_copier_destination_closed(void *data, const Efl_Event *event EINA_UNUSED
      {
         Eina_Error err = EBADF;
         if (pd->inactivity_timer) efl_future_cancel(pd->inactivity_timer);
+        WRN("copier %p destination %p closed with %zd bytes pending...",
+            o, pd->destination, eina_binbuf_length_get(pd->buf));
         efl_event_callback_call(o, EFL_IO_COPIER_EVENT_ERROR, &err);
      }
 }
@@ -604,6 +607,34 @@ _efl_io_copier_efl_io_closer_close(Eo *o, Efl_Io_Copier_Data *pd)
 
    _COPIER_DBG(o, pd);
 
+   while (pd->buf)
+     {
+        size_t pending = eina_binbuf_length_get(pd->buf);
+        if (pending == 0) break;
+        else if (pd->destination && efl_io_writer_can_write_get(pd->destination))
+          {
+             DBG("copier %p destination %p closed with %zd bytes pending, do final write...",
+                 o, pd->destination, pending);
+             pd->force_dispatch = EINA_TRUE;
+             _efl_io_copier_write(o, pd);
+             pd->force_dispatch = EINA_FALSE;
+          }
+        else if (!pd->destination)
+          {
+             Eina_Slice binbuf_slice = eina_binbuf_slice_get(pd->buf);
+             DBG("copier %p destination %p closed with %zd bytes pending, dispatch events...",
+                 o, pd->destination, pending);
+             _efl_io_copier_dispatch_data_events(o, pd, binbuf_slice);
+             break;
+          }
+        else
+          {
+             DBG("copier %p destination %p closed with %zd bytes pending...",
+                 o, pd->destination, pending);
+             break;
+          }
+     }
+
    if (pd->job)
      efl_future_cancel(pd->job);
 
@@ -699,7 +730,7 @@ _efl_io_copier_inactivity_timeout_get(Eo *o EINA_UNUSED, Efl_Io_Copier_Data *pd)
 }
 
 EOLIAN static Eina_Bool
-_efl_io_copier_flush(Eo *o, Efl_Io_Copier_Data *pd)
+_efl_io_copier_flush(Eo *o, Efl_Io_Copier_Data *pd, Eina_Bool may_block, Eina_Bool ignore_line_delimiter)
 {
    uint64_t old_read = pd->progress.read;
    uint64_t old_written = pd->progress.written;
@@ -708,10 +739,29 @@ _efl_io_copier_flush(Eo *o, Efl_Io_Copier_Data *pd)
    _COPIER_DBG(o, pd);
 
    if (pd->source && !efl_io_reader_eos_get(pd->source))
-     _efl_io_copier_read(o, pd);
+     {
+        if (may_block || efl_io_reader_can_read_get(pd->source))
+          _efl_io_copier_read(o, pd);
+     }
 
    if (pd->destination)
-     _efl_io_copier_write(o, pd);
+     {
+        if (may_block || efl_io_writer_can_write_get(pd->source))
+          {
+             pd->force_dispatch = ignore_line_delimiter;
+             _efl_io_copier_write(o, pd);
+             pd->force_dispatch = EINA_FALSE;
+          }
+     }
+   else if (ignore_line_delimiter && pd->buf)
+     {
+        size_t pending = eina_binbuf_length_get(pd->buf);
+        if (pending)
+          {
+             Eina_Slice binbuf_slice = eina_binbuf_slice_get(pd->buf);
+             _efl_io_copier_dispatch_data_events(o, pd, binbuf_slice);
+          }
+     }
 
    if ((old_read != pd->progress.read) ||
        (old_written != pd->progress.written) ||
diff --git a/src/lib/ecore/efl_io_copier.eo b/src/lib/ecore/efl_io_copier.eo
index 1c0bd5fd93..36ad624843 100644
--- a/src/lib/ecore/efl_io_copier.eo
+++ b/src/lib/ecore/efl_io_copier.eo
@@ -335,6 +335,10 @@ class Efl.Io.Copier (Efl.Loop_User, Efl.Io.Closer) {
               approach is to operate asynchronously and wait for
               "done" event.
             ]]
+            params {
+                may_block: bool; [[If $true, then @Efl.Io.Reader.can_read and @Efl.Io.Writer.can_write are not checked and the call may block.]]
+                ignore_line_delimiter: bool; [[Force flush ignoring line delimiters]]
+            }
             return: bool(true); [[$true on success, $false otherwise]]
         }
     }
diff --git a/src/lib/ecore_ipc/ecore_ipc.c b/src/lib/ecore_ipc/ecore_ipc.c
index e227251467..e4287aa4fa 100644
--- a/src/lib/ecore_ipc/ecore_ipc.c
+++ b/src/lib/ecore_ipc/ecore_ipc.c
@@ -1210,7 +1210,7 @@ ecore_ipc_server_flush(Ecore_Ipc_Server *svr)
    if (svr->dialer.input)
      {
         while (efl_io_queue_usage_get(svr->dialer.input) > 0)
-          efl_io_copier_flush(svr->dialer.send_copier);
+          efl_io_copier_flush(svr->dialer.send_copier, EINA_TRUE, EINA_TRUE);
         return;
      }
    else if (svr->server)
@@ -1627,7 +1627,7 @@ ecore_ipc_client_flush(Ecore_Ipc_Client *cl)
    if (cl->socket.input)
      {
         while (efl_io_queue_usage_get(cl->socket.input) > 0)
-          efl_io_copier_flush(cl->socket.send_copier);
+          efl_io_copier_flush(cl->socket.send_copier, EINA_TRUE, EINA_TRUE);
         return;
      }
 #ifndef EFL_NET_SERVER_UNIX_CLASS