cleanups: doc updates

pinos/spi/defs.h

@@ -37,7 +37,6 @@ typedef enum {
   SPI_RESULT_INVALID_PORT             = -5,
   SPI_RESULT_HAVE_ENOUGH_INPUT        = -6,
   SPI_RESULT_NEED_MORE_INPUT          = -7,
-  SPI_RESULT_HAVE_EVENT               = -8,
   SPI_RESULT_PORTS_CHANGED            = -9,
   SPI_RESULT_FORMAT_CHANGED           = -10,
   SPI_RESULT_PROPERTIES_CHANGED       = -11,

pinos/spi/event.h

@@ -28,13 +28,30 @@ typedef struct _SpiEvent SpiEvent;
 
 #include <spi/defs.h>
 
+/**
+ * SpiEventType:
+ * @SPI_EVENT_TYPE_INVALID: invalid event, should be ignored
+ * @SPI_EVENT_TYPE_ACTIVATED: emited when the ACTIVATE command completes
+ * @SPI_EVENT_TYPE_DEACTIVATED: emited when the DEACTIVATE command completes
+ * @SPI_EVENT_TYPE_HAVE_OUTPUT: emited when an async node has output
+ * @SPI_EVENT_TYPE_NEED_INPUT: emited when an async node needs input. The data
+ *              member could contain a SpiBuffer that should be filled or it
+ *              can be %NULL.
+ * @SPI_EVENT_TYPE_ADD_POLL: emited when a pollfd should be added
+ * @SPI_EVENT_TYPE_REMOVE_POLL: emited when a pollfd should be removed
+ * @SPI_EVENT_TYPE_DRAINED: emited when DRAIN command completed
+ * @SPI_EVENT_TYPE_MARKER: emited when MARK command completed
+ * @SPI_EVENT_TYPE_ERROR: emited when error occured
+ * @SPI_EVENT_TYPE_BUFFERING: emited when buffering is in progress
+ */
 typedef enum {
   SPI_EVENT_TYPE_INVALID                  = 0,
   SPI_EVENT_TYPE_ACTIVATED,
   SPI_EVENT_TYPE_DEACTIVATED,
   SPI_EVENT_TYPE_HAVE_OUTPUT,
   SPI_EVENT_TYPE_NEED_INPUT,
-  SPI_EVENT_TYPE_REQUEST_DATA,
+  SPI_EVENT_TYPE_ADD_POLL,
+  SPI_EVENT_TYPE_REMOVE_POLL,
   SPI_EVENT_TYPE_DRAINED,
   SPI_EVENT_TYPE_MARKER,
   SPI_EVENT_TYPE_ERROR,

pinos/spi/node.h

@@ -34,37 +34,68 @@ typedef struct _SpiNode SpiNode;
 #include <spi/command.h>
 
 /**
- * SpiDataFlags:
+ * SpiInputFlags:
- * @SPI_DATA_FLAG_NONE: no flag
+ * @SPI_INPUT_FLAG_NONE: no flag
- * @SPI_DATA_FLAG_DISCARD: the buffer can be discarded
- * @SPI_DATA_FLAG_FORMAT_CHANGED: the format of this port changed
- * @SPI_DATA_FLAG_PROPERTIES_CHANGED: properties of this port changed
- * @SPI_DATA_FLAG_REMOVED: this port is removed
- * @SPI_DATA_FLAG_NO_BUFFER: no buffer was produced
  */
 typedef enum {
-  SPI_DATA_FLAG_NONE                  =  0,
+  SPI_INPUT_FLAG_NONE                  =  0,
-  SPI_DATA_FLAG_DISCARD               = (1 << 0),
+} SpiInputFlags;
-  SPI_DATA_FLAG_FORMAT_CHANGED        = (1 << 1),
-  SPI_DATA_FLAG_PROPERTIES_CHANGED    = (1 << 2),
-  SPI_DATA_FLAG_REMOVED               = (1 << 3),
-  SPI_DATA_FLAG_NO_BUFFER             = (1 << 4),
-} SpiDataFlags;
 
 /**
- * SpiDataInfo:
+ * SpiOutputFlags:
+ * @SPI_OUTPUT_FLAG_NONE: no flag
+ * @SPI_OUTPUT_FLAG_PULL: force a #SPI_EVENT_NEED_INPUT event on the
+ *                        peer input ports when no data is available.
+ * @SPI_OUTPUT_FLAG_DISCARD: discard the buffer data
+ */
+typedef enum {
+  SPI_OUTPUT_FLAG_NONE                  =  0,
+  SPI_OUTPUT_FLAG_PULL                  = (1 << 0),
+  SPI_OUTPUT_FLAG_DISCARD               = (1 << 1),
+} SpiOutputFlags;
+
+/**
+ * SpiInputInfo:
+ * @port_id: the port id
+ * @flags: extra flags
+ * @buffer: a buffer
+ *
+ * Input information for a node.
+ */
+typedef struct {
+  uint32_t        port_id;
+  SpiInputFlags   flags;
+  SpiBuffer      *buffer;
+  SpiEvent       *event;
+  SpiResult       status;
+} SpiInputInfo;
+
+/**
+ * SpiOutputInfo:
  * @port_id: the port id
  * @flags: extra flags
  * @buffer: a buffer
  * @event: an event
+ *
+ * Output information for a node.
  */
 typedef struct {
   uint32_t        port_id;
-  SpiDataFlags  flags;
+  SpiOutputFlags  flags;
   SpiBuffer      *buffer;
   SpiEvent       *event;
-} SpiDataInfo;
+  SpiResult       status;
+} SpiOutputInfo;
 
+/**
+ * SpiEventCallback:
+ * @node: a #SpiNode emiting the event
+ * @event: the event that was emited
+ * @user_data: user data provided when registering the callback
+ *
+ * This will be called when an out-of-bound event is notified
+ * on @node.
+ */
 typedef void   (*SpiEventCallback)   (SpiNode     *node,
                                       SpiEvent    *event,
                                       void        *user_data);
@@ -99,7 +130,7 @@ struct _SpiNode {
 
   /**
    * SpiNode::get_params:
-   * @Node: a #SpiNode
+   * @node: a #SpiNode
    * @props: a location for a #SpiParams pointer
    *
    * Get the configurable parameters of @node.
@@ -117,7 +148,7 @@ struct _SpiNode {
                                        SpiParams       **props);
   /**
    * SpiNode::set_params:
-   * @Node: a #SpiNode
+   * @node: a #SpiNode
    * @props: a #SpiParams
    *
    * Set the configurable parameters in @node.
@@ -141,7 +172,7 @@ struct _SpiNode {
                                        const SpiParams  *props);
   /**
    * SpiNode::send_command:
-   * @Node: a #SpiNode
+   * @node: a #SpiNode
    * @command: a #SpiCommand
    *
    * Send a command to @node.
@@ -153,14 +184,12 @@ struct _SpiNode {
    */
   SpiResult   (*send_command)         (SpiNode          *node,
                                        SpiCommand       *command);
-  SpiResult   (*get_event)            (SpiNode          *node,
-                                       SpiEvent        **event);
   SpiResult   (*set_event_callback)   (SpiNode          *node,
                                        SpiEventCallback  callback,
                                        void             *user_data);
   /**
    * SpiNode::get_n_ports:
-   * @Node: a #SpiNode
+   * @node: a #SpiNode
    * @n_input_ports: location to hold the number of input ports or %NULL
    * @max_input_ports: location to hold the maximum number of input ports or %NULL
    * @n_output_ports: location to hold the number of output ports or %NULL
@@ -179,7 +208,7 @@ struct _SpiNode {
                                        unsigned int     *max_output_ports);
   /**
    * SpiNode::get_port_ids:
-   * @Node: a #SpiNode
+   * @node: a #SpiNode
    * @n_input_ports: size of the @input_ids array
    * @input_ids: array to store the input stream ids
    * @n_output_ports: size of the @output_ids array
@@ -203,14 +232,68 @@ struct _SpiNode {
   SpiResult   (*remove_port)          (SpiNode          *node,
                                        uint32_t          port_id);
 
+  /**
+   * SpiNode::enum_port_formats:
+   * @node: a #SpiNode
+   * @port_id: the port to query
+   * @index: the format index to retrieve
+   * @format: pointer to a format
+   *
+   * Enumerate all possible formats on @port_id of @node.
+   *
+   * Use the index to retrieve the formats one by one until the function
+   * returns #SPI_RESULT_ENUM_END.
+   *
+   * The result format can be queried and modified and ultimately be used
+   * to call SpiNode::set_port_format.
+   *
+   * Returns: #SPI_RESULT_OK on success
+   *          #SPI_RESULT_INVALID_ARGUMENTS when node or format is %NULL
+   *          #SPI_RESULT_INVALID_PORT when port_id is not valid
+   *          #SPI_RESULT_ENUM_END when no format exists for @index
+   *
+   */
   SpiResult   (*enum_port_formats)    (SpiNode          *node,
                                        uint32_t          port_id,
                                        unsigned int      index,
                                        SpiParams       **format);
+  /**
+   * SpiNode::set_port_format:
+   * @node: a #SpiNode
+   * @port_id: the port to configure
+   * @format: a #SpiParam with the format
+   *
+   * Set a format on @port_id of @node.
+   *
+   * When @format is %NULL, the current format will be removed.
+   *
+   * Returns: #SPI_RESULT_OK on success
+   *          #SPI_RESULT_INVALID_ARGUMENTS when node is %NULL
+   *          #SPI_RESULT_INVALID_PORT when port_id is not valid
+   *          #SPI_RESULT_INVALID_MEDIA_TYPE when the media type is not valid
+   *          #SPI_RESULT_INVALID_FORMAT_PARAMS when one of the mandatory format
+   *                 parameters is not specified.
+   *          #SPI_RESULT_WRONG_PARAM_TYPE when the type of size of a parameter
+   *                 is not correct.
+   */
   SpiResult   (*set_port_format)      (SpiNode          *node,
                                        uint32_t          port_id,
                                        int               test_only,
                                        const SpiParams  *format);
+  /**
+   * SpiNode::get_port_format:
+   * @node: a #SpiNode
+   * @port_id: the port to query
+   * @format: a pointer to a location to hold the #SpiParam
+   *
+   * Get the format on @port_id of @node. The result #SpiParam can
+   * not be modified.
+   *
+   * Returns: #SPI_RESULT_OK on success
+   *          #SPI_RESULT_INVALID_ARGUMENTS when @node or @format is %NULL
+   *          #SPI_RESULT_INVALID_PORT when @port_id is not valid
+   *          #SPI_RESULT_INVALID_NO_FORMAT when no format was set
+   */
   SpiResult   (*get_port_format)      (SpiNode          *node,
                                        uint32_t          port_id,
                                        const SpiParams **format);
@@ -230,11 +313,50 @@ struct _SpiNode {
                                        uint32_t          port_id,
                                        SpiPortStatus    *status);
 
-  SpiResult   (*send_port_data)       (SpiNode          *node,
+  /**
-                                       SpiDataInfo      *data);
+   * SpiNode::push_port_input:
-  SpiResult   (*receive_port_data)    (SpiNode          *node,
+   * @node: a #SpiNode
-                                       unsigned int      n_data,
+   * @n_info: number of #SpiInputInfo in @info
-                                       SpiDataInfo      *data);
+   * @info: array of #SpiInputInfo
+   *
+   * Push a buffer and/or an event into one or more input ports of
+   * @node.
+   *
+   * Returns: #SPI_RESULT_OK on success
+   *          #SPI_RESULT_INVALID_ARGUMENTS when node or info is %NULL
+   *          #SPI_RESULT_ERROR when one or more of the @info has an
+   *                         error result. Check the status of all the
+   *                         @info.
+   *          #SPI_RESULT_HAVE_ENOUGH_INPUT when output can be produced.
+   */
+  SpiResult   (*push_port_input)      (SpiNode          *node,
+                                       unsigned int      n_info,
+                                       SpiInputInfo     *info);
+  /**
+   * SpiNode::pull_port_output:
+   * @node: a #SpiNode
+   * @n_info: number of #SpiOutputInfo in @info
+   * @info: array of #SpiOutputInfo
+   *
+   * Pull a buffer and/or an event from one or more output ports of
+   * @node.
+   *
+   * Returns: #SPI_RESULT_OK on success
+   *          #SPI_RESULT_INVALID_ARGUMENTS when node or info is %NULL
+   *          #SPI_RESULT_PORTS_CHANGED the number of ports has changed. None
+   *                   of the @info fields are modified
+   *          #SPI_RESULT_FORMAT_CHANGED a format changed on some port.
+   *                   the ports that changed are marked in the status.
+   *          #SPI_RESULT_PROPERTIES_CHANGED port properties changed. The
+   *                   changed ports are marked in the status.
+   *          #SPI_RESULT_ERROR when one or more of the @info has an
+   *                   error result. Check the status of all the @info.
+   *          #SPI_RESULT_NEED_MORE_INPUT when no output can be produced
+   *                   because more input is needed.
+   */
+  SpiResult   (*pull_port_output)     (SpiNode          *node,
+                                       unsigned int      n_info,
+                                       SpiOutputInfo    *info);
 
 
   /**

pinos/tests/alsa-utils.c

@@ -335,7 +335,7 @@ direct_loop (void *user_data)
 
         event.refcount = 1;
         event.notify = NULL;
-        event.type = SPI_EVENT_TYPE_REQUEST_DATA;
+        event.type = SPI_EVENT_TYPE_NEED_INPUT;
         event.port_id = 0;
         event.data = buffer;
 

pinos/tests/spi-alsa-sink.c

@@ -28,8 +28,8 @@ typedef struct _SpiALSASink SpiALSASink;
 
 
 static const char default_device[] = "default";
-static const uint32_t default_buffer_time = 40000;
+static const uint32_t default_buffer_time = 10000;
-static const uint32_t default_period_time = 20000;
+static const uint32_t default_period_time = 5000;
 static const bool default_period_event = 0;
 
 typedef struct {
@@ -93,9 +93,6 @@ struct _SpiALSASink {
 
   bool activated;
 
-  SpiEvent *event;
-  SpiEvent last_event;
-
   SpiEventCallback event_cb;
   void *user_data;
 
@@ -188,6 +185,9 @@ enum_param_info (const SpiParams     *params,
                  unsigned int         index,
                  const SpiParamInfo **info)
 {
+  if (params == NULL || info == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (index >= PARAM_ID_LAST)
     return SPI_RESULT_ENUM_END;
   *info = &param_info[index];
@@ -204,6 +204,9 @@ set_param (SpiParams    *params,
   SpiResult res = SPI_RESULT_OK;
   SpiALSASinkParams *p = (SpiALSASinkParams *) params;
 
+  if (params == NULL || value == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   switch (id) {
     case PARAM_ID_DEVICE:
       CHECK_TYPE (type, SPI_PARAM_TYPE_STRING);
@@ -242,6 +245,9 @@ get_param (const SpiParams *params,
   SpiResult res = SPI_RESULT_OK;
   SpiALSASinkParams *p = (SpiALSASinkParams *) params;
 
+  if (params == NULL || type == NULL || size == NULL || value == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   switch (id) {
     case PARAM_ID_DEVICE:
       *type = SPI_PARAM_TYPE_STRING;
@@ -288,6 +294,9 @@ spi_alsa_sink_node_get_params (SpiNode        *node,
   static SpiALSASinkParams p;
   SpiALSASink *this = (SpiALSASink *) node;
 
+  if (node == NULL || params == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   memcpy (&p, &this->params, sizeof (p));
   *params = &p.param;
 
@@ -304,6 +313,9 @@ spi_alsa_sink_node_set_params (SpiNode          *node,
   size_t size;
   const void *value;
 
+  if (node == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (params == NULL) {
     reset_alsa_sink_params (p);
     return SPI_RESULT_OK;
@@ -337,64 +349,61 @@ spi_alsa_sink_node_send_command (SpiNode       *node,
                                  SpiCommand    *command)
 {
   SpiALSASink *this = (SpiALSASink *) node;
-  SpiResult res = SPI_RESULT_NOT_IMPLEMENTED;
+
+  if (node == NULL || command == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
 
   switch (command->type) {
     case SPI_COMMAND_INVALID:
-      res = SPI_RESULT_INVALID_COMMAND;
+      return SPI_RESULT_INVALID_COMMAND;
-      break;
+
     case SPI_COMMAND_ACTIVATE:
       if (!this->activated) {
         spi_alsa_open (this);
         this->activated = true;
       }
-      this->last_event.type = SPI_EVENT_TYPE_ACTIVATED;
+      if (this->event_cb) {
-      this->last_event.data = NULL;
+        SpiEvent event;
-      this->last_event.size = 0;
+
-      this->event = &this->last_event;
+        event.refcount = 1;
-      res = SPI_RESULT_HAVE_EVENT;
+        event.notify = NULL;
+        event.type = SPI_EVENT_TYPE_ACTIVATED;
+        event.port_id = -1;
+        event.data = NULL;
+        event.size = 0;
+
+        this->event_cb (node, &event, this->user_data);
+      }
       break;
     case SPI_COMMAND_DEACTIVATE:
       if (this->activated) {
         spi_alsa_close (this);
         this->activated = false;
       }
-      this->last_event.type = SPI_EVENT_TYPE_DEACTIVATED;
+      if (this->event_cb) {
-      this->last_event.data = NULL;
+        SpiEvent event;
-      this->last_event.size = 0;
+
-      this->event = &this->last_event;
+        event.refcount = 1;
-      res = SPI_RESULT_HAVE_EVENT;
+        event.notify = NULL;
+        event.type = SPI_EVENT_TYPE_DEACTIVATED;
+        event.port_id = -1;
+        event.data = NULL;
+        event.size = 0;
+
+        this->event_cb (node, &event, this->user_data);
+      }
       break;
     case SPI_COMMAND_START:
       spi_alsa_start (this);
-      res = SPI_RESULT_OK;
       break;
     case SPI_COMMAND_STOP:
       spi_alsa_stop (this);
-      res = SPI_RESULT_OK;
       break;
     case SPI_COMMAND_FLUSH:
-      break;
     case SPI_COMMAND_DRAIN:
-      break;
     case SPI_COMMAND_MARKER:
-      break;
+      return SPI_RESULT_NOT_IMPLEMENTED;
   }
-  return res;
-}
-
-static SpiResult
-spi_alsa_sink_node_get_event (SpiNode     *node,
-                              SpiEvent   **event)
-{
-  SpiALSASink *this = (SpiALSASink *) node;
-
-  if (this->event == NULL)
-    return SPI_RESULT_ERROR;
-
-  *event = this->event;
-  this->event = NULL;
-
   return SPI_RESULT_OK;
 }
 
@@ -405,6 +414,9 @@ spi_alsa_sink_node_set_event_callback (SpiNode       *node,
 {
   SpiALSASink *this = (SpiALSASink *) node;
 
+  if (node == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   this->event_cb = event;
   this->user_data = user_data;
 
@@ -418,10 +430,18 @@ spi_alsa_sink_node_get_n_ports (SpiNode       *node,
                                 unsigned int  *n_output_ports,
                                 unsigned int  *max_output_ports)
 {
+  if (node == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
+  if (n_input_ports)
     *n_input_ports = 1;
+  if (n_output_ports)
     *n_output_ports = 0;
+  if (max_input_ports)
     *max_input_ports = 1;
+  if (max_output_ports)
     *max_output_ports = 0;
+
   return SPI_RESULT_OK;
 }
 
@@ -432,6 +452,9 @@ spi_alsa_sink_node_get_port_ids (SpiNode       *node,
                                  unsigned int   n_output_ports,
                                  uint32_t      *output_ids)
 {
+  if (node == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (n_input_ports > 0)
     input_ids[0] = 0;
 
@@ -545,6 +568,9 @@ enum_raw_format_param_info (const SpiParams     *params,
                             unsigned int         index,
                             const SpiParamInfo **info)
 {
+  if (params == NULL || info == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (index >= 5)
     return SPI_RESULT_ENUM_END;
   *info = &raw_format_param_info[index];
@@ -599,6 +625,9 @@ enum_mpeg_format_param_info (const SpiParams     *params,
                              unsigned int         index,
                              const SpiParamInfo **info)
 {
+  if (params == NULL || info == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (index >= 4)
     return SPI_RESULT_ENUM_END;
   *info = &mpeg_format_param_info[index];
@@ -618,6 +647,9 @@ set_format_param (SpiParams    *params,
 {
   SpiALSASinkFormat *f = (SpiALSASinkFormat *) params;
 
+  if (params == NULL || value == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   switch (id) {
     case SPI_PARAM_ID_FORMAT:
       CHECK_TYPE (type, SPI_PARAM_TYPE_STRING);
@@ -665,6 +697,9 @@ get_format_param (const SpiParams *params,
 {
   SpiALSASinkFormat *f = (SpiALSASinkFormat *) params;
 
+  if (params == NULL || type == NULL || size == NULL || value == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   switch (id) {
     case SPI_PARAM_ID_MEDIA_TYPE:
       CHECK_UNSET (f->unset_mask, 0);
@@ -729,6 +764,9 @@ spi_alsa_sink_node_enum_port_formats (SpiNode          *node,
 {
   static SpiALSASinkFormat fmt;
 
+  if (node == NULL || format == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (port_id != 0)
     return SPI_RESULT_INVALID_PORT;
 
@@ -767,11 +805,16 @@ spi_alsa_sink_node_set_port_format (SpiNode         *node,
   SpiParamType type;
   size_t size;
   const void *value;
-  SpiALSASinkFormat *fmt = &this->current_format;
+  SpiALSASinkFormat *fmt;
+
+  if (node == NULL || format == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
 
   if (port_id != 0)
     return SPI_RESULT_INVALID_PORT;
 
+  fmt = &this->current_format;
+
   if (format == NULL) {
     fmt->param.get_param = NULL;
     this->have_format = 0;
@@ -834,6 +877,9 @@ spi_alsa_sink_node_get_port_format (SpiNode          *node,
 {
   SpiALSASink *this = (SpiALSASink *) node;
 
+  if (node == NULL || format == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (port_id != 0)
     return SPI_RESULT_INVALID_PORT;
 
@@ -850,6 +896,9 @@ spi_alsa_sink_node_get_port_info (SpiNode       *node,
                                   uint32_t       port_id,
                                   SpiPortInfo   *info)
 {
+  if (node == NULL || info == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (port_id != 0)
     return SPI_RESULT_INVALID_PORT;
 
@@ -879,6 +928,9 @@ spi_alsa_sink_node_get_port_status (SpiNode        *node,
                                     uint32_t        port_id,
                                     SpiPortStatus  *status)
 {
+  if (node == NULL || status == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (port_id != 0)
     return SPI_RESULT_INVALID_PORT;
 
@@ -888,34 +940,74 @@ spi_alsa_sink_node_get_port_status (SpiNode        *node,
 }
 
 static SpiResult
-spi_alsa_sink_node_send_port_data (SpiNode       *node,
+spi_alsa_sink_node_push_port_input (SpiNode        *node,
-                                   SpiDataInfo   *data)
+                                    unsigned int    n_info,
+                                    SpiInputInfo   *info)
 {
   SpiALSASink *this = (SpiALSASink *) node;
+  unsigned int i;
+  bool have_error = false, have_enough = false;
 
-  if (data->port_id != 0)
+  if (node == NULL || n_info == 0 || info == NULL)
-    return SPI_RESULT_INVALID_PORT;
+    return SPI_RESULT_INVALID_ARGUMENTS;
 
-  if (data->buffer != NULL) {
+  for (i = 0; i < n_info; i++) {
-    if (!this->have_format)
+    if (info[i].port_id != 0) {
-      return SPI_RESULT_NO_FORMAT;
+      info[i].status = SPI_RESULT_INVALID_PORT;
+      have_error = true;
+      continue;
+    }
 
-    if (this->input_buffer != NULL)
+    if (info[i].buffer != NULL) {
+      if (!this->have_format) {
+        info[i].status = SPI_RESULT_NO_FORMAT;
+        have_error = true;
+        continue;
+      }
+
+      if (this->input_buffer != NULL) {
+        info[i].status = SPI_RESULT_HAVE_ENOUGH_INPUT;
+        have_enough = true;
+        continue;
+      }
+      this->input_buffer = spi_buffer_ref (info[i].buffer);
+    }
+    info[i].status = SPI_RESULT_OK;
+  }
+  if (have_error)
+    return SPI_RESULT_ERROR;
+  if (have_enough)
     return SPI_RESULT_HAVE_ENOUGH_INPUT;
 
-    this->input_buffer = spi_buffer_ref (data->buffer);
-  }
   return SPI_RESULT_OK;
 }
 
 static SpiResult
-spi_alsa_sink_node_receive_port_data (SpiNode      *node,
+spi_alsa_sink_node_pull_port_output (SpiNode        *node,
-                                      unsigned int  n_data,
+                                     unsigned int    n_info,
-                                      SpiDataInfo  *data)
+                                     SpiOutputInfo  *info)
 {
   return SPI_RESULT_INVALID_PORT;
 }
 
+static SpiResult
+spi_alsa_sink_node_enum_interface_info (SpiNode                  *node,
+                                        unsigned int             index,
+                                        const SpiInterfaceInfo **info)
+
+{
+  return SPI_RESULT_NOT_IMPLEMENTED;
+}
+
+static SpiResult
+spi_alsa_sink_node_get_interface (SpiNode                 *node,
+                                  uint32_t                 interface_id,
+                                  void                   **interface)
+{
+  return SPI_RESULT_NOT_IMPLEMENTED;
+}
+
+
 SpiNode *
 spi_alsa_sink_new (void)
 {
@@ -927,7 +1019,6 @@ spi_alsa_sink_new (void)
   node->get_params = spi_alsa_sink_node_get_params;
   node->set_params = spi_alsa_sink_node_set_params;
   node->send_command = spi_alsa_sink_node_send_command;
-  node->get_event = spi_alsa_sink_node_get_event;
   node->set_event_callback = spi_alsa_sink_node_set_event_callback;
   node->get_n_ports = spi_alsa_sink_node_get_n_ports;
   node->get_port_ids = spi_alsa_sink_node_get_port_ids;
@@ -940,8 +1031,10 @@ spi_alsa_sink_new (void)
   node->get_port_params = spi_alsa_sink_node_get_port_params;
   node->set_port_params = spi_alsa_sink_node_set_port_params;
   node->get_port_status = spi_alsa_sink_node_get_port_status;
-  node->send_port_data = spi_alsa_sink_node_send_port_data;
+  node->push_port_input = spi_alsa_sink_node_push_port_input;
-  node->receive_port_data = spi_alsa_sink_node_receive_port_data;
+  node->pull_port_output = spi_alsa_sink_node_pull_port_output;
+  node->enum_interface_info = spi_alsa_sink_node_enum_interface_info;
+  node->get_interface = spi_alsa_sink_node_get_interface;
 
   this = (SpiALSASink *) node;
   this->params.param.enum_param_info = enum_param_info;

pinos/tests/spi-audiotestsrc.c

@@ -48,9 +48,6 @@ struct _SpiAudioTestSrc {
   SpiAudioTestSrcParams params;
   SpiAudioTestSrcParams tmp_params;
 
-  SpiEvent *event;
-  SpiEvent last_event;
-
   SpiEventCallback event_cb;
   void *user_data;
 
@@ -277,58 +274,51 @@ spi_audiotestsrc_node_send_command (SpiNode       *node,
                                     SpiCommand    *command)
 {
   SpiAudioTestSrc *this = (SpiAudioTestSrc *) node;
-  SpiResult res = SPI_RESULT_NOT_IMPLEMENTED;
 
   if (node == NULL || command == NULL)
     return SPI_RESULT_INVALID_ARGUMENTS;
 
   switch (command->type) {
     case SPI_COMMAND_INVALID:
-      res = SPI_RESULT_INVALID_COMMAND;
+      return SPI_RESULT_INVALID_COMMAND;
-      break;
+
     case SPI_COMMAND_ACTIVATE:
-      this->last_event.type = SPI_EVENT_TYPE_ACTIVATED;
+      if (this->event_cb) {
-      this->last_event.data = NULL;
+        SpiEvent event;
-      this->last_event.size = 0;
+
-      this->event = &this->last_event;
+        event.refcount = 1;
-      res = SPI_RESULT_HAVE_EVENT;
+        event.notify = NULL;
+        event.type = SPI_EVENT_TYPE_ACTIVATED;
+        event.port_id = -1;
+        event.data = NULL;
+        event.size = 0;
+
+        this->event_cb (node, &event, this->user_data);
+      }
       break;
+
     case SPI_COMMAND_DEACTIVATE:
-      this->last_event.type = SPI_EVENT_TYPE_DEACTIVATED;
+      if (this->event_cb) {
-      this->last_event.data = NULL;
+        SpiEvent event;
-      this->last_event.size = 0;
+
-      this->event = &this->last_event;
+        event.refcount = 1;
-      res = SPI_RESULT_HAVE_EVENT;
+        event.notify = NULL;
+        event.type = SPI_EVENT_TYPE_DEACTIVATED;
+        event.port_id = -1;
+        event.data = NULL;
+        event.size = 0;
+
+        this->event_cb (node, &event, this->user_data);
+      }
       break;
+
     case SPI_COMMAND_START:
-      break;
     case SPI_COMMAND_STOP:
-      break;
     case SPI_COMMAND_FLUSH:
-      break;
     case SPI_COMMAND_DRAIN:
-      break;
     case SPI_COMMAND_MARKER:
-      break;
+      return SPI_RESULT_NOT_IMPLEMENTED;
   }
-  return res;
-}
-
-static SpiResult
-spi_audiotestsrc_node_get_event (SpiNode     *node,
-                                 SpiEvent   **event)
-{
-  SpiAudioTestSrc *this = (SpiAudioTestSrc *) node;
-
-  if (node == NULL || event == NULL)
-    return SPI_RESULT_INVALID_ARGUMENTS;
-
-  if (this->event == NULL)
-    return SPI_RESULT_ERROR;
-
-  *event = this->event;
-  this->event = NULL;
-
   return SPI_RESULT_OK;
 }
 
@@ -790,42 +780,78 @@ spi_audiotestsrc_node_get_port_status (SpiNode        *node,
 }
 
 static SpiResult
-spi_audiotestsrc_node_send_port_data (SpiNode       *node,
+spi_audiotestsrc_node_push_port_input (SpiNode       *node,
-                                      SpiDataInfo   *data)
+                                       unsigned int   n_info,
+                                       SpiInputInfo  *info)
 {
   return SPI_RESULT_INVALID_PORT;
 }
 
 static SpiResult
-spi_audiotestsrc_node_receive_port_data (SpiNode      *node,
+spi_audiotestsrc_node_pull_port_output (SpiNode       *node,
-                                         unsigned int  n_data,
+                                        unsigned int   n_info,
-                                         SpiDataInfo  *data)
+                                        SpiOutputInfo *info)
 {
   SpiAudioTestSrc *this = (SpiAudioTestSrc *) node;
-  size_t i, size;
+  size_t j, size;
   uint8_t *ptr;
+  unsigned int i;
+  bool have_error = false;
 
-  if (node == NULL || n_data == 0 || data == NULL)
+  if (node == NULL || n_info == 0 || info == NULL)
     return SPI_RESULT_INVALID_ARGUMENTS;
 
-  if (data->port_id != 0)
+  for (i = 0; i < n_info; i++) {
-    return SPI_RESULT_INVALID_PORT;
+    if (info[i].port_id != 0) {
+      info[i].status = SPI_RESULT_INVALID_PORT;
+      have_error = true;
+      continue;
+    }
 
-  if (!this->have_format)
+    if (!this->have_format) {
-    return SPI_RESULT_NO_FORMAT;
+      info[i].status = SPI_RESULT_NO_FORMAT;
+      have_error = true;
+      continue;
+    }
 
-  if (data->buffer == NULL)
+    if (info[i].buffer == NULL || info[i].buffer->n_datas == 0) {
-    return SPI_RESULT_INVALID_ARGUMENTS;
+      info[i].status = SPI_RESULT_INVALID_ARGUMENTS;
+      have_error = true;
+      continue;
+    }
 
-  ptr = data->buffer->datas[0].data;
+    ptr = info[i].buffer->datas[0].data;
-  size = data->buffer->datas[0].size;
+    size = info[i].buffer->datas[0].size;
 
-  for (i = 0; i < size; i++)
+    for (j = 0; j < size; j++)
-    ptr[i] = rand();
+      ptr[j] = rand();
+
+    info[i].status = SPI_RESULT_OK;
+  }
+  if (have_error)
+    return SPI_RESULT_ERROR;
 
   return SPI_RESULT_OK;
 }
 
+static SpiResult
+spi_audiotestsrc_node_enum_interface_info (SpiNode                  *node,
+                                           unsigned int             index,
+                                           const SpiInterfaceInfo **info)
+
+{
+  return SPI_RESULT_NOT_IMPLEMENTED;
+}
+
+static SpiResult
+spi_audiotestsrc_node_get_interface (SpiNode                 *node,
+                                     uint32_t                 interface_id,
+                                     void                   **interface)
+{
+  return SPI_RESULT_NOT_IMPLEMENTED;
+}
+
+
 SpiNode *
 spi_audiotestsrc_new (void)
 {
@@ -837,7 +863,6 @@ spi_audiotestsrc_new (void)
   node->get_params = spi_audiotestsrc_node_get_params;
   node->set_params = spi_audiotestsrc_node_set_params;
   node->send_command = spi_audiotestsrc_node_send_command;
-  node->get_event = spi_audiotestsrc_node_get_event;
   node->set_event_callback = spi_audiotestsrc_node_set_event_callback;
   node->get_n_ports = spi_audiotestsrc_node_get_n_ports;
   node->get_port_ids = spi_audiotestsrc_node_get_port_ids;
@@ -850,8 +875,10 @@ spi_audiotestsrc_new (void)
   node->get_port_params = spi_audiotestsrc_node_get_port_params;
   node->set_port_params = spi_audiotestsrc_node_set_port_params;
   node->get_port_status = spi_audiotestsrc_node_get_port_status;
-  node->send_port_data = spi_audiotestsrc_node_send_port_data;
+  node->push_port_input = spi_audiotestsrc_node_push_port_input;
-  node->receive_port_data = spi_audiotestsrc_node_receive_port_data;
+  node->pull_port_output = spi_audiotestsrc_node_pull_port_output;
+  node->enum_interface_info = spi_audiotestsrc_node_enum_interface_info;
+  node->get_interface = spi_audiotestsrc_node_get_interface;
 
   this = (SpiAudioTestSrc *) node;
   this->params.param.enum_param_info = enum_param_info;

pinos/tests/spi-volume.c

@@ -48,9 +48,6 @@ struct _SpiVolume {
   SpiVolumeParams params;
   SpiVolumeParams tmp_params;
 
-  SpiEvent *event;
-  SpiEvent last_event;
-
   SpiEventCallback event_cb;
   void *user_data;
 
@@ -109,9 +106,14 @@ enum_param_info (const SpiParams     *params,
                  unsigned int         index,
                  const SpiParamInfo **info)
 {
+  if (params == NULL || info == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (index >= PARAM_ID_LAST)
     return SPI_RESULT_ENUM_END;
+
   *info = &param_info[index];
+
   return SPI_RESULT_OK;
 }
 
@@ -235,55 +237,51 @@ spi_volume_node_send_command (SpiNode       *node,
                               SpiCommand    *command)
 {
   SpiVolume *this = (SpiVolume *) node;
-  SpiResult res = SPI_RESULT_NOT_IMPLEMENTED;
 
   if (node == NULL || command == NULL)
     return SPI_RESULT_INVALID_ARGUMENTS;
 
   switch (command->type) {
     case SPI_COMMAND_INVALID:
-      res = SPI_RESULT_INVALID_COMMAND;
+      return SPI_RESULT_INVALID_COMMAND;
-      break;
+
     case SPI_COMMAND_ACTIVATE:
-      this->last_event.type = SPI_EVENT_TYPE_ACTIVATED;
+      if (this->event_cb) {
-      this->last_event.data = NULL;
+        SpiEvent event;
-      this->last_event.size = 0;
+
-      this->event = &this->last_event;
+        event.refcount = 1;
-      res = SPI_RESULT_HAVE_EVENT;
+        event.notify = NULL;
+        event.type = SPI_EVENT_TYPE_ACTIVATED;
+        event.port_id = -1;
+        event.data = NULL;
+        event.size = 0;
+
+        this->event_cb (node, &event, this->user_data);
+      }
       break;
+
     case SPI_COMMAND_DEACTIVATE:
-      this->last_event.type = SPI_EVENT_TYPE_DEACTIVATED;
+      if (this->event_cb) {
-      this->last_event.data = NULL;
+        SpiEvent event;
-      this->last_event.size = 0;
+
-      this->event = &this->last_event;
+        event.refcount = 1;
-      res = SPI_RESULT_HAVE_EVENT;
+        event.notify = NULL;
+        event.type = SPI_EVENT_TYPE_DEACTIVATED;
+        event.port_id = -1;
+        event.data = NULL;
+        event.size = 0;
+
+        this->event_cb (node, &event, this->user_data);
+      }
       break;
+
     case SPI_COMMAND_START:
-      break;
     case SPI_COMMAND_STOP:
-      break;
     case SPI_COMMAND_FLUSH:
-      break;
     case SPI_COMMAND_DRAIN:
-      break;
     case SPI_COMMAND_MARKER:
-      break;
+      return SPI_RESULT_NOT_IMPLEMENTED;
   }
-  return res;
-}
-
-static SpiResult
-spi_volume_node_get_event (SpiNode     *node,
-                           SpiEvent   **event)
-{
-  SpiVolume *this = (SpiVolume *) node;
-
-  if (this->event == NULL)
-    return SPI_RESULT_ERROR;
-
-  *event = this->event;
-  this->event = NULL;
-
   return SPI_RESULT_OK;
 }
 
@@ -294,6 +292,9 @@ spi_volume_node_set_event_callback (SpiNode       *node,
 {
   SpiVolume *this = (SpiVolume *) node;
 
+  if (node == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   this->event_cb = event;
   this->user_data = user_data;
 
@@ -329,9 +330,12 @@ spi_volume_node_get_port_ids (SpiNode       *node,
                               unsigned int   n_output_ports,
                               uint32_t      *output_ids)
 {
-  if (n_input_ports > 0)
+  if (node == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
+  if (n_input_ports > 0 && input_ids)
     input_ids[0] = 0;
-  if (n_output_ports > 0)
+  if (n_output_ports > 0 && output_ids)
     output_ids[0] = 1;
 
   return SPI_RESULT_OK;
@@ -451,9 +455,14 @@ enum_raw_format_param_info (const SpiParams     *params,
                             unsigned int         index,
                             const SpiParamInfo **info)
 {
+  if (params == NULL || info == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (index >= 4)
     return SPI_RESULT_ENUM_END;
+
   *info = &raw_format_param_info[index];
+
   return SPI_RESULT_OK;
 }
 
@@ -469,6 +478,9 @@ set_format_param (SpiParams    *params,
 {
   SpiVolumeFormat *f = (SpiVolumeFormat *) params;
 
+  if (params == NULL || value == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   switch (id) {
     case SPI_PARAM_ID_FORMAT:
       CHECK_TYPE (type, SPI_PARAM_TYPE_STRING);
@@ -512,6 +524,9 @@ get_format_param (const SpiParams *params,
 {
   SpiVolumeFormat *f = (SpiVolumeFormat *) params;
 
+  if (params == NULL || type == NULL || size == NULL || value == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   switch (id) {
     case SPI_PARAM_ID_MEDIA_TYPE:
       CHECK_UNSET (f->unset_mask, 0);
@@ -558,6 +573,9 @@ spi_volume_node_enum_port_formats (SpiNode          *node,
 {
   static SpiVolumeFormat fmt;
 
+  if (node == NULL || format == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (port_id != 0)
     return SPI_RESULT_INVALID_PORT;
 
@@ -587,11 +605,16 @@ spi_volume_node_set_port_format (SpiNode         *node,
   SpiParamType type;
   size_t size;
   const void *value;
-  SpiVolumeFormat *fmt = &this->current_format;
+  SpiVolumeFormat *fmt;
+
+  if (node == NULL || format == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
 
   if (port_id != 0)
     return SPI_RESULT_INVALID_PORT;
 
+  fmt = &this->current_format;
+
   if (format == NULL) {
     fmt->param.get_param = NULL;
     this->have_format = false;
@@ -653,6 +676,9 @@ spi_volume_node_get_port_format (SpiNode          *node,
 {
   SpiVolume *this = (SpiVolume *) node;
 
+  if (node == NULL || format == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (port_id != 0)
     return SPI_RESULT_INVALID_PORT;
 
@@ -669,6 +695,9 @@ spi_volume_node_get_port_info (SpiNode       *node,
                                uint32_t       port_id,
                                SpiPortInfo   *info)
 {
+  if (node == NULL || info == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   switch (port_id) {
     case 0:
       info->flags = SPI_PORT_INFO_FLAG_CAN_USE_BUFFER |
@@ -709,6 +738,9 @@ spi_volume_node_get_port_status (SpiNode        *node,
   SpiVolume *this = (SpiVolume *) node;
   SpiPortStatusFlags flags = 0;
 
+  if (node == NULL || status == NULL)
+    return SPI_RESULT_INVALID_ARGUMENTS;
+
   if (!this->have_format)
     return SPI_RESULT_NO_FORMAT;
 
@@ -730,32 +762,48 @@ spi_volume_node_get_port_status (SpiNode        *node,
 }
 
 static SpiResult
-spi_volume_node_send_port_data (SpiNode       *node,
+spi_volume_node_push_port_input (SpiNode        *node,
-                                SpiDataInfo   *data)
+                                 unsigned int    n_info,
+                                 SpiInputInfo   *info)
 {
   SpiVolume *this = (SpiVolume *) node;
   SpiBuffer *buffer;
   SpiEvent *event;
+  unsigned int i;
+  bool have_error = false;
+  bool have_enough = false;
 
-  if (node == NULL || data == NULL)
+  if (node == NULL || n_info == 0 || info == NULL)
     return SPI_RESULT_INVALID_ARGUMENTS;
 
-  if (data->port_id != 0)
+  for (i = 0; i < n_info; i++) {
-    return SPI_RESULT_INVALID_PORT;
+    if (info[i].port_id != 0) {
+      info[i].status = SPI_RESULT_INVALID_PORT;
+      have_error = true;
+      continue;
+    }
 
-  event = data->event;
+    event = info[i].event;
-  buffer = data->buffer;
+    buffer = info[i].buffer;
 
-  if (buffer == NULL && event == NULL)
+    if (buffer == NULL && event == NULL) {
-    return SPI_RESULT_INVALID_ARGUMENTS;
+      info[i].status = SPI_RESULT_INVALID_ARGUMENTS;
-
+      have_error = true;
-  if (!this->have_format)
+      continue;
-    return SPI_RESULT_NO_FORMAT;
+    }
 
     if (buffer) {
-    if (this->input_buffer != NULL)
+      if (!this->have_format) {
-      return SPI_RESULT_HAVE_ENOUGH_INPUT;
+        info[i].status = SPI_RESULT_NO_FORMAT;
+        have_error = true;
+        continue;
+      }
 
+      if (this->input_buffer != NULL) {
+        info[i].status = SPI_RESULT_HAVE_ENOUGH_INPUT;
+        have_enough = true;
+        continue;
+      }
       this->input_buffer = spi_buffer_ref (buffer);
     }
     if (event) {
@@ -764,6 +812,12 @@ spi_volume_node_send_port_data (SpiNode       *node,
           break;
       }
     }
+    info[i].status = SPI_RESULT_OK;
+  }
+  if (have_error)
+    return SPI_RESULT_ERROR;
+  if (have_enough)
+    return SPI_RESULT_HAVE_ENOUGH_INPUT;
 
   return SPI_RESULT_OK;
 }
@@ -771,9 +825,9 @@ spi_volume_node_send_port_data (SpiNode       *node,
 #define MIN(a,b) ((a) < (b) ? (a) : (b))
 
 static SpiResult
-spi_volume_node_receive_port_data (SpiNode      *node,
+spi_volume_node_pull_port_output (SpiNode        *node,
-                                   unsigned int  n_data,
+                                  unsigned int    n_info,
-                                   SpiDataInfo  *data)
+                                  SpiOutputInfo  *info)
 {
   SpiVolume *this = (SpiVolume *) node;
   unsigned int si, di, i, n_samples, n_bytes, soff, doff ;
@@ -782,10 +836,10 @@ spi_volume_node_receive_port_data (SpiNode      *node,
   uint16_t *src, *dst;
   double volume;
 
-  if (node == NULL || n_data == 0 || data == NULL)
+  if (node == NULL || n_info == 0 || info == NULL)
     return SPI_RESULT_INVALID_ARGUMENTS;
 
-  if (data->port_id != 1)
+  if (info->port_id != 1)
     return SPI_RESULT_INVALID_PORT;
 
   if (!this->have_format)
@@ -797,7 +851,7 @@ spi_volume_node_receive_port_data (SpiNode      *node,
   volume = this->params.volume;
 
   sbuf = this->input_buffer;
-  dbuf = data->buffer ? data->buffer : this->input_buffer;
+  dbuf = info->buffer ? info->buffer : this->input_buffer;
 
   si = di = 0;
   soff = doff = 0;
@@ -843,11 +897,28 @@ spi_volume_node_receive_port_data (SpiNode      *node,
     spi_buffer_unref (sbuf);
 
   this->input_buffer = NULL;
-  data->buffer = dbuf;
+  info->buffer = dbuf;
 
   return SPI_RESULT_OK;
 }
 
+static SpiResult
+spi_volume_node_enum_interface_info (SpiNode                  *node,
+                                     unsigned int             index,
+                                     const SpiInterfaceInfo **info)
+
+{
+  return SPI_RESULT_NOT_IMPLEMENTED;
+}
+
+static SpiResult
+spi_volume_node_get_interface (SpiNode                 *node,
+                               uint32_t                 interface_id,
+                               void                   **interface)
+{
+  return SPI_RESULT_NOT_IMPLEMENTED;
+}
+
 SpiNode *
 spi_volume_new (void)
 {
@@ -859,7 +930,6 @@ spi_volume_new (void)
   node->get_params = spi_volume_node_get_params;
   node->set_params = spi_volume_node_set_params;
   node->send_command = spi_volume_node_send_command;
-  node->get_event = spi_volume_node_get_event;
   node->set_event_callback = spi_volume_node_set_event_callback;
   node->get_n_ports = spi_volume_node_get_n_ports;
   node->get_port_ids = spi_volume_node_get_port_ids;
@@ -872,8 +942,10 @@ spi_volume_new (void)
   node->get_port_params = spi_volume_node_get_port_params;
   node->set_port_params = spi_volume_node_set_port_params;
   node->get_port_status = spi_volume_node_get_port_status;
-  node->send_port_data = spi_volume_node_send_port_data;
+  node->push_port_input = spi_volume_node_push_port_input;
-  node->receive_port_data = spi_volume_node_receive_port_data;
+  node->pull_port_output = spi_volume_node_pull_port_output;
+  node->enum_interface_info = spi_volume_node_enum_interface_info;
+  node->get_interface = spi_volume_node_get_interface;
 
   this = (SpiVolume *) node;
   this->params.param.enum_param_info = enum_param_info;

pinos/tests/test-node.c

@@ -176,11 +176,6 @@ inspect_node (SpiNode *node)
     printf ("got params %p\n", params);
 }
 
-static void
-set_params (AppData *data)
-{
-}
-
 static void
 set_format (AppData *data)
 {
@@ -210,49 +205,6 @@ set_format (AppData *data)
     printf ("set format failed: %d\n", res);
 }
 
-static void
-handle_event (SpiNode *node)
-{
-  SpiEvent *event;
-  SpiResult res;
-
-  if ((res = node->get_event (node, &event)) < 0)
-    printf ("got result %d\n", res);
-
-  switch (event->type) {
-    case SPI_EVENT_TYPE_INVALID:
-      printf ("got invalid notify\n");
-      break;
-    case SPI_EVENT_TYPE_ACTIVATED:
-      printf ("got activated notify\n");
-      break;
-    case SPI_EVENT_TYPE_DEACTIVATED:
-      printf ("got deactivated notify\n");
-      break;
-    case SPI_EVENT_TYPE_HAVE_OUTPUT:
-      printf ("got have-output notify\n");
-      break;
-    case SPI_EVENT_TYPE_NEED_INPUT:
-      printf ("got need-input notify\n");
-      break;
-    case SPI_EVENT_TYPE_REQUEST_DATA:
-      printf ("got request-data notify\n");
-      break;
-    case SPI_EVENT_TYPE_DRAINED:
-      printf ("got drained notify\n");
-      break;
-    case SPI_EVENT_TYPE_MARKER:
-      printf ("got marker notify\n");
-      break;
-    case SPI_EVENT_TYPE_ERROR:
-      printf ("got error notify\n");
-      break;
-    case SPI_EVENT_TYPE_BUFFERING:
-      printf ("got noffering notify\n");
-      break;
-  }
-}
-
 typedef struct _MyBuffer MyBuffer;
 
 struct _MyBuffer {
@@ -411,27 +363,34 @@ on_event (SpiNode *node, SpiEvent *event, void *user_data)
   AppData *data = user_data;
 
   switch (event->type) {
-    case SPI_EVENT_TYPE_REQUEST_DATA:
+    case SPI_EVENT_TYPE_NEED_INPUT:
     {
       SpiBuffer *buf;
-      SpiDataInfo info;
+      SpiInputInfo iinfo;
+      SpiOutputInfo oinfo;
       SpiResult res;
 
       buf = event->data;
 
-      info.port_id = event->port_id;
+      oinfo.port_id = event->port_id;
-      info.flags = SPI_DATA_FLAG_NONE;
+      oinfo.flags = SPI_OUTPUT_FLAG_NONE;
-      info.buffer = buf;
+      oinfo.buffer = buf;
-      info.event = NULL;
+      oinfo.event = NULL;
 
-      if ((res = data->src->receive_port_data (data->src, 1, &info)) < 0)
+      if ((res = data->src->pull_port_output (data->src, 1, &oinfo)) < 0)
         printf ("got error %d\n", res);
 
-      if ((res = data->sink->send_port_data (data->sink, &info)) < 0)
+      iinfo.port_id = 0;
+      iinfo.flags = SPI_INPUT_FLAG_NONE;
+      iinfo.buffer = oinfo.buffer;
+      iinfo.event = oinfo.event;
+
+      if ((res = data->sink->push_port_input (data->sink, 1, &iinfo)) < 0)
         printf ("got error %d\n", res);
       break;
     }
     default:
+      printf ("got event %d\n", event->type);
       break;
   }
 }
@@ -444,8 +403,6 @@ run_async_sink (AppData *data)
 
   set_format (data);
 
-  data->sink->set_event_callback (data->sink, on_event, data);
-
   cmd.type = SPI_COMMAND_START;
   if ((res = data->sink->send_command (data->sink, &cmd)) < 0)
     printf ("got error %d\n", res);
@@ -480,6 +437,8 @@ setup_sink (AppData *data)
 
   data->sink = spi_alsa_sink_new ();
 
+  data->sink->set_event_callback (data->sink, on_event, data);
+
   if ((res = data->sink->get_params (data->sink, &params)) < 0)
     printf ("got get_params error %d\n", res);