Remove direction

Direction is tied to the port id.
Handle nodes with fixed ports.

pinos/client/stream.c

@@ -575,8 +575,6 @@ add_port_update (PinosStream *stream, SpaControlBuilder *builder, uint32_t chang
 
   pu.port_id = 0;
   pu.change_mask = change_mask;
-  if (change_mask & SPA_CONTROL_CMD_PORT_UPDATE_DIRECTION)
-    pu.direction = priv->direction;
   if (change_mask & SPA_CONTROL_CMD_PORT_UPDATE_POSSIBLE_FORMATS) {
     pu.n_possible_formats = priv->possible_formats->len;
     pu.possible_formats = (SpaFormat **)priv->possible_formats->pdata;
@@ -1003,8 +1001,7 @@ on_node_proxy (GObject      *source_object,
                                      SPA_CONTROL_CMD_NODE_UPDATE_MAX_OUTPUTS);
 
   priv->port_info.flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
-  add_port_update (stream, &builder, SPA_CONTROL_CMD_PORT_UPDATE_DIRECTION |
+  add_port_update (stream, &builder, SPA_CONTROL_CMD_PORT_UPDATE_POSSIBLE_FORMATS |
-                                     SPA_CONTROL_CMD_PORT_UPDATE_POSSIBLE_FORMATS |
                                      SPA_CONTROL_CMD_PORT_UPDATE_INFO);
 
   add_state_change (stream, &builder, SPA_NODE_STATE_CONFIGURE);

pinos/modules/gst/gst-sink.c

@@ -31,7 +31,6 @@
 typedef struct {
   PinosGstSink *sink;
 
-  guint id;
   PinosPort *port;
 
   GstElement *src;
@@ -384,7 +383,6 @@ free_sink_port_data (SinkPortData *data)
 
 static PinosPort *
 add_port (PinosNode       *node,
-          PinosDirection   direction,
           guint            id,
           GError         **error)
 {
@@ -394,9 +392,8 @@ add_port (PinosNode       *node,
 
   data = g_slice_new0 (SinkPortData);
   data->sink = sink;
-  data->id = id;
   data->port = PINOS_NODE_CLASS (pinos_gst_sink_parent_class)
-                ->add_port (node, direction, id, error);
+                ->add_port (node, id, error);
 
   g_debug ("connecting signals");
   g_signal_connect (data->port, "activate", (GCallback) on_activate, data);
@@ -422,7 +419,7 @@ add_port (PinosNode       *node,
 
 static gboolean
 remove_port (PinosNode       *node,
-             guint            id)
+             PinosPort       *port)
 {
   PinosGstSink *sink = PINOS_GST_SINK (node);
   PinosGstSinkPrivate *priv = sink->priv;
@@ -431,7 +428,7 @@ remove_port (PinosNode       *node,
   for (walk = priv->ports; walk; walk = g_list_next (walk)) {
     SinkPortData *data = walk->data;
 
-    if (data->id == id) {
+    if (data->port == port) {
       free_sink_port_data (data);
       priv->ports = g_list_delete_link (priv->ports, walk);
       break;

pinos/modules/gst/gst-source.c

@@ -433,7 +433,7 @@ free_source_port_data (SourcePortData *data)
 
 static gboolean
 remove_port (PinosNode       *node,
-             guint            id)
+             PinosPort       *port)
 {
   PinosGstSource *source = PINOS_GST_SOURCE (node);
   PinosGstSourcePrivate *priv = source->priv;
@@ -442,7 +442,7 @@ remove_port (PinosNode       *node,
   for (walk = priv->ports; walk; walk = g_list_next (walk)) {
     SourcePortData *data = walk->data;
 
-    if (data->id == id) {
+    if (data->port == port) {
       free_source_port_data (data);
       priv->ports = g_list_delete_link (priv->ports, walk);
       break;
@@ -552,7 +552,6 @@ create_best_element (GstCaps *caps)
 
 static PinosPort *
 add_port (PinosNode       *node,
-          PinosDirection   direction,
           guint            id,
           GError         **error)
 {
@@ -577,9 +576,8 @@ add_port (PinosNode       *node,
 
   data = g_slice_new0 (SourcePortData);
   data->source = source;
-  data->id = id;
   data->port = PINOS_NODE_CLASS (pinos_gst_source_parent_class)
-                ->add_port (node, direction, id, error);
+                ->add_port (node, id, error);
 
   g_debug ("connecting signals");
   g_signal_connect (data->port, "activate", (GCallback) on_activate, data);

pinos/modules/spa/spa-alsa-sink.c

@@ -38,7 +38,6 @@
 typedef struct {
   PinosSpaAlsaSink *sink;
 
-  guint id;
   PinosPort *port;
 } SinkPortData;
 
@@ -166,6 +165,7 @@ on_sink_event (SpaNode *node, SpaEvent *event, void *user_data)
       PinosRingbufferArea areas[2];
       uint8_t *data;
       size_t size, towrite, total;
+      SpaEventNeedInput *ni = event->data;
 
       size = 0;
       data = NULL;
@@ -186,7 +186,7 @@ on_sink_event (SpaNode *node, SpaEvent *event, void *user_data)
 
       pinos_ringbuffer_read_advance (priv->ringbuffer, total);
 
-      iinfo.port_id = event->port_id;
+      iinfo.port_id = ni->port_id;
       iinfo.flags = SPA_INPUT_FLAG_NONE;
       iinfo.buffer_id = 0;
 
@@ -417,7 +417,6 @@ on_received_event (PinosPort   *port,
 
 static PinosPort *
 add_port (PinosNode       *node,
-          PinosDirection   direction,
           guint            id,
           GError         **error)
 {
@@ -427,9 +426,8 @@ add_port (PinosNode       *node,
 
   data = g_slice_new0 (SinkPortData);
   data->sink = sink;
-  data->id = id;
   data->port = PINOS_NODE_CLASS (pinos_spa_alsa_sink_parent_class)
-                ->add_port (node, direction, id, error);
+                ->add_port (node, id, error);
 
   pinos_port_set_received_cb (data->port, on_received_buffer, on_received_event, sink, NULL);
 
@@ -444,7 +442,7 @@ add_port (PinosNode       *node,
 
 static gboolean
 remove_port (PinosNode       *node,
-             guint            id)
+             PinosPort       *port)
 {
   PinosSpaAlsaSink *sink = PINOS_SPA_ALSA_SINK (node);
   PinosSpaAlsaSinkPrivate *priv = sink->priv;
@@ -453,7 +451,7 @@ remove_port (PinosNode       *node,
   for (walk = priv->ports; walk; walk = g_list_next (walk)) {
     SinkPortData *data = walk->data;
 
-    if (data->id == id) {
+    if (data->port == port) {
       free_sink_port_data (data);
       priv->ports = g_list_delete_link (priv->ports, walk);
       break;

pinos/modules/spa/spa-v4l2-source.c

@@ -332,10 +332,10 @@ on_deactivate (PinosPort *port, gpointer user_data)
 
 static gboolean
 remove_port (PinosNode       *node,
-             guint            id)
+             PinosPort       *port)
 {
   return PINOS_NODE_CLASS (pinos_spa_v4l2_source_parent_class)
-                    ->remove_port (node, id);
+                    ->remove_port (node, port);
 }
 
 static void
@@ -377,7 +377,7 @@ on_received_event (PinosPort *port, SpaEvent *event, GError **error, gpointer us
       SpaEventReuseBuffer *rb = event->data;
 
       if ((res = spa_node_port_reuse_buffer (node->node,
-                                             event->port_id,
+                                             rb->port_id,
                                              rb->buffer_id)) < 0)
         g_warning ("client-node %p: error reuse buffer: %d", node, res);
       break;
@@ -393,7 +393,6 @@ on_received_event (PinosPort *port, SpaEvent *event, GError **error, gpointer us
 
 static PinosPort *
 add_port (PinosNode       *node,
-          PinosDirection   direction,
           guint            id,
           GError         **error)
 {
@@ -401,7 +400,7 @@ add_port (PinosNode       *node,
   PinosPort *port;
 
   port = PINOS_NODE_CLASS (pinos_spa_v4l2_source_parent_class)
-                    ->add_port (node, direction, id, error);
+                    ->add_port (node, id, error);
 
   pinos_port_set_received_cb (port, on_received_buffer, on_received_event, node, NULL);
 

pinos/server/client-node.c

@@ -273,8 +273,7 @@ on_node_event (SpaNode *node, SpaEvent *event, void *user_data)
       GError *error = NULL;
 
       port = PINOS_NODE_CLASS (pinos_client_node_parent_class)->add_port (pnode,
-                                                                          pa->direction,
+                                                                          pa->port_id,
-                                                                          event->port_id,
                                                                           &error);
 
       if (port == NULL) {
@@ -339,7 +338,7 @@ on_node_event (SpaNode *node, SpaEvent *event, void *user_data)
       if ((res = spa_node_port_pull_output (node, 1, info)) < 0)
         g_debug ("client-node %p: got pull error %d, %d", this, res, info[0].status);
 
-      port = pinos_node_find_port (PINOS_NODE (this), info[0].port_id);
+      port = pinos_node_find_port_by_id (PINOS_NODE (this), info[0].port_id);
 
       if (!pinos_port_send_buffer (port, info[0].buffer_id, &error)) {
         g_debug ("send failed: %s", error->message);
@@ -351,8 +350,9 @@ on_node_event (SpaNode *node, SpaEvent *event, void *user_data)
     {
       PinosPort *port;
       GError *error = NULL;
+      SpaEventReuseBuffer *rb = event->data;
 
-      port = pinos_node_find_port (PINOS_NODE (this), event->port_id);
+      port = pinos_node_find_port_by_id (PINOS_NODE (this), rb->port_id);
       pinos_port_send_event (port, event, &error);
       break;
     }
@@ -374,16 +374,15 @@ setup_node (PinosClientNode *this)
 
 static PinosPort *
 add_port (PinosNode       *node,
-          PinosDirection   direction,
           guint            id,
           GError         **error)
 {
   PinosPort *port;
 
-  if (spa_node_add_port (node->node, direction, id) < 0)
+  if (spa_node_add_port (node->node, id) < 0)
     g_warning ("client-node %p: error adding port", node);
 
-  port = PINOS_NODE_CLASS (pinos_client_node_parent_class)->add_port (node, direction, id, error);
+  port = PINOS_NODE_CLASS (pinos_client_node_parent_class)->add_port (node, id, error);
 
   if (port) {
     pinos_port_set_received_cb (port, on_received_buffer, on_received_event, node, NULL);
@@ -393,12 +392,12 @@ add_port (PinosNode       *node,
 
 static gboolean
 remove_port (PinosNode       *node,
-             guint            id)
+             PinosPort       *port)
 {
-  if (spa_node_remove_port (node->node, id) < 0)
+  if (spa_node_remove_port (node->node, port->id) < 0)
     g_warning ("client-node %p: error removing port", node);
 
-  return PINOS_NODE_CLASS (pinos_client_node_parent_class)->remove_port (node, id);
+  return PINOS_NODE_CLASS (pinos_client_node_parent_class)->remove_port (node, port);
 }
 
 static void

pinos/server/daemon.c

@@ -573,11 +573,21 @@ pinos_daemon_find_port (PinosDaemon     *daemon,
       }
     }
     if (best == NULL && node_found) {
-      g_debug ("node %p: making port", n);
+      guint id;
-      best = pinos_node_add_port (n, direction, 0, NULL);
+
-      if (best != NULL) {
+      id = pinos_node_get_free_port_id (n, direction);
-        created_port = TRUE;
+      if (id != SPA_ID_INVALID) {
-        break;
+        g_debug ("node %p: making port with id %u", n, id);
+        best = pinos_node_add_port (n, id, NULL);
+        if (best != NULL) {
+          created_port = TRUE;
+          break;
+        }
+      } else {
+        g_debug ("node %p: using port with id %u", n, id);
+        best = pinos_node_find_port_by_id (n, id);
+        if (best != NULL)
+          break;
       }
     }
   }

pinos/server/node.c

@@ -43,6 +43,13 @@ struct _PinosNodePrivate
   gchar *object_path;
   gchar *name;
 
+  unsigned int max_input_ports;
+  unsigned int max_output_ports;
+  unsigned int n_input_ports;
+  unsigned int n_output_ports;
+  uint32_t *input_port_ids;
+  uint32_t *output_port_ids;
+
   PinosNodeState state;
   GError *error;
   guint idle_timeout;
@@ -87,17 +94,53 @@ node_set_state (PinosNode       *node,
 static void
 do_remove_port (PinosPort *port, PinosNode *node)
 {
-  pinos_node_remove_port (node, port->id);
+  pinos_node_remove_port (node, port);
+}
+
+static void
+update_port_ids (PinosNode *node, gboolean create)
+{
+  PinosNodePrivate *priv = node->priv;
+  guint i;
+
+  if (node->node == NULL)
+    return;
+
+  spa_node_get_n_ports (node->node,
+                        &priv->n_input_ports,
+                        &priv->max_input_ports,
+                        &priv->n_output_ports,
+                        &priv->max_output_ports);
+
+  priv->input_port_ids = g_realloc_n (priv->input_port_ids, priv->max_input_ports, sizeof (uint32_t));
+  priv->output_port_ids = g_realloc_n (priv->output_port_ids, priv->max_output_ports, sizeof (uint32_t));
+
+  spa_node_get_port_ids (node->node,
+                         priv->max_input_ports,
+                         priv->input_port_ids,
+                         priv->max_output_ports,
+                         priv->output_port_ids);
+
+  if (create) {
+    for (i = 0; i < priv->n_input_ports; i++)
+      pinos_node_add_port (node, priv->input_port_ids[i], NULL);
+    for (i = 0; i < priv->n_output_ports; i++)
+      pinos_node_add_port (node, priv->output_port_ids[i], NULL);
+  }
 }
 
 static PinosPort *
 node_add_port (PinosNode       *node,
-               PinosDirection   direction,
                guint            id,
                GError         **error)
 {
   PinosNodePrivate *priv = node->priv;
   PinosPort *port;
+  PinosDirection direction;
+
+  update_port_ids (node, FALSE);
+
+  direction = id < priv->max_input_ports ? PINOS_DIRECTION_INPUT : PINOS_DIRECTION_OUTPUT;
 
   port = g_object_new (PINOS_TYPE_PORT,
                        "daemon", priv->daemon,
@@ -115,17 +158,16 @@ node_add_port (PinosNode       *node,
 
 static gboolean
 node_remove_port (PinosNode       *node,
-                  guint            id)
+                  PinosPort       *port)
 {
   PinosNodePrivate *priv = node->priv;
-  PinosPort *port;
 
-  g_debug ("node %p: removed port %u", node, id);
+  g_debug ("node %p: removed port %u", node, port->id);
-  port = g_hash_table_lookup (priv->ports, GUINT_TO_POINTER (id));
+  g_object_ref (port);
-  if (port) {
+  if (g_hash_table_remove (priv->ports, GUINT_TO_POINTER (port->id)))
     g_signal_emit (node, signals[SIGNAL_PORT_REMOVED], 0, port);
-    g_hash_table_remove (priv->ports, GUINT_TO_POINTER (id));
+  g_object_unref (port);
-  }
+
   return TRUE;
 }
 
@@ -146,7 +188,7 @@ handle_add_port (PinosNode1             *interface,
   if (g_strcmp0 (priv->sender, sender) != 0)
     goto not_allowed;
 
-  port = pinos_node_add_port (node, arg_direction, arg_id, &error);
+  port = pinos_node_add_port (node, arg_id, &error);
   if (port == NULL)
     goto no_port;
 
@@ -182,12 +224,17 @@ handle_remove_port (PinosNode1             *interface,
   PinosNode *node = user_data;
   PinosNodePrivate *priv = node->priv;
   const gchar *sender;
+  PinosPort *port;
 
   sender = g_dbus_method_invocation_get_sender (invocation);
   if (g_strcmp0 (priv->sender, sender) != 0)
     goto not_allowed;
 
-  if (!pinos_node_remove_port (node, arg_id))
+  port = pinos_node_find_port_by_id (node, arg_id);
+  if (port == NULL)
+    goto no_port;
+
+  if (!pinos_node_remove_port (node, port))
     goto no_port;
 
   g_debug ("node %p: remove port %u", node, arg_id);
@@ -377,6 +424,8 @@ pinos_node_constructed (GObject * obj)
   g_signal_connect (node, "notify", (GCallback) on_property_notify, node);
   G_OBJECT_CLASS (pinos_node_parent_class)->constructed (obj);
 
+  update_port_ids (node, TRUE);
+
   if (priv->sender == NULL) {
     priv->sender = g_strdup (pinos_daemon_get_sender (priv->daemon));
   }
@@ -415,6 +464,8 @@ pinos_node_finalize (GObject * obj)
   g_clear_error (&priv->error);
   if (priv->properties)
     pinos_properties_free (priv->properties);
+  g_free (priv->input_port_ids);
+  g_free (priv->output_port_ids);
 
   G_OBJECT_CLASS (pinos_node_parent_class)->finalize (obj);
 }
@@ -743,7 +794,6 @@ pinos_node_remove (PinosNode *node)
  */
 PinosPort *
 pinos_node_add_port (PinosNode       *node,
-                     PinosDirection   direction,
                      guint            id,
                      GError         **error)
 {
@@ -759,7 +809,7 @@ pinos_node_add_port (PinosNode       *node,
   }
 
   g_debug ("node %p: add port", node);
-  port = klass->add_port (node, direction, id, error);
+  port = klass->add_port (node, id, error);
 
   return port;
 }
@@ -767,31 +817,75 @@ pinos_node_add_port (PinosNode       *node,
 /**
  * pinos_node_remove_port:
  * @node: a #PinosNode
- * @id: a #PinosPort id
+ * @port: a #PinosPort
  *
- * Remove the #PinosPort with @id from @node
+ * Remove @port from @node
  *
  * Returns: %TRUE when the port was removed
  */
 gboolean
-pinos_node_remove_port (PinosNode *node, guint id)
+pinos_node_remove_port (PinosNode *node, PinosPort *port)
 {
   PinosNodeClass *klass;
   gboolean res = FALSE;
 
   g_return_val_if_fail (PINOS_IS_NODE (node), FALSE);
+  g_return_val_if_fail (PINOS_IS_PORT (port), FALSE);
 
   klass = PINOS_NODE_GET_CLASS (node);
 
   if (!klass->remove_port)
     return FALSE;
 
-  res = klass->remove_port (node, id);
+  res = klass->remove_port (node, port);
   return res;
 }
 
 /**
- * pinos_node_find_port:
+ * pinos_node_get_free_port_id:
+ * @node: a #PinosNode
+ * @direction: a #PinosDirection
+ *
+ * Find a new unused port id in @node with @direction
+ *
+ * Returns: the new port id of %SPA_INVALID_ID on error
+ */
+guint
+pinos_node_get_free_port_id (PinosNode       *node,
+                             PinosDirection   direction)
+{
+  PinosNodePrivate *priv;
+  guint i, free_port = 0, n_ports, max_ports;
+  uint32_t *ports;
+
+  g_return_val_if_fail (PINOS_IS_NODE (node), -1);
+  priv = node->priv;
+
+  if (direction == PINOS_DIRECTION_INPUT) {
+    max_ports = priv->max_input_ports;
+    n_ports = priv->n_input_ports;
+    ports = priv->input_port_ids;
+  } else {
+    max_ports = priv->max_output_ports;
+    n_ports = priv->n_output_ports;
+    ports = priv->output_port_ids;
+  }
+
+  g_debug ("direction %d max %u, n %u\n", direction, max_ports, n_ports);
+
+  for (i = 0; i < n_ports; i++) {
+    if (free_port < ports[i])
+      break;
+    free_port = ports[i] + 1;
+  }
+  if (free_port >= max_ports)
+    return -1;
+
+  return free_port;
+}
+
+/**
+ * pinos_node_find_port_by_id:
  * @node: a #PinosNode
  * @id: a #PinosPort id
  *
@@ -800,7 +894,7 @@ pinos_node_remove_port (PinosNode *node, guint id)
  * Returns: a #PinosPort with @id or %NULL when not found
  */
 PinosPort *
-pinos_node_find_port (PinosNode *node, guint id)
+pinos_node_find_port_by_id (PinosNode *node, guint id)
 {
   PinosNodePrivate *priv;
 
@@ -989,5 +1083,9 @@ pinos_node_update_node_state (PinosNode    *node,
     g_debug ("node %p: update SPA state to %d", node, state);
     node->node_state = state;
     g_object_notify (G_OBJECT (node), "node-state");
+
+    if (state == SPA_NODE_STATE_CONFIGURE) {
+      update_port_ids (node, FALSE);
+    }
   }
 }

pinos/server/node.h

@@ -70,11 +70,10 @@ struct _PinosNodeClass {
                                PinosNodeState   state);
 
   PinosPort * (*add_port)     (PinosNode       *node,
-                               PinosDirection   direction,
                                guint            id,
                                GError         **error);
   gboolean    (*remove_port)  (PinosNode       *node,
-                               guint            id);
+                               PinosPort       *port);
 };
 
 /* normal GObject stuff */
@@ -94,13 +93,14 @@ PinosDaemon *       pinos_node_get_daemon              (PinosNode       *node);
 const gchar *       pinos_node_get_sender              (PinosNode       *node);
 const gchar *       pinos_node_get_object_path         (PinosNode       *node);
 
+guint               pinos_node_get_free_port_id        (PinosNode       *node,
+                                                        PinosDirection   direction);
 PinosPort *         pinos_node_add_port                (PinosNode       *node,
-                                                        PinosDirection   direction,
                                                         guint            id,
                                                         GError         **error);
 gboolean            pinos_node_remove_port             (PinosNode       *node,
-                                                        guint            id);
+                                                        PinosPort       *port);
-PinosPort *         pinos_node_find_port               (PinosNode       *node,
+PinosPort *         pinos_node_find_port_by_id         (PinosNode       *node,
                                                         guint            id);
 GList *             pinos_node_get_ports               (PinosNode       *node);
 

spa/include/spa/control.h

@@ -103,17 +103,15 @@ typedef struct {
 typedef struct {
   uint32_t           port_id;
   uint32_t           change_mask;
-  SpaDirection       direction;
   unsigned int       n_possible_formats;
   SpaFormat        **possible_formats;
   const SpaProps    *props;
   const SpaPortInfo *info;
 } SpaControlCmdPortUpdate;
 
-#define SPA_CONTROL_CMD_PORT_UPDATE_DIRECTION         (1 << 0)
+#define SPA_CONTROL_CMD_PORT_UPDATE_POSSIBLE_FORMATS  (1 << 0)
-#define SPA_CONTROL_CMD_PORT_UPDATE_POSSIBLE_FORMATS  (1 << 1)
+#define SPA_CONTROL_CMD_PORT_UPDATE_PROPS             (1 << 1)
-#define SPA_CONTROL_CMD_PORT_UPDATE_PROPS             (1 << 2)
+#define SPA_CONTROL_CMD_PORT_UPDATE_INFO              (1 << 2)
-#define SPA_CONTROL_CMD_PORT_UPDATE_INFO              (1 << 3)
 
 /* SPA_CONTROL_CMD_PORT_REMOVED */
 typedef struct {
@@ -141,7 +139,6 @@ typedef struct {
 /* SPA_CONTROL_CMD_ADD_PORT */
 typedef struct {
   uint32_t     port_id;
-  SpaDirection direction;
 } SpaControlCmdAddPort;
 
 /* SPA_CONTROL_CMD_REMOVE_PORT */

spa/include/spa/defs.h

@@ -61,12 +61,6 @@ typedef enum {
   SPA_RESULT_WRONG_STATE               = -29,
 } SpaResult;
 
-typedef enum {
-  SPA_DIRECTION_INVALID         = 0,
-  SPA_DIRECTION_INPUT,
-  SPA_DIRECTION_OUTPUT
-} SpaDirection;
-
 typedef void (*SpaNotify) (void *data);
 
 #define SPA_N_ELEMENTS(arr)  (sizeof (arr) / sizeof ((arr)[0]))

spa/include/spa/event.h

@@ -66,20 +66,32 @@ typedef enum {
 
 struct _SpaEvent {
   SpaEventType   type;
-  uint32_t       port_id;
   void          *data;
   size_t         size;
 };
 
 typedef struct {
-  SpaDirection direction;
+  uint32_t     port_id;
 } SpaEventPortAdded;
 
+typedef struct {
+  uint32_t     port_id;
+} SpaEventPortRemoved;
+
 typedef struct {
   SpaNodeState state;
 } SpaEventStateChange;
 
 typedef struct {
+  uint32_t     port_id;
+} SpaEventHaveOutput;
+
+typedef struct {
+  uint32_t     port_id;
+} SpaEventNeedInput;
+
+typedef struct {
+  uint32_t port_id;
   uint32_t buffer_id;
 } SpaEventReuseBuffer;
 

spa/include/spa/node.h

@@ -267,8 +267,22 @@ struct _SpaNode {
                                        unsigned int      n_output_ports,
                                        uint32_t         *output_ids);
 
+  /**
+   * SpaNode::add_port:
+   * @node: a #SpaNode
+   * @port_id: an unused port id
+   *
+   * Make a new port with @port_id. The called should use get_port_ids() to
+   * find an unused id.
+   *
+   * Input port ids should be between 0 and max_input_ports and output ports
+   * between max_input_ports and max_input_ports + max_output_ports as obtained
+   * from get_port_ids().
+   *
+   * Returns: #SPA_RESULT_OK on success
+   *          #SPA_RESULT_INVALID_ARGUMENTS when node is %NULL
+   */
   SpaResult   (*add_port)             (SpaNode          *node,
-                                       SpaDirection      direction,
                                        uint32_t          port_id);
   SpaResult   (*remove_port)          (SpaNode          *node,
                                        uint32_t          port_id);

spa/plugins/alsa/alsa-sink.c

@@ -225,7 +225,6 @@ spa_alsa_sink_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_STREAMING;
@@ -241,7 +240,6 @@ spa_alsa_sink_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_PAUSED;
@@ -316,7 +314,6 @@ spa_alsa_sink_node_get_port_ids (SpaNode       *node,
 
 static SpaResult
 spa_alsa_sink_node_add_port (SpaNode        *node,
-                             SpaDirection    direction,
                              uint32_t        port_id)
 {
   return SPA_RESULT_NOT_IMPLEMENTED;

spa/plugins/alsa/alsa-utils.c

@@ -225,11 +225,12 @@ static void
 pull_input (SpaALSASink *this, void *data, snd_pcm_uframes_t frames)
 {
   SpaEvent event;
+  SpaEventNeedInput ni;
 
   event.type = SPA_EVENT_TYPE_NEED_INPUT;
-  event.port_id = 0;
+  event.size = sizeof (ni);
-  event.size = 0;
+  event.data = ∋
-  event.data = NULL;
+  ni.port_id = 0;
   this->event_cb (&this->node, &event, this->user_data);
 }
 
@@ -344,7 +345,6 @@ spa_alsa_start (SpaALSASink *this)
   }
 
   event.type = SPA_EVENT_TYPE_ADD_POLL;
-  event.port_id = 0;
   event.data = &state->poll;
   event.size = sizeof (state->poll);
 
@@ -373,7 +373,6 @@ spa_alsa_stop (SpaALSASink *this)
   snd_pcm_drop (state->handle);
 
   event.type = SPA_EVENT_TYPE_REMOVE_POLL;
-  event.port_id = 0;
   event.data = &state->poll;
   event.size = sizeof (state->poll);
   this->event_cb (&this->node, &event, this->user_data);

spa/plugins/audiomixer/audiomixer.c

@@ -152,7 +152,6 @@ spa_audiomixer_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_STREAMING;
@@ -167,7 +166,6 @@ spa_audiomixer_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_PAUSED;
@@ -253,7 +251,6 @@ spa_audiomixer_node_get_port_ids (SpaNode       *node,
 
 static SpaResult
 spa_audiomixer_node_add_port (SpaNode        *node,
-                              SpaDirection    direction,
                               uint32_t        port_id)
 {
   SpaAudioMixer *this;
@@ -263,9 +260,6 @@ spa_audiomixer_node_add_port (SpaNode        *node,
 
   this = (SpaAudioMixer *) node->handle;
 
-  if (direction != SPA_DIRECTION_INPUT)
-    return SPA_RESULT_INVALID_DIRECTION;
-
   if (port_id >= MAX_PORTS)
     return SPA_RESULT_INVALID_PORT;
 
@@ -571,11 +565,12 @@ static void
 pull_port (SpaAudioMixer *this, uint32_t port_id, SpaOutputInfo *info, size_t pull_size)
 {
   SpaEvent event;
+  SpaEventNeedInput ni;
 
   event.type = SPA_EVENT_TYPE_NEED_INPUT;
-  event.port_id = port_id;
+  event.size = sizeof (ni);
-  event.size = 0;
+  event.data = ∋
-  event.data = NULL;
+  ni.port_id = port_id;
   this->event_cb (&this->node, &event, this->user_data);
 }
 

spa/plugins/audiotestsrc/audiotestsrc.c

@@ -173,7 +173,6 @@ spa_audiotestsrc_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_STREAMING;
@@ -188,7 +187,6 @@ spa_audiotestsrc_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_PAUSED;
@@ -263,7 +261,6 @@ spa_audiotestsrc_node_get_port_ids (SpaNode       *node,
 
 static SpaResult
 spa_audiotestsrc_node_add_port (SpaNode        *node,
-                                SpaDirection    direction,
                                 uint32_t        port_id)
 {
   return SPA_RESULT_NOT_IMPLEMENTED;

spa/plugins/ffmpeg/ffmpeg-dec.c

@@ -148,7 +148,6 @@ spa_ffmpeg_dec_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_STREAMING;
@@ -162,7 +161,6 @@ spa_ffmpeg_dec_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_PAUSED;
@@ -240,7 +238,6 @@ spa_ffmpeg_dec_node_get_port_ids (SpaNode       *node,
 
 static SpaResult
 spa_ffmpeg_dec_node_add_port (SpaNode        *node,
-                              SpaDirection    direction,
                               uint32_t        port_id)
 {
   return SPA_RESULT_NOT_IMPLEMENTED;

spa/plugins/ffmpeg/ffmpeg-enc.c

@@ -148,7 +148,6 @@ spa_ffmpeg_enc_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_STREAMING;
@@ -162,7 +161,6 @@ spa_ffmpeg_enc_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_PAUSED;
@@ -240,7 +238,6 @@ spa_ffmpeg_enc_node_get_port_ids (SpaNode       *node,
 
 static SpaResult
 spa_ffmpeg_enc_node_add_port (SpaNode        *node,
-                              SpaDirection    direction,
                               uint32_t        port_id)
 {
   return SPA_RESULT_NOT_IMPLEMENTED;

spa/plugins/remote/proxy.c

@@ -39,7 +39,8 @@
 
 #define CHECK_FREE_PORT_ID(this,id)     ((id) < MAX_PORTS && !(this)->ports[id].valid)
 #define CHECK_PORT_ID(this,id)          ((id) < MAX_PORTS && (this)->ports[id].valid)
-#define CHECK_PORT_ID_DIR(this,id,dir)  (CHECK_PORT_ID(this,id) && (this)->ports[i].direction == (dir))
+#define CHECK_PORT_ID_IN(this,id)       (CHECK_PORT_ID(this,id) && (id < this->max_inputs))
+#define CHECK_PORT_ID_OUT(this,id)      (CHECK_PORT_ID(this,id) && (id >= this->max_inputs))
 
 typedef struct _SpaProxy SpaProxy;
 
@@ -50,7 +51,6 @@ typedef struct {
 
 typedef struct {
   bool           valid;
-  SpaDirection   direction;
   SpaPortInfo    info;
   SpaFormat     *format;
   unsigned int   n_formats;
@@ -112,7 +112,6 @@ update_poll (SpaProxy *this, int socketfd)
 
   if (p->socketfd != -1) {
     event.type = SPA_EVENT_TYPE_REMOVE_POLL;
-    event.port_id = 0;
     event.data = &this->poll;
     event.size = sizeof (this->poll);
     this->event_cb (&this->node, &event, this->user_data);
@@ -122,7 +121,6 @@ update_poll (SpaProxy *this, int socketfd)
   if (p->socketfd != -1) {
     this->fds[0].fd = p->socketfd;
     event.type = SPA_EVENT_TYPE_ADD_POLL;
-    event.port_id = 0;
     event.data = &this->poll;
     event.size = sizeof (this->poll);
     this->event_cb (&this->node, &event, this->user_data);
@@ -139,7 +137,6 @@ update_state (SpaProxy *this, SpaNodeState state)
     this->state = state;
 
     event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-    event.port_id = -1;
     event.data = &sc;
     event.size = sizeof (sc);
     sc.state = state;
@@ -318,17 +315,15 @@ spa_proxy_node_get_port_ids (SpaNode       *node,
   this = (SpaProxy *) node->handle;
 
   if (input_ids) {
-    n_input_ports = SPA_MIN (n_input_ports, MAX_PORTS);
+    for (c = 0, i = 0; i < MAX_INPUTS && c < n_input_ports; i++) {
-    for (c = 0, i = 0; i < n_input_ports; i++) {
+      if (this->ports[i].valid)
-      if (this->ports[i].valid && this->ports[i].direction == SPA_DIRECTION_INPUT)
         input_ids[c++] = i;
     }
   }
   if (output_ids) {
-    n_output_ports = SPA_MIN (n_output_ports, MAX_PORTS);
+    for (c = 0, i = 0; i < MAX_OUTPUTS && c < n_output_ports; i++) {
-    for (c = 0, i = 0; i < n_output_ports; i++) {
+      if (this->ports[MAX_INPUTS + i].valid)
-      if (this->ports[i].valid && this->ports[i].direction == SPA_DIRECTION_OUTPUT)
+        output_ids[c++] = MAX_INPUTS + i;
-        output_ids[c++] = i;
     }
   }
   return SPA_RESULT_OK;
@@ -361,21 +356,19 @@ do_update_port (SpaProxy                *this,
   }
 
   if (!port->valid) {
-    fprintf (stderr, "%p: adding port %d, %d\n", this, pu->port_id, pu->direction);
+    fprintf (stderr, "%p: adding port %d\n", this, pu->port_id);
-    port->direction = pu->direction;
     port->format = NULL;
     port->valid = true;
 
-    if (pu->direction == SPA_DIRECTION_INPUT)
+    if (pu->port_id < MAX_INPUTS)
       this->n_inputs++;
     else
       this->n_outputs++;
 
     event.type = SPA_EVENT_TYPE_PORT_ADDED;
-    event.port_id = pu->port_id;
-    event.data = &pa;
     event.size = sizeof (pa);
-    pa.direction = pu->direction;
+    event.data = &pa;
+    pa.port_id = pu->port_id;
     this->event_cb (&this->node, &event, this->user_data);
   }
 }
@@ -386,31 +379,30 @@ do_uninit_port (SpaProxy     *this,
 {
   SpaEvent event;
   SpaProxyPort *port;
+  SpaEventPortRemoved pr;
 
   fprintf (stderr, "%p: removing port %d\n", this, port_id);
   port = &this->ports[port_id];
 
-  if (port->direction == SPA_DIRECTION_INPUT)
+  if (port_id < MAX_INPUTS)
     this->n_inputs--;
   else
     this->n_outputs--;
 
-  port->direction = SPA_DIRECTION_INVALID;
   port->valid = false;
   if (port->format)
     spa_format_unref (port->format);
   port->format = NULL;
 
   event.type = SPA_EVENT_TYPE_PORT_REMOVED;
-  event.port_id = port_id;
+  event.size = sizeof (pr);
-  event.data = NULL;
+  event.data = &pr;
-  event.size = 0;
+  pr.port_id = port_id;
   this->event_cb (&this->node, &event, this->user_data);
 }
 
 static SpaResult
 spa_proxy_node_add_port (SpaNode        *node,
-                         SpaDirection    direction,
                          uint32_t        port_id)
 {
   SpaProxy *this;
@@ -424,12 +416,10 @@ spa_proxy_node_add_port (SpaNode        *node,
   if (!CHECK_FREE_PORT_ID (this, port_id))
     return SPA_RESULT_INVALID_PORT;
 
-  pu.change_mask = SPA_CONTROL_CMD_PORT_UPDATE_DIRECTION |
+  pu.change_mask = SPA_CONTROL_CMD_PORT_UPDATE_POSSIBLE_FORMATS |
-                   SPA_CONTROL_CMD_PORT_UPDATE_POSSIBLE_FORMATS |
                    SPA_CONTROL_CMD_PORT_UPDATE_PROPS |
                    SPA_CONTROL_CMD_PORT_UPDATE_INFO;
   pu.port_id = port_id;
-  pu.direction = direction;
   pu.n_possible_formats = 0;
   pu.possible_formats = NULL;
   pu.props = NULL;
@@ -793,7 +783,7 @@ spa_proxy_node_port_push_input (SpaNode        *node,
   spa_control_builder_init_into (&builder, buf, sizeof(buf), NULL, 0);
 
   for (i = 0; i < n_info; i++) {
-    if (!CHECK_PORT_ID_DIR (this, info[i].port_id, SPA_DIRECTION_INPUT)) {
+    if (!CHECK_PORT_ID_IN (this, info[i].port_id)) {
       info[i].status = SPA_RESULT_INVALID_PORT;
       have_error = true;
       continue;
@@ -852,7 +842,7 @@ spa_proxy_node_port_pull_output (SpaNode        *node,
   this = (SpaProxy *) node->handle;
 
   for (i = 0; i < n_info; i++) {
-    if (!CHECK_PORT_ID_DIR (this, info[i].port_id, SPA_DIRECTION_OUTPUT)) {
+    if (!CHECK_PORT_ID_OUT (this, info[i].port_id)) {
       info[i].status = SPA_RESULT_INVALID_PORT;
       have_error = true;
       continue;
@@ -901,7 +891,7 @@ spa_proxy_node_port_push_event (SpaNode        *node,
 
       /* send start */
       spa_control_builder_init_into (&builder, buf, sizeof (buf), NULL, 0);
-      crb.port_id = event->port_id;
+      crb.port_id = rb->port_id;
       crb.buffer_id = rb->buffer_id;
       spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_REUSE_BUFFER, &crb);
       spa_control_builder_end (&builder, &control);
@@ -946,12 +936,13 @@ parse_control (SpaProxy   *this,
       {
         SpaControlCmdNodeUpdate nu;
 
-        fprintf (stderr, "proxy %p: got node update %d\n", this, cmd);
         if (spa_control_iter_parse_cmd (&it, &nu) < 0)
           break;
 
         this->max_inputs = nu.max_input_ports;
         this->max_outputs = nu.max_output_ports;
+        fprintf (stderr, "proxy %p: got node update %d, %u, %u\n", this, cmd,
+            this->max_inputs, this->max_outputs);
         break;
       }
 
@@ -967,8 +958,7 @@ parse_control (SpaProxy   *this,
         if (pu.port_id >= MAX_PORTS)
           break;
 
-        remove = (pu.change_mask & SPA_CONTROL_CMD_PORT_UPDATE_DIRECTION) &&
+        remove = (pu.change_mask == 0);
-                 (pu.direction == SPA_DIRECTION_INVALID);
 
         if (remove) {
           do_uninit_port (this, pu.port_id);
@@ -1001,15 +991,16 @@ parse_control (SpaProxy   *this,
       case SPA_CONTROL_CMD_HAVE_OUTPUT:
       {
         SpaEvent event;
+        SpaEventHaveOutput hu;
         SpaControlCmdHaveOutput cmd;
 
         if (spa_control_iter_parse_cmd (&it, &cmd) < 0)
           break;
 
         event.type = SPA_EVENT_TYPE_HAVE_OUTPUT;
-        event.port_id = cmd.port_id;
+        event.data = &hu;
-        event.data = NULL;
+        event.size = sizeof (hu);
-        event.size = 0;
+        hu.port_id = cmd.port_id;
         this->event_cb (&this->node, &event, this->user_data);
         break;
       }
@@ -1043,9 +1034,9 @@ parse_control (SpaProxy   *this,
           break;
 
         event.type = SPA_EVENT_TYPE_REUSE_BUFFER;
-        event.port_id = crb.port_id;
         event.data = &rb;
         event.size = sizeof (rb);
+        rb.port_id = crb.port_id;
         rb.buffer_id = crb.buffer_id;
         this->event_cb (&this->node, &event, this->user_data);
 

spa/plugins/v4l2/v4l2-source.c

@@ -213,7 +213,6 @@ update_state (SpaV4l2Source *this, SpaNodeState state)
   this->node_state = state;
 
   event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-  event.port_id = -1;
   event.data = ≻
   event.size = sizeof (sc);
   sc.state = state;
@@ -309,10 +308,10 @@ spa_v4l2_source_node_get_n_ports (SpaNode       *node,
 
   if (n_input_ports)
     *n_input_ports = 0;
-  if (n_output_ports)
-    *n_output_ports = 1;
   if (max_input_ports)
     *max_input_ports = 0;
+  if (n_output_ports)
+    *n_output_ports = 1;
   if (max_output_ports)
     *max_output_ports = 1;
 
@@ -338,7 +337,6 @@ spa_v4l2_source_node_get_port_ids (SpaNode       *node,
 
 static SpaResult
 spa_v4l2_source_node_add_port (SpaNode        *node,
-                               SpaDirection    direction,
                                uint32_t        port_id)
 {
   return SPA_RESULT_NOT_IMPLEMENTED;
@@ -383,6 +381,8 @@ spa_v4l2_source_node_port_enum_formats (SpaNode         *node,
 
   this = (SpaV4l2Source *) node->handle;
 
+  fprintf (stderr, "%d\n", port_id);
+
   if (port_id != 0)
     return SPA_RESULT_INVALID_PORT;
 

spa/plugins/v4l2/v4l2-utils.c

@@ -583,14 +583,15 @@ v4l2_on_fd_events (SpaPollNotifyData *data)
 {
   SpaV4l2Source *this = data->user_data;
   SpaEvent event;
+  SpaEventHaveOutput ho;
 
   if (mmap_read (this) < 0)
     return 0;
 
   event.type = SPA_EVENT_TYPE_HAVE_OUTPUT;
-  event.port_id = 0;
+  event.size = sizeof (ho);
-  event.size = 0;
+  event.data = &ho;
-  event.data = NULL;
+  ho.port_id = 0;
   this->event_cb (&this->node, &event, this->user_data);
 
   return 0;
@@ -856,7 +857,6 @@ spa_v4l2_start (SpaV4l2Source *this)
   }
 
   event.type = SPA_EVENT_TYPE_ADD_POLL;
-  event.port_id = 0;
   event.data = &state->poll;
   event.size = sizeof (state->poll);
 
@@ -884,7 +884,6 @@ spa_v4l2_pause (SpaV4l2Source *this)
   SpaEvent event;
 
   event.type = SPA_EVENT_TYPE_REMOVE_POLL;
-  event.port_id = 0;
   event.data = &state->poll;
   event.size = sizeof (state->poll);
   this->event_cb (&this->node, &event, this->user_data);

spa/plugins/volume/volume.c

@@ -157,7 +157,6 @@ spa_volume_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_STREAMING;
@@ -172,7 +171,6 @@ spa_volume_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_PAUSED;
@@ -250,7 +248,6 @@ spa_volume_node_get_port_ids (SpaNode       *node,
 
 static SpaResult
 spa_volume_node_add_port (SpaNode        *node,
-                          SpaDirection    direction,
                           uint32_t        port_id)
 {
   return SPA_RESULT_NOT_IMPLEMENTED;
@@ -524,9 +521,9 @@ release_buffer (SpaVolume *this, SpaBuffer *buffer)
   SpaEventReuseBuffer rb;
 
   event.type = SPA_EVENT_TYPE_REUSE_BUFFER;
-  event.port_id = 0;
   event.data = &rb;
   event.size = sizeof (rb);
+  rb.port_id = 0;
   rb.buffer_id = buffer->id;
   this->event_cb (&this->node, &event, this->user_data);
 }

spa/plugins/xv/xv-sink.c

@@ -181,7 +181,6 @@ spa_xv_sink_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_STREAMING;
@@ -197,7 +196,6 @@ spa_xv_sink_node_send_command (SpaNode       *node,
         SpaEventStateChange sc;
 
         event.type = SPA_EVENT_TYPE_STATE_CHANGE;
-        event.port_id = -1;
         event.data = ≻
         event.size = sizeof (sc);
         sc.state = SPA_NODE_STATE_PAUSED;
@@ -273,7 +271,6 @@ spa_xv_sink_node_get_port_ids (SpaNode       *node,
 
 static SpaResult
 spa_xv_sink_node_add_port (SpaNode        *node,
-                           SpaDirection    direction,
                            uint32_t        port_id)
 {
   return SPA_RESULT_NOT_IMPLEMENTED;

spa/tests/test-mixer.c

@@ -99,11 +99,12 @@ on_mix_event (SpaNode *node, SpaEvent *event, void *user_data)
       SpaInputInfo iinfo;
       SpaOutputInfo oinfo;
       SpaResult res;
+      SpaEventNeedInput *ni = event->data;
 
       oinfo.port_id = 0;
       oinfo.flags = SPA_OUTPUT_FLAG_NONE;
 
-      if (event->port_id == data->mix_ports[0]) {
+      if (ni->port_id == data->mix_ports[0]) {
         if ((res = spa_node_port_pull_output (data->source1, 1, &oinfo)) < 0)
           printf ("got error %d\n", res);
       } else {
@@ -111,7 +112,7 @@ on_mix_event (SpaNode *node, SpaEvent *event, void *user_data)
           printf ("got error %d\n", res);
       }
 
-      iinfo.port_id = event->port_id;
+      iinfo.port_id = ni->port_id;
       iinfo.flags = SPA_INPUT_FLAG_NONE;
       iinfo.buffer_id = oinfo.buffer_id;
 
@@ -136,6 +137,7 @@ on_sink_event (SpaNode *node, SpaEvent *event, void *user_data)
       SpaInputInfo iinfo;
       SpaOutputInfo oinfo;
       SpaResult res;
+      SpaEventNeedInput *ni = event->data;
 
       oinfo.port_id = 0;
       oinfo.flags = SPA_OUTPUT_FLAG_PULL;
@@ -143,7 +145,7 @@ on_sink_event (SpaNode *node, SpaEvent *event, void *user_data)
       if ((res = spa_node_port_pull_output (data->mix, 1, &oinfo)) < 0)
         printf ("got error %d\n", res);
 
-      iinfo.port_id = event->port_id;
+      iinfo.port_id = ni->port_id;
       iinfo.flags = SPA_INPUT_FLAG_NONE;
       iinfo.buffer_id = oinfo.buffer_id;
 
@@ -251,7 +253,7 @@ negotiate_formats (AppData *data)
     return res;
 
   data->mix_ports[0] = 0;
-  if ((res = spa_node_add_port (data->mix, SPA_DIRECTION_INPUT, 0)) < 0)
+  if ((res = spa_node_add_port (data->mix, 0)) < 0)
     return res;
 
   if ((res = spa_node_port_set_format (data->mix, data->mix_ports[0], false, format)) < 0)
@@ -261,7 +263,7 @@ negotiate_formats (AppData *data)
     return res;
 
   data->mix_ports[1] = 0;
-  if ((res = spa_node_add_port (data->mix, SPA_DIRECTION_INPUT, 1)) < 0)
+  if ((res = spa_node_add_port (data->mix, 1)) < 0)
     return res;
 
   if ((res = spa_node_port_set_format (data->mix, data->mix_ports[1], false, format)) < 0)