Add poll interface and remove poll events

Use an interface to change items in a poll loop.

pinos/client/stream.c

@@ -886,9 +886,6 @@ handle_node_event (PinosStream  *stream,
       }
       break;
     }
-    case SPA_NODE_EVENT_TYPE_ADD_POLL:
-    case SPA_NODE_EVENT_TYPE_UPDATE_POLL:
-    case SPA_NODE_EVENT_TYPE_REMOVE_POLL:
     case SPA_NODE_EVENT_TYPE_DRAINED:
     case SPA_NODE_EVENT_TYPE_MARKER:
     case SPA_NODE_EVENT_TYPE_ERROR:

pinos/server/daemon.c

@@ -57,7 +57,7 @@ struct _PinosDaemonPrivate
 
   GHashTable *node_factories;
 
-  SpaSupport support[2];
+  SpaSupport support[3];
   SpaLog log;
 };
 
@@ -842,11 +842,12 @@ do_logv (SpaLog        *log,
 {
   char text[16*1024], location[128];
   static const char *levels[] = {
-    "NONE",
+    "-",
-    "ERROR",
+    "E",
-    "WARN",
+    "W",
-    "INFO",
+    "I",
-    "TRACE",
+    "D",
+    "T",
   };
   vsnprintf (text, sizeof(text), fmt, args);
   snprintf (location, sizeof(location), "%s:%i %s()", file, line, func);
@@ -899,8 +900,10 @@ pinos_daemon_init (PinosDaemon * daemon)
   priv->support[0].data = daemon->map;
   priv->support[1].uri = SPA_LOG_URI;
   priv->support[1].data = daemon->log;
+  priv->support[2].uri = SPA_POLL__DataLoop;
+  priv->support[2].data = &priv->loop->poll;
   daemon->support = priv->support;
-  daemon->n_support = 2;
+  daemon->n_support = 3;
 }
 
 /**

pinos/server/node.c

@@ -453,24 +453,6 @@ on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
       break;
     }
 
-    case SPA_NODE_EVENT_TYPE_ADD_POLL:
-    {
-      SpaPollItem *item = event->data;
-      pinos_rtloop_add_poll (priv->loop, item);
-      break;
-    }
-    case SPA_NODE_EVENT_TYPE_UPDATE_POLL:
-    {
-      SpaPollItem *item = event->data;
-      pinos_rtloop_update_poll (priv->loop, item);
-      break;
-    }
-    case SPA_NODE_EVENT_TYPE_REMOVE_POLL:
-    {
-      SpaPollItem *item = event->data;
-      pinos_rtloop_remove_poll (priv->loop, item);
-      break;
-    }
     case SPA_NODE_EVENT_TYPE_NEED_INPUT:
     {
       SpaNodeEventNeedInput *ni = event->data;

pinos/server/rt-loop.c

@@ -190,9 +190,11 @@ stop_thread (PinosRTLoop *this, gboolean in_thread)
   }
 }
 
-gboolean
+static SpaResult
-pinos_rtloop_add_poll (PinosRTLoop *this, SpaPollItem *item)
+do_add_item (SpaPoll         *poll,
+             SpaPollItem     *item)
 {
+  PinosRTLoop *this = SPA_CONTAINER_OF (poll, PinosRTLoop, poll);
   PinosRTLoopPrivate *priv = this->priv;
   gboolean in_thread = pthread_equal (priv->thread, pthread_self());
   unsigned int i;
@@ -211,12 +213,15 @@ pinos_rtloop_add_poll (PinosRTLoop *this, SpaPollItem *item)
     if (priv->poll[i].fds)
       g_debug ("poll %d: %p %d", i, priv->poll[i].user_data, priv->poll[i].fds[0].fd);
   }
-  return TRUE;
+  return SPA_RESULT_OK;
 }
 
-gboolean
+
-pinos_rtloop_update_poll (PinosRTLoop *this, SpaPollItem *item)
+static SpaResult
+do_update_item (SpaPoll         *poll,
+                SpaPollItem     *item)
 {
+  PinosRTLoop *this = SPA_CONTAINER_OF (poll, PinosRTLoop, poll);
   PinosRTLoopPrivate *priv = this->priv;
   gboolean in_thread = pthread_equal (priv->thread, pthread_self());
   unsigned int i;
@@ -231,13 +236,14 @@ pinos_rtloop_update_poll (PinosRTLoop *this, SpaPollItem *item)
   if (!in_thread)
     wakeup_thread (this);
 
-  return TRUE;
+  return SPA_RESULT_OK;
 }
 
-
+static SpaResult
-gboolean
+do_remove_item (SpaPoll         *poll,
-pinos_rtloop_remove_poll (PinosRTLoop *this, SpaPollItem *item)
+                SpaPollItem     *item)
 {
+  PinosRTLoop *this = SPA_CONTAINER_OF (poll, PinosRTLoop, poll);
   PinosRTLoopPrivate *priv = this->priv;
   gboolean in_thread = pthread_equal (priv->thread, pthread_self());
   unsigned int i;
@@ -263,7 +269,7 @@ pinos_rtloop_remove_poll (PinosRTLoop *this, SpaPollItem *item)
     if (priv->poll[i].fds)
       g_debug ("poll %d: %p %d", i, priv->poll[i].user_data, priv->poll[i].fds[0].fd);
   }
-  return TRUE;
+  return SPA_RESULT_OK;
 }
 
 static void
@@ -321,6 +327,13 @@ pinos_rtloop_init (PinosRTLoop * this)
   this->priv = PINOS_RTLOOP_GET_PRIVATE (this);
 
   g_debug ("rt-loop %p: new", this);
+
+  this->poll.handle = NULL;
+  this->poll.size = sizeof (SpaPoll);
+  this->poll.info = NULL;
+  this->poll.add_item = do_add_item;
+  this->poll.update_item = do_update_item;
+  this->poll.remove_item = do_remove_item;
 }
 
 /**

pinos/server/rt-loop.h

@@ -47,6 +47,8 @@ typedef struct _PinosRTLoopPrivate PinosRTLoopPrivate;
 struct _PinosRTLoop {
   GObject object;
 
+  SpaPoll poll;
+
   PinosRTLoopPrivate *priv;
 };
 
@@ -64,6 +66,7 @@ GType               pinos_rtloop_get_type                (void);
 
 PinosRTLoop *       pinos_rtloop_new                     (void);
 
+
 gboolean            pinos_rtloop_add_poll                (PinosRTLoop *loop,
                                                           SpaPollItem *item);
 gboolean            pinos_rtloop_update_poll             (PinosRTLoop *loop,

spa/include/spa/defs.h

@@ -88,6 +88,9 @@ typedef void (*SpaNotify) (void *data);
 #define SPA_CLAMP(v,a,b)  ((v)>(b) ? (b) : ((v) < (a) ? (a) : (v)))
 
 #define SPA_MEMBER(b,o,t) ((t*)((uint8_t*)(b) + (o)))
+
+#define SPA_CONTAINER_OF(p,t,m) (t*)((uint8_t*)p - offsetof (t,m))
+
 #define SPA_PTRDIFF(p1,p2) ((uint8_t*)(p1) - (uint8_t*)(p2))
 
 #define SPA_PTR_TO_INT(p) ((int) ((intptr_t) (p)))

spa/include/spa/dict.h

@@ -26,6 +26,8 @@ extern "C" {
 
 typedef struct _SpaDict SpaDict;
 
+#include <string.h>
+
 #include <spa/defs.h>
 
 typedef struct {
@@ -38,6 +40,18 @@ struct _SpaDict {
   SpaDictItem  *items;
 };
 
+static inline const char *
+spa_dict_lookup (const SpaDict *dict, const char *key)
+{
+  unsigned int i;
+
+  for (i = 0; i < dict->n_items; i++) {
+    if (!strcmp (dict->items[i].key, key))
+      return dict->items[i].value;
+  }
+  return NULL;
+}
+
 #ifdef __cplusplus
 }  /* extern "C" */
 #endif

spa/include/spa/node-event.h

@@ -37,9 +37,6 @@ typedef struct _SpaNodeEvent SpaNodeEvent;
 #define SPA_NODE_EVENT__HaveOutput            SPA_NODE_EVENT_PREFIX "HaveOutput"
 #define SPA_NODE_EVENT__NeedInput             SPA_NODE_EVENT_PREFIX "NeedInput"
 #define SPA_NODE_EVENT__ReuseBuffer           SPA_NODE_EVENT_PREFIX "ReuseBuffer"
-#define SPA_NODE_EVENT__AddPoll               SPA_NODE_EVENT_PREFIX "AddPoll"
-#define SPA_NODE_EVENT__UpdatePoll            SPA_NODE_EVENT_PREFIX "UpdatePoll"
-#define SPA_NODE_EVENT__RemovePoll            SPA_NODE_EVENT_PREFIX "RemovePoll"
 #define SPA_NODE_EVENT__Drained               SPA_NODE_EVENT_PREFIX "Drained"
 #define SPA_NODE_EVENT__Marker                SPA_NODE_EVENT_PREFIX "Marker"
 #define SPA_NODE_EVENT__Error                 SPA_NODE_EVENT_PREFIX "Error"
@@ -53,9 +50,6 @@ typedef struct _SpaNodeEvent SpaNodeEvent;
  * @SPA_NODE_EVENT_TYPE_HAVE_OUTPUT: emited when an async node has output that can be pulled
  * @SPA_NODE_EVENT_TYPE_NEED_INPUT: emited when more data can be pushed to an async node
  * @SPA_NODE_EVENT_TYPE_REUSE_BUFFER: emited when a buffer can be reused
- * @SPA_NODE_EVENT_TYPE_ADD_POLL: emited when a pollfd should be added. data points to #SpaPollItem
- * @SPA_NODE_EVENT_TYPE_UPDATE_POLL: update the pollfd item
- * @SPA_NODE_EVENT_TYPE_REMOVE_POLL: emited when a pollfd should be removed. data points to #SpaPollItem
  * @SPA_NODE_EVENT_TYPE_DRAINED: emited when DRAIN command completed
  * @SPA_NODE_EVENT_TYPE_MARKER: emited when MARK command completed
  * @SPA_NODE_EVENT_TYPE_ERROR: emited when error occured
@@ -69,9 +63,6 @@ typedef enum {
   SPA_NODE_EVENT_TYPE_HAVE_OUTPUT,
   SPA_NODE_EVENT_TYPE_NEED_INPUT,
   SPA_NODE_EVENT_TYPE_REUSE_BUFFER,
-  SPA_NODE_EVENT_TYPE_ADD_POLL,
-  SPA_NODE_EVENT_TYPE_UPDATE_POLL,
-  SPA_NODE_EVENT_TYPE_REMOVE_POLL,
   SPA_NODE_EVENT_TYPE_DRAINED,
   SPA_NODE_EVENT_TYPE_MARKER,
   SPA_NODE_EVENT_TYPE_ERROR,

spa/include/spa/poll.h

@@ -24,7 +24,16 @@
 extern "C" {
 #endif
 
+typedef struct _SpaPoll SpaPoll;
+
+#define SPA_POLL_URI             "http://spaplug.in/ns/poll"
+#define SPA_POLL_PREFIX          SPA_POLL_URI "#"
+#define SPA_POLL__MainLoop       SPA_POLL_PREFIX "MainLoop"
+#define SPA_POLL__DataLoop       SPA_POLL_PREFIX "DataLoop"
+
 #include <spa/defs.h>
+#include <spa/plugin.h>
+#include <spa/dict.h>
 
 /**
  * SpaPollFd:
@@ -56,7 +65,8 @@ typedef int (*SpaPollNotify) (SpaPollNotifyData *data);
 
 /**
  * SpaPollItem:
- * @id: id of the poll item
+ * @id: id of the poll item. This will be set when
+ *      adding the item to #SpaPoll.
  * @enabled: if the item is enabled
  * @fds: array of file descriptors to watch
  * @n_fds: number of elements in @fds
@@ -76,6 +86,50 @@ typedef struct {
   void          *user_data;
 } SpaPollItem;
 
+/**
+ * SpaPoll:
+ *
+ * Register poll events
+ */
+struct _SpaPoll {
+  /* pointer to the handle owning this interface */
+  SpaHandle *handle;
+  /* the total size of this structure. This can be used to expand this
+   * structure in the future */
+  size_t size;
+  /**
+   * SpaPoll::info
+   *
+   * Extra information
+   */
+  const SpaDict *info;
+  /**
+   * SpaPoll::add_item:
+   * @poll: a #SpaPoll
+   * @item: a #SpaPollItem
+   *
+   * Add @item to the list of polled items.
+   *
+   * The id in @item will be set and must be passed when updating or removing
+   * the @item.
+   *
+   * Returns: #SPA_RESULT_OK on success
+   *          #SPA_RESULT_INVALID_ARGUMENTS when @poll or @item is %NULL
+   */
+  SpaResult   (*add_item)            (SpaPoll         *poll,
+                                      SpaPollItem     *item);
+
+  SpaResult   (*update_item)         (SpaPoll         *poll,
+                                      SpaPollItem     *item);
+
+  SpaResult   (*remove_item)         (SpaPoll         *poll,
+                                      SpaPollItem     *item);
+};
+
+#define spa_poll_add_item(n,...)          (n)->add_item((n),__VA_ARGS__)
+#define spa_poll_update_item(n,...)       (n)->update_item((n),__VA_ARGS__)
+#define spa_poll_remove_item(n,...)       (n)->remove_item((n),__VA_ARGS__)
+
 #ifdef __cplusplus
 }  /* extern "C" */
 #endif

spa/lib/mapper.c

@@ -31,6 +31,7 @@
 #include <spa/node.h>
 #include <spa/node-command.h>
 #include <spa/node-event.h>
+#include <spa/poll.h>
 #include <spa/port.h>
 #include <spa/props.h>
 #include <spa/queue.h>
@@ -51,6 +52,8 @@ static const char *uris[] = {
   SPA_PROPS_URI,
   SPA_QUEUE_URI,
   SPA_RINGBUFFER_URI,
+  SPA_POLL__MainLoop,
+  SPA_POLL__DataLoop,
 };
 
 static uint32_t

spa/plugins/alsa/alsa-source.c

@@ -768,6 +768,8 @@ alsa_source_init (const SpaHandleFactory  *factory,
       this->map = support[i].data;
     else if (strcmp (support[i].uri, SPA_LOG_URI) == 0)
       this->log = support[i].data;
+    else if (strcmp (support[i].uri, SPA_POLL__DataLoop) == 0)
+      this->data_loop = support[i].data;
   }
   if (this->map == NULL) {
     spa_log_error (this->log, "an id-map is needed");

spa/plugins/alsa/alsa-utils.c

@@ -434,7 +434,6 @@ int
 spa_alsa_start (SpaALSAState *state)
 {
   int err;
-  SpaNodeEvent event;
 
   if (spa_alsa_open (state) < 0)
     return -1;
@@ -455,10 +454,6 @@ spa_alsa_start (SpaALSAState *state)
     return err;
   }
 
-  event.type = SPA_NODE_EVENT_TYPE_ADD_POLL;
-  event.data = &state->poll;
-  event.size = sizeof (state->poll);
-
   state->poll.id = 0;
   state->poll.enabled = true;
   state->poll.fds = state->fds;
@@ -466,7 +461,7 @@ spa_alsa_start (SpaALSAState *state)
   state->poll.before_cb = NULL;
   state->poll.after_cb = alsa_on_fd_events;
   state->poll.user_data = state;
-  state->event_cb (&state->node, &event, state->user_data);
+  spa_poll_add_item (state->data_loop, &state->poll);
 
   mmap_write (state);
   err = snd_pcm_start (state->hndl);
@@ -477,18 +472,11 @@ spa_alsa_start (SpaALSAState *state)
 int
 spa_alsa_stop (SpaALSAState *state)
 {
-  SpaNodeEvent event;
-
   if (!state->opened)
     return 0;
 
-  event.type = SPA_NODE_EVENT_TYPE_REMOVE_POLL;
+  spa_poll_remove_item (state->data_loop, &state->poll);
-  event.data = &state->poll;
-  event.size = sizeof (state->poll);
-  state->event_cb (&state->node, &event, state->user_data);
-
   snd_pcm_drop (state->hndl);
-
   spa_alsa_close (state);
 
   return 0;

spa/plugins/alsa/alsa-utils.h

@@ -69,6 +69,7 @@ struct _SpaALSAState {
   URI uri;
   SpaIDMap *map;
   SpaLog *log;
+  SpaPoll *data_loop;
 
   snd_pcm_stream_t stream;
   snd_output_t *output;

spa/plugins/audiotestsrc/audiotestsrc.c

@@ -70,6 +70,7 @@ struct _SpaAudioTestSrc {
   URI uri;
   SpaIDMap *map;
   SpaLog *log;
+  SpaPoll *data_loop;
 
   SpaAudioTestSrcProps props[2];
 
@@ -311,10 +312,7 @@ update_state (SpaAudioTestSrc *this, SpaNodeState state)
 static SpaResult
 update_poll_enabled (SpaAudioTestSrc *this, bool enabled)
 {
-  SpaNodeEvent event;
-
   if (this->event_cb) {
-    event.type = SPA_NODE_EVENT_TYPE_UPDATE_POLL;
     this->timer.enabled = enabled;
     if (this->props[1].live) {
       if (enabled) {
@@ -337,9 +335,7 @@ update_poll_enabled (SpaAudioTestSrc *this, bool enabled)
       this->timer.idle_cb = audiotestsrc_on_output;
       this->timer.after_cb = NULL;
     }
-    event.data = &this->timer;
+    spa_poll_update_item (this->data_loop, &this->timer);
-    event.size = sizeof (this->timer);
-    this->event_cb (&this->node, &event, this->user_data);
   }
   return SPA_RESULT_OK;
 }
@@ -416,7 +412,6 @@ spa_audiotestsrc_node_set_event_callback (SpaNode              *node,
                                           void                 *user_data)
 {
   SpaAudioTestSrc *this;
-  SpaNodeEvent event;
 
   if (node == NULL || node->handle == NULL)
     return SPA_RESULT_INVALID_ARGUMENTS;
@@ -424,20 +419,14 @@ spa_audiotestsrc_node_set_event_callback (SpaNode              *node,
   this = (SpaAudioTestSrc *) node->handle;
 
   if (event_cb == NULL && this->event_cb) {
-    event.type = SPA_NODE_EVENT_TYPE_REMOVE_POLL;
+    spa_poll_remove_item (this->data_loop, &this->timer);
-    event.data = &this->timer;
-    event.size = sizeof (this->timer);
-    this->event_cb (&this->node, &event, this->user_data);
   }
 
   this->event_cb = event_cb;
   this->user_data = user_data;
 
   if (this->event_cb) {
-    event.type = SPA_NODE_EVENT_TYPE_ADD_POLL;
+    spa_poll_add_item (this->data_loop, &this->timer);
-    event.data = &this->timer;
-    event.size = sizeof (this->timer);
-    this->event_cb (&this->node, &event, this->user_data);
   }
   return SPA_RESULT_OK;
 }
@@ -1016,11 +1005,17 @@ audiotestsrc_init (const SpaHandleFactory  *factory,
       this->map = support[i].data;
     else if (strcmp (support[i].uri, SPA_LOG_URI) == 0)
       this->log = support[i].data;
+    else if (strcmp (support[i].uri, SPA_POLL__DataLoop) == 0)
+      this->data_loop = support[i].data;
   }
   if (this->map == NULL) {
     spa_log_error (this->log, "an id-map is needed");
     return SPA_RESULT_ERROR;
   }
+  if (this->data_loop == NULL) {
+    spa_log_error (this->log, "a data_loop is needed");
+    return SPA_RESULT_ERROR;
+  }
   this->uri.node = spa_id_map_get_id (this->map, SPA_NODE_URI);
   this->uri.clock = spa_id_map_get_id (this->map, SPA_CLOCK_URI);
 

spa/plugins/remote/proxy.c

@@ -100,6 +100,7 @@ struct _SpaProxy {
   URI uri;
   SpaIDMap *map;
   SpaLog *log;
+  SpaPoll *data_loop;
 
   SpaProxyProps props[2];
 
@@ -143,26 +144,19 @@ reset_proxy_props (SpaProxyProps *props)
 static SpaResult
 update_poll (SpaProxy *this, int socketfd)
 {
-  SpaNodeEvent event;
   SpaProxyProps *p;
   SpaResult res = SPA_RESULT_OK;
 
   p = &this->props[1];
 
   if (p->socketfd != -1) {
-    event.type = SPA_NODE_EVENT_TYPE_REMOVE_POLL;
+    spa_poll_remove_item (this->data_loop, &this->poll);
-    event.data = &this->poll;
-    event.size = sizeof (this->poll);
-    this->event_cb (&this->node, &event, this->user_data);
   }
   p->socketfd = socketfd;
 
   if (p->socketfd != -1) {
     this->fds[0].fd = p->socketfd;
-    event.type = SPA_NODE_EVENT_TYPE_ADD_POLL;
+    spa_poll_add_item (this->data_loop, &this->poll);
-    event.data = &this->poll;
-    event.size = sizeof (this->poll);
-    this->event_cb (&this->node, &event, this->user_data);
   }
   return res;
 }
@@ -1121,9 +1115,6 @@ handle_node_event (SpaProxy     *this,
     case SPA_NODE_EVENT_TYPE_HAVE_OUTPUT:
     case SPA_NODE_EVENT_TYPE_NEED_INPUT:
     case SPA_NODE_EVENT_TYPE_REUSE_BUFFER:
-    case SPA_NODE_EVENT_TYPE_ADD_POLL:
-    case SPA_NODE_EVENT_TYPE_UPDATE_POLL:
-    case SPA_NODE_EVENT_TYPE_REMOVE_POLL:
     case SPA_NODE_EVENT_TYPE_DRAINED:
     case SPA_NODE_EVENT_TYPE_MARKER:
     case SPA_NODE_EVENT_TYPE_ERROR:
@@ -1377,11 +1368,17 @@ proxy_init (const SpaHandleFactory  *factory,
       this->map = support[i].data;
     else if (strcmp (support[i].uri, SPA_LOG_URI) == 0)
       this->log = support[i].data;
+    else if (strcmp (support[i].uri, SPA_POLL__DataLoop) == 0)
+      this->data_loop = support[i].data;
   }
   if (this->map == NULL) {
     spa_log_error (this->log, "an id-map is needed");
     return SPA_RESULT_ERROR;
   }
+  if (this->data_loop == NULL) {
+    spa_log_error (this->log, "a data-loop is needed");
+    return SPA_RESULT_ERROR;
+  }
   this->uri.node = spa_id_map_get_id (this->map, SPA_NODE_URI);
 
   this->node = proxy_node;

spa/plugins/v4l2/v4l2-source.c

@@ -86,6 +86,7 @@ typedef struct {
 
 typedef struct {
   SpaLog *log;
+  SpaPoll *data_loop;
 
   bool export_buf;
   bool started;
@@ -843,6 +844,7 @@ v4l2_source_init (const SpaHandleFactory  *factory,
 {
   SpaV4l2Source *this;
   unsigned int i;
+  const char *str;
 
   if (factory == NULL || handle == NULL)
     return SPA_RESULT_INVALID_ARGUMENTS;
@@ -857,11 +859,17 @@ v4l2_source_init (const SpaHandleFactory  *factory,
       this->map = support[i].data;
     else if (strcmp (support[i].uri, SPA_LOG_URI) == 0)
       this->log = support[i].data;
+    else if (strcmp (support[i].uri, SPA_POLL__DataLoop) == 0)
+      this->state[0].data_loop = support[i].data;
   }
   if (this->map == NULL) {
     spa_log_error (this->log, "an id-map is needed");
     return SPA_RESULT_ERROR;
   }
+  if (this->state[0].data_loop == NULL) {
+    spa_log_error (this->log, "a data_loop is needed");
+    return SPA_RESULT_ERROR;
+  }
   this->uri.node = spa_id_map_get_id (this->map, SPA_NODE_URI);
   this->uri.clock = spa_id_map_get_id (this->map, SPA_CLOCK_URI);
 
@@ -881,11 +889,9 @@ v4l2_source_init (const SpaHandleFactory  *factory,
 
   this->state[0].export_buf = true;
 
-  for (i = 0; info && i < info->n_items; i++) {
+  if (info && (str = spa_dict_lookup (info, "device.path"))) {
-    if (!strcmp (info->items[i].key, "device.path")) {
+    strncpy (this->props[1].device, str, 63);
-      strncpy (this->props[1].device, info->items[i].value, 63);
+    this->props[1].props.unset_mask &= ~1;
-      this->props[1].props.unset_mask &= ~1;
-    }
   }
 
   update_state (this, SPA_NODE_STATE_CONFIGURE);

spa/plugins/v4l2/v4l2-utils.c

@@ -1121,7 +1121,6 @@ spa_v4l2_start (SpaV4l2Source *this)
 {
   SpaV4l2State *state = &this->state[0];
   enum v4l2_buf_type type;
-  SpaNodeEvent event;
 
   if (state->started)
     return SPA_RESULT_OK;
@@ -1134,10 +1133,6 @@ spa_v4l2_start (SpaV4l2Source *this)
   state->started = true;
   update_state (this, SPA_NODE_STATE_STREAMING);
 
-  event.type = SPA_NODE_EVENT_TYPE_ADD_POLL;
-  event.data = &state->poll;
-  event.size = sizeof (state->poll);
-
   state->fds[0].fd = state->fd;
   state->fds[0].events = POLLIN | POLLPRI | POLLERR;
   state->fds[0].revents = 0;
@@ -1150,7 +1145,7 @@ spa_v4l2_start (SpaV4l2Source *this)
   state->poll.before_cb = NULL;
   state->poll.after_cb = v4l2_on_fd_events;
   state->poll.user_data = this;
-  this->event_cb (&this->node, &event, this->user_data);
+  spa_poll_add_item (state->data_loop, &state->poll);
 
   return SPA_RESULT_OK;
 }
@@ -1160,7 +1155,6 @@ spa_v4l2_pause (SpaV4l2Source *this)
 {
   SpaV4l2State *state = &this->state[0];
   enum v4l2_buf_type type;
-  SpaNodeEvent event;
   int i;
 
   if (!state->started)
@@ -1168,10 +1162,7 @@ spa_v4l2_pause (SpaV4l2Source *this)
 
   state->started = false;
 
-  event.type = SPA_NODE_EVENT_TYPE_REMOVE_POLL;
+  spa_poll_remove_item (state->data_loop, &state->poll);
-  event.data = &state->poll;
-  event.size = sizeof (state->poll);
-  this->event_cb (&this->node, &event, this->user_data);
 
   type = V4L2_BUF_TYPE_VIDEO_CAPTURE;
   if (xioctl (state->fd, VIDIOC_STREAMOFF, &type) < 0) {

spa/plugins/videotestsrc/videotestsrc.c

@@ -70,9 +70,10 @@ struct _SpaVideoTestSrc {
   SpaNode node;
   SpaClock clock;
 
+  URI uri;
   SpaIDMap *map;
   SpaLog *log;
-  URI uri;
+  SpaPoll *data_loop;
 
   SpaVideoTestSrcProps props[2];
 
@@ -259,10 +260,7 @@ update_state (SpaVideoTestSrc *this, SpaNodeState state)
 static SpaResult
 update_poll_enabled (SpaVideoTestSrc *this, bool enabled)
 {
-  SpaNodeEvent event;
-
   if (this->event_cb && this->timer.enabled != enabled) {
-    event.type = SPA_NODE_EVENT_TYPE_UPDATE_POLL;
     this->timer.enabled = enabled;
     if (this->props[1].live) {
       if (enabled) {
@@ -285,9 +283,7 @@ update_poll_enabled (SpaVideoTestSrc *this, bool enabled)
       this->timer.idle_cb = videotestsrc_on_output;
       this->timer.after_cb = NULL;
     }
-    event.data = &this->timer;
+    spa_poll_update_item (this->data_loop, &this->timer);
-    event.size = sizeof (this->timer);
-    this->event_cb (&this->node, &event, this->user_data);
   }
   return SPA_RESULT_OK;
 }
@@ -364,7 +360,6 @@ spa_videotestsrc_node_set_event_callback (SpaNode              *node,
                                           void                 *user_data)
 {
   SpaVideoTestSrc *this;
-  SpaNodeEvent event;
 
   if (node == NULL || node->handle == NULL)
     return SPA_RESULT_INVALID_ARGUMENTS;
@@ -372,20 +367,14 @@ spa_videotestsrc_node_set_event_callback (SpaNode              *node,
   this = (SpaVideoTestSrc *) node->handle;
 
   if (event_cb == NULL && this->event_cb) {
-    event.type = SPA_NODE_EVENT_TYPE_REMOVE_POLL;
+    spa_poll_remove_item (this->data_loop, &this->timer);
-    event.data = &this->timer;
-    event.size = sizeof (this->timer);
-    this->event_cb (&this->node, &event, this->user_data);
   }
 
   this->event_cb = event_cb;
   this->user_data = user_data;
 
   if (this->event_cb) {
-    event.type = SPA_NODE_EVENT_TYPE_ADD_POLL;
+    spa_poll_add_item (this->data_loop, &this->timer);
-    event.data = &this->timer;
-    event.size = sizeof (this->timer);
-    this->event_cb (&this->node, &event, this->user_data);
   }
   return SPA_RESULT_OK;
 }
@@ -964,6 +953,8 @@ videotestsrc_init (const SpaHandleFactory  *factory,
       this->map = support[i].data;
     else if (strcmp (support[i].uri, SPA_LOG_URI) == 0)
       this->log = support[i].data;
+    else if (strcmp (support[i].uri, SPA_POLL__DataLoop) == 0)
+      this->data_loop = support[i].data;
   }
   if (this->map == NULL) {
     spa_log_error (this->log, "an id-map is needed");

spa/tests/test-mixer.c

@@ -51,6 +51,7 @@ typedef struct {
   unsigned int n_support;
   SpaIDMap *map;
   SpaLog *log;
+  SpaPoll data_loop;
   URI uri;
 } AppData;
 
@@ -165,25 +166,29 @@ on_sink_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
         printf ("got error %d\n", res);
       break;
     }
-    case SPA_NODE_EVENT_TYPE_ADD_POLL:
-    {
-      SpaPollItem *poll = event->data;
-      int i;
-
-      data->poll = *poll;
-      for (i = 0; i < data->poll.n_fds; i++) {
-        data->fds[i] = poll->fds[i];
-      }
-      data->n_fds = data->poll.n_fds;
-      data->poll.fds = data->fds;
-      break;
-    }
     default:
       printf ("got event %d\n", event->type);
       break;
   }
 }
 
+static SpaResult
+do_add_item (SpaPoll     *poll,
+             SpaPollItem *item)
+{
+  AppData *data = SPA_CONTAINER_OF (poll, AppData, data_loop);
+  int i;
+
+  data->poll = *item;
+  for (i = 0; i < data->poll.n_fds; i++) {
+    data->fds[i] = item->fds[i];
+  }
+  data->n_fds = data->poll.n_fds;
+  data->poll.fds = data->fds;
+
+  return SPA_RESULT_OK;
+}
+
 static SpaResult
 make_nodes (AppData *data)
 {
@@ -355,9 +360,18 @@ main (int argc, char *argv[])
   SpaResult res;
 
   data.map = spa_id_map_get_default();
+  data.data_loop.handle = NULL;
+  data.data_loop.size = sizeof (SpaPoll);
+  data.data_loop.info = NULL;
+  data.data_loop.add_item = do_add_item;
+  data.data_loop.update_item = NULL;
+  data.data_loop.remove_item = NULL;
+
   data.support[0].uri = SPA_ID_MAP_URI;
   data.support[0].data = data.map;
-  data.n_support = 1;
+  data.support[1].uri = SPA_POLL__DataLoop;
+  data.support[1].data = &data.data_loop;
+  data.n_support = 2;
 
   data.uri.node = spa_id_map_get_id (data.map, SPA_NODE_URI);
 

spa/tests/test-v4l2.c

@@ -71,6 +71,7 @@ typedef struct {
   unsigned int n_support;
   SpaIDMap *map;
   SpaLog *log;
+  SpaPoll data_loop;
   URI uri;
 } AppData;
 
@@ -106,7 +107,7 @@ make_node (AppData *data, SpaNode **node, const char *lib, const char *name)
       continue;
 
     handle = calloc (1, factory->size);
-    if ((res = spa_handle_factory_init (factory, handle, NULL, NULL, 0)) < 0) {
+    if ((res = spa_handle_factory_init (factory, handle, NULL, data->support, data->n_support)) < 0) {
       printf ("can't make factory instance: %d\n", res);
       return res;
     }
@@ -188,22 +189,43 @@ on_source_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
       spa_node_port_reuse_buffer (data->source, 0, info->buffer_id);
       break;
     }
-    case SPA_NODE_EVENT_TYPE_ADD_POLL:
-    {
-      SpaPollItem *poll = event->data;
-
-      data->poll = *poll;
-      data->fds[0] = poll->fds[0];
-      data->n_fds = 1;
-      data->poll.fds = data->fds;
-      break;
-    }
     default:
       printf ("got event %d\n", event->type);
       break;
   }
 }
 
+static SpaResult
+do_add_item (SpaPoll     *poll,
+             SpaPollItem *item)
+{
+  AppData *data = SPA_CONTAINER_OF (poll, AppData, data_loop);
+  int i;
+
+  data->poll = *item;
+  for (i = 0; i < data->poll.n_fds; i++) {
+    data->fds[i] = item->fds[i];
+  }
+  data->n_fds = data->poll.n_fds;
+  data->poll.fds = data->fds;
+
+  return SPA_RESULT_OK;
+}
+
+static SpaResult
+do_update_item (SpaPoll     *poll,
+             SpaPollItem *item)
+{
+  return SPA_RESULT_OK;
+}
+
+static SpaResult
+do_remove_item (SpaPoll     *poll,
+             SpaPollItem *item)
+{
+  return SPA_RESULT_OK;
+}
+
 static SpaResult
 make_nodes (AppData *data, const char *device)
 {
@@ -220,7 +242,7 @@ make_nodes (AppData *data, const char *device)
   if ((res = spa_node_get_props (data->source, &props)) < 0)
     printf ("got get_props error %d\n", res);
 
-  value.value = device ? device : "/dev/video0";
+  value.value = device ? device : "/dev/video7";
   value.size = strlen (value.value)+1;
   spa_props_set_value (props, spa_props_index_for_name (props, "device"), &value);
 
@@ -433,9 +455,18 @@ main (int argc, char *argv[])
 
   data.map = spa_id_map_get_default ();
 
+  data.data_loop.handle = NULL;
+  data.data_loop.size = sizeof (SpaPoll);
+  data.data_loop.info = NULL;
+  data.data_loop.add_item = do_add_item;
+  data.data_loop.update_item = do_update_item;
+  data.data_loop.remove_item = do_remove_item;
+
   data.support[0].uri = SPA_ID_MAP_URI;
   data.support[0].data = data.map;
-  data.n_support = 1;
+  data.support[1].uri = SPA_POLL__DataLoop;
+  data.support[1].data = &data.data_loop;
+  data.n_support = 2;
 
   data.uri.node = spa_id_map_get_id (data.map, SPA_NODE_URI);