rework the test case to have two separate transactions,

an input transaction that reads from stdin and an output
transaction that writes to stdout. Next up is making them
cooridinate with each other, so the write transaction
writes after the read transaction fills up its buffer or
blocks waiting for more data.

1 files changed, 98 insertions, 14 deletions

diff --git a/src/pop-transaction.c b/src/pop-transaction.c
index f135c9d..1657d1c 100644
--- a/src/pop-transaction.c
+++ b/src/pop-transaction.c
@@ -1507,34 +1507,118 @@ process_write_action (PopTransaction *transaction,
 }
 
 static PopActionProcessStatus
-process_cat_action (PopTransaction  *transaction,
-                    gpointer         data)
+process_set_fd_nonblocking_action (PopTransaction *transaction,
+                                   gpointer        data)
 {
-  char *buffer;
-  static const size_t buffer_size = 10;
-  InputOutputArguments *input_arguments, *output_arguments;
+  int fd;
+  int flags, status;
+
+  fd = GPOINTER_TO_INT (data);
+
+  flags = fcntl (fd, F_GETFL);
+
+  if (flags < 0)
+    {
+      pop_transaction_set_error_from_errno (transaction);
+      return POP_ACTION_PROCESS_STATUS_FAILED;
+    }
+
+  status = fcntl (fd, F_SETFL, flags | O_NONBLOCK);
+
+  if (status < 0)
+    {
+      pop_transaction_set_error_from_errno (transaction);
+      return POP_ACTION_PROCESS_STATUS_FAILED;
+    }
 
-  fcntl (STDIN_FILENO, F_SETFL, O_NONBLOCK);
-  fcntl (STDOUT_FILENO, F_SETFL, O_NONBLOCK);
+  g_object_set_data (G_OBJECT (transaction), "file-flags",
+                     GINT_TO_POINTER (flags));
+
+  return POP_ACTION_PROCESS_STATUS_SUCCEEDED;
+}
+
+static PopActionProcessStatus
+rollback_set_fd_nonblocking_action (PopTransaction *transaction,
+                                    gpointer        data)
+{
+  int fd, flags;
+
+  fd = GPOINTER_TO_INT (data);
+
+  flags = g_object_get_data (G_OBJECT (transaction), "file-flags");
+
+  fcntl (fd, F_SETFL, flags);
+
+  return POP_ACTION_ROLLBACK_STATUS_FINISHED;
+}
+
+static PopTransaction *
+input_transaction_new (guint8 *buffer,
+                       gsize   buffer_size)
+{
+  PopTransaction *transaction;
+
+  InputOutputArguments *input_arguments;
+
+  transaction = pop_transaction_new ();
+
+  pop_transaction_add_action (transaction, 
+                              process_set_fd_nonblocking_action 
+                              rollback_set_fd_nonblocking_action,
+                              GINT_TO_POINTER (STDIN_FILENO));
 
   input_arguments = g_slice_new (InputOutputArguments);
   input_arguments->fd = STDIN_FILENO; 
-  input_arguments->buffer = g_malloc0 (buffer_size);
+  input_arguments->buffer = buffer;
   input_arguments->number_of_bytes_to_process = buffer_size;
 
   pop_transaction_add_action (transaction, process_read_action, 
                               NULL, input_arguments);
-  g_object_set_data (G_OBJECT (transaction), "cat-input-args", input_arguments);
+
+  return transaction;
+}
+
+static PopTransaction *
+output_transaction_new (guint8 *buffer,
+                        gsize   buffer_size)
+{
+  PopTransaction *transaction;
+
+  InputOutputArguments *output_arguments;
+
+  transaction = pop_transaction_new ();
+
+  pop_transaction_add_action (transaction, 
+                              process_set_fd_nonblocking_action 
+                              rollback_set_fd_nonblocking_action,
+                              GINT_TO_POINTER (STDOUT_FILENO));
 
   output_arguments = g_slice_new (InputOutputArguments);
   output_arguments->fd = STDOUT_FILENO; 
-  output_arguments->buffer = input_arguments->buffer;
+  output_arguments->buffer = buffer;
   output_arguments->number_of_bytes_to_process = buffer_size;
 
-  pop_transaction_add_action (transaction, process_write_action, NULL,
-                              output_arguments);
-  g_object_set_data (G_OBJECT (transaction), "cat-output-args", output_arguments);
- 
+  pop_transaction_add_action (transaction, process_write_action, 
+                              NULL, output_arguments);
+
+  return transaction;
+}
+
+static PopActionProcessStatus
+process_cat_action (PopTransaction *transaction,
+                    gpointer        data)
+{
+  PopTransaction *input_transaction, *output_transaction;
+  char *buffer;
+  static const int buffer_size = 4096;
+
+  input_transaction = input_transaction_new (buffer, buffer_size);
+  pop_transaction_commit (input_transaction);
+
+  output_transaction = output_transaction_new (buffer, buffer_size);
+  pop_transaction_wait (output_transaction);
+  pop_transaction_commit (output_transaction);
+
   return POP_ACTION_PROCESS_STATUS_SUCCEEDED;
 }