diff --git a/pinos/server/client-node.c b/pinos/server/client-node.c index 8fb6db433..d1bcb3783 100644 --- a/pinos/server/client-node.c +++ b/pinos/server/client-node.c @@ -217,10 +217,12 @@ spa_proxy_node_send_command (SpaNode *node, cnc.command = command; pinos_connection_add_cmd (this->conn, PINOS_CONTROL_CMD_NODE_COMMAND, &cnc); - if (!pinos_connection_flush (this->conn)) + if (!pinos_connection_flush (this->conn)) { spa_log_error (this->log, "proxy %p: error writing connection\n", this); - - res = SPA_RESULT_RETURN_ASYNC (cnc.seq); + res = SPA_RESULT_ERROR; + } + else + res = SPA_RESULT_RETURN_ASYNC (cnc.seq); break; } @@ -232,8 +234,10 @@ spa_proxy_node_send_command (SpaNode *node, cnc.command = command; pinos_connection_add_cmd (this->conn, PINOS_CONTROL_CMD_NODE_COMMAND, &cnc); - if (!pinos_connection_flush (this->conn)) + if (!pinos_connection_flush (this->conn)) { spa_log_error (this->log, "proxy %p: error writing connection\n", this); + res = SPA_RESULT_ERROR; + } break; } @@ -1287,7 +1291,7 @@ proxy_init (SpaProxy *this, this->rtpoll.after_cb = proxy_on_rtfd_events; this->rtpoll.user_data = this; - return SPA_RESULT_OK; + return SPA_RESULT_RETURN_ASYNC (this->seq++); } static SpaResult diff --git a/pinos/server/data-loop.c b/pinos/server/data-loop.c index 5d19d8da0..3bff749d8 100644 --- a/pinos/server/data-loop.c +++ b/pinos/server/data-loop.c @@ -24,15 +24,29 @@ #include #include -#include +#include "spa/include/spa/ringbuffer.h" #include "pinos/server/data-loop.h" #define PINOS_DATA_LOOP_GET_PRIVATE(loop) \ (G_TYPE_INSTANCE_GET_PRIVATE ((loop), PINOS_TYPE_DATA_LOOP, PinosDataLoopPrivate)) +#define DATAS_SIZE (4096 * 8) + +typedef struct { + size_t item_size; + SpaPollInvokeFunc func; + uint32_t seq; + size_t size; + void *data; + void *user_data; +} InvokeItem; + struct _PinosDataLoopPrivate { + SpaRingbuffer buffer; + uint8_t buffer_data[DATAS_SIZE]; + unsigned int n_poll; SpaPollItem poll[16]; int idx[16]; @@ -42,6 +56,7 @@ struct _PinosDataLoopPrivate unsigned int n_fds; uint32_t counter; + uint32_t seq; gboolean running; pthread_t thread; @@ -64,6 +79,7 @@ loop (void *user_data) { PinosDataLoop *this = user_data; PinosDataLoopPrivate *priv = this->priv; + SpaPoll *p = &this->poll; unsigned int i, j; g_debug ("data-loop %p: enter thread", this); @@ -80,7 +96,7 @@ loop (void *user_data) ndata.fds = NULL; ndata.n_fds = 0; ndata.user_data = p->user_data; - if (p->idle_cb (&ndata) < 0) + if (SPA_RESULT_IS_ERROR (p->idle_cb (&ndata))) p->enabled = false; n_idle++; } @@ -114,7 +130,8 @@ loop (void *user_data) ndata.fds = &priv->fds[priv->idx[i]]; ndata.n_fds = p->n_fds; ndata.user_data = p->user_data; - p->before_cb (&ndata); + if (SPA_RESULT_IS_ERROR (p->before_cb (&ndata))) + p->enabled = false; } } @@ -125,15 +142,24 @@ loop (void *user_data) break; } if (r == 0) { - g_debug ("data-loop %p: select timeout", this); - break; + g_debug ("data-loop %p: select timeout should not happen", this); + continue; } /* check wakeup */ if (priv->fds[0].revents & POLLIN) { uint64_t u; + size_t offset; + if (read (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t)) - g_warning ("data-loop %p: failed to read fd", strerror (errno)); + g_warning ("data-loop %p: failed to read fd: %s", this, strerror (errno)); + + while (spa_ringbuffer_get_read_offset (&priv->buffer, &offset) > 0) { + InvokeItem *item = SPA_MEMBER (priv->buffer_data, offset, InvokeItem); + g_debug ("data-loop %p: invoke %d", this, item->seq); + item->func (p, item->seq, item->size, item->data, item->user_data); + spa_ringbuffer_read_advance (&priv->buffer, item->item_size); + } continue; } @@ -145,7 +171,8 @@ loop (void *user_data) ndata.fds = &priv->fds[priv->idx[i]]; ndata.n_fds = p->n_fds; ndata.user_data = p->user_data; - p->after_cb (&ndata); + if (SPA_RESULT_IS_ERROR (p->after_cb (&ndata))) + p->enabled = false; } } } @@ -161,7 +188,7 @@ wakeup_thread (PinosDataLoop *this) uint64_t u = 1; if (write (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t)) - g_warning ("data-loop %p: failed to write fd", strerror (errno)); + g_warning ("data-loop %p: failed to write fd: %s", this, strerror (errno)); } static void @@ -173,7 +200,7 @@ start_thread (PinosDataLoop *this) if (!priv->running) { priv->running = true; if ((err = pthread_create (&priv->thread, NULL, loop, this)) != 0) { - g_warning ("data-loop %p: can't create thread", strerror (err)); + g_warning ("data-loop %p: can't create thread: %s", this, strerror (err)); priv->running = false; } } @@ -276,6 +303,48 @@ do_remove_item (SpaPoll *poll, return SPA_RESULT_OK; } +static SpaResult +do_invoke (SpaPoll *poll, + SpaPollInvokeFunc func, + uint32_t seq, + size_t size, + void *data, + void *user_data) +{ + PinosDataLoop *this = SPA_CONTAINER_OF (poll, PinosDataLoop, poll); + PinosDataLoopPrivate *priv = this->priv; + SpaRingbufferArea areas[2]; + InvokeItem *item; + + spa_ringbuffer_get_write_areas (&priv->buffer, areas); + if (areas[0].len < sizeof (InvokeItem)) { + g_warning ("queue full"); + return SPA_RESULT_ERROR; + } + item = SPA_MEMBER (priv->buffer_data, areas[0].offset, InvokeItem); + item->seq = seq; + item->func = func; + item->user_data = user_data; + item->size = size; + + if (areas[0].len > sizeof (InvokeItem) + size) { + item->data = SPA_MEMBER (item, sizeof (InvokeItem), void); + item->item_size = sizeof (InvokeItem) + size; + if (areas[0].len < sizeof (InvokeItem) + item->item_size) + item->item_size = areas[0].len; + } else { + item->data = SPA_MEMBER (priv->buffer_data, areas[1].offset, void); + item->item_size = areas[0].len + 1 + size; + } + memcpy (item->data, data, size); + + spa_ringbuffer_write_advance (&priv->buffer, item->item_size); + + wakeup_thread (this); + + return SPA_RESULT_RETURN_ASYNC (seq); +} + static void pinos_data_loop_constructed (GObject * obj) { @@ -328,7 +397,7 @@ pinos_data_loop_class_init (PinosDataLoopClass * klass) static void pinos_data_loop_init (PinosDataLoop * this) { - this->priv = PINOS_DATA_LOOP_GET_PRIVATE (this); + PinosDataLoopPrivate *priv = this->priv = PINOS_DATA_LOOP_GET_PRIVATE (this); g_debug ("data-loop %p: new", this); @@ -337,6 +406,9 @@ pinos_data_loop_init (PinosDataLoop * this) this->poll.add_item = do_add_item; this->poll.update_item = do_update_item; this->poll.remove_item = do_remove_item; + this->poll.invoke = do_invoke; + + spa_ringbuffer_init (&priv->buffer, DATAS_SIZE); } /** diff --git a/pinos/server/data-loop.h b/pinos/server/data-loop.h index 0ff1a1c41..84e2f36bf 100644 --- a/pinos/server/data-loop.h +++ b/pinos/server/data-loop.h @@ -25,6 +25,7 @@ G_BEGIN_DECLS #include +#include typedef struct _PinosDataLoop PinosDataLoop; typedef struct _PinosDataLoopClass PinosDataLoopClass; @@ -61,10 +62,14 @@ struct _PinosDataLoopClass { GObjectClass parent_class; }; -/* normal GObject stuff */ -GType pinos_data_loop_get_type (void); +typedef void (*PinosCommandFunc) (SpaNodeCommand *command, + uint32_t seq, + void *user_data); -PinosDataLoop * pinos_data_loop_new (void); +/* normal GObject stuff */ +GType pinos_data_loop_get_type (void); + +PinosDataLoop * pinos_data_loop_new (void); G_END_DECLS diff --git a/pinos/server/main-loop.c b/pinos/server/main-loop.c index c5f1c95f6..987bbfc39 100644 --- a/pinos/server/main-loop.c +++ b/pinos/server/main-loop.c @@ -19,19 +19,56 @@ #include #include +#include +#include +#include #include +#include "spa/include/spa/queue.h" +#include "spa/include/spa/ringbuffer.h" #include "pinos/server/main-loop.h" #define PINOS_MAIN_LOOP_GET_PRIVATE(loop) \ (G_TYPE_INSTANCE_GET_PRIVATE ((loop), PINOS_TYPE_MAIN_LOOP, PinosMainLoopPrivate)) +#define DATAS_SIZE (4096 * 8) + +typedef struct { + size_t item_size; + SpaPollInvokeFunc func; + uint32_t seq; + size_t size; + void *data; + void *user_data; +} InvokeItem; + +typedef struct _WorkItem WorkItem; + +struct _WorkItem { + gulong id; + gpointer obj; + uint32_t seq; + SpaResult res; + PinosDeferFunc func; + gpointer *data; + GDestroyNotify notify; + WorkItem *next; +}; + struct _PinosMainLoopPrivate { gulong counter; - GQueue work; + SpaRingbuffer buffer; + uint8_t buffer_data[DATAS_SIZE]; + + SpaPollFd fds[1]; + SpaPollItem wakeup; + + SpaQueue work; + WorkItem *free_list; + gulong work_id; }; @@ -115,13 +152,94 @@ do_remove_item (SpaPoll *poll, return SPA_RESULT_OK; } +static int +main_loop_dispatch (SpaPollNotifyData *data) +{ + PinosMainLoop *this = data->user_data; + PinosMainLoopPrivate *priv = this->priv; + SpaPoll *p = &this->poll; + uint64_t u; + size_t offset; + InvokeItem *item; + + if (read (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t)) + g_warning ("main-loop %p: failed to read fd", strerror (errno)); + + while (spa_ringbuffer_get_read_offset (&priv->buffer, &offset) > 0) { + item = SPA_MEMBER (priv->buffer_data, offset, InvokeItem); + item->func (p, item->seq, item->size, item->data, item->user_data); + spa_ringbuffer_read_advance (&priv->buffer, item->item_size); + } + + return 0; +} + +static SpaResult +do_invoke (SpaPoll *poll, + SpaPollInvokeFunc func, + uint32_t seq, + size_t size, + void *data, + void *user_data) +{ + PinosMainLoop *this = SPA_CONTAINER_OF (poll, PinosMainLoop, poll); + PinosMainLoopPrivate *priv = this->priv; + SpaRingbufferArea areas[2]; + InvokeItem *item; + uint64_t u = 1; + + spa_ringbuffer_get_write_areas (&priv->buffer, areas); + if (areas[0].len < sizeof (InvokeItem)) { + g_warning ("queue full"); + return SPA_RESULT_ERROR; + } + item = SPA_MEMBER (priv->buffer_data, areas[0].offset, InvokeItem); + item->seq = seq; + item->func = func; + item->user_data = user_data; + item->size = size; + + if (areas[0].len > sizeof (InvokeItem) + size) { + item->data = SPA_MEMBER (item, sizeof (InvokeItem), void); + item->item_size = sizeof (InvokeItem) + size; + if (areas[0].len < sizeof (InvokeItem) + item->item_size) + item->item_size = areas[0].len; + } else { + item->data = SPA_MEMBER (priv->buffer_data, areas[1].offset, void); + item->item_size = areas[0].len + 1 + size; + } + memcpy (item->data, data, size); + + spa_ringbuffer_write_advance (&priv->buffer, item->item_size); + + if (write (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t)) + g_warning ("data-loop %p: failed to write fd", strerror (errno)); + + return SPA_RESULT_RETURN_ASYNC (seq); +} + static void pinos_main_loop_constructed (GObject * obj) { PinosMainLoop *this = PINOS_MAIN_LOOP (obj); + PinosMainLoopPrivate *priv = this->priv; g_debug ("main-loop %p: constructed", this); + priv->fds[0].fd = eventfd (0, 0); + priv->fds[0].events = POLLIN | POLLPRI | POLLERR; + priv->fds[0].revents = 0; + + priv->wakeup.id = SPA_ID_INVALID; + priv->wakeup.enabled = false; + priv->wakeup.fds = priv->fds; + priv->wakeup.n_fds = 1; + priv->wakeup.idle_cb = NULL; + priv->wakeup.before_cb = NULL; + priv->wakeup.after_cb = main_loop_dispatch; + priv->wakeup.user_data = this; + do_add_item (&this->poll, &priv->wakeup); + G_OBJECT_CLASS (pinos_main_loop_parent_class)->constructed (obj); } @@ -169,8 +287,10 @@ pinos_main_loop_init (PinosMainLoop * this) this->poll.add_item = do_add_item; this->poll.update_item = do_update_item; this->poll.remove_item = do_remove_item; + this->poll.invoke = do_invoke; - g_queue_init (&priv->work); + SPA_QUEUE_INIT (&priv->work); + spa_ringbuffer_init (&priv->buffer, DATAS_SIZE); } /** @@ -186,42 +306,36 @@ pinos_main_loop_new (void) return g_object_new (PINOS_TYPE_MAIN_LOOP, NULL); } -typedef struct { - gulong id; - gpointer obj; - uint32_t seq; - SpaResult res; - PinosDeferFunc func; - gpointer *data; - GDestroyNotify notify; -} WorkItem; - static gboolean process_work_queue (PinosMainLoop *this) { PinosMainLoopPrivate *priv = this->priv; - GList *walk, *next; + WorkItem *prev, *item, *next; - for (walk = priv->work.head; walk; walk = next) { - WorkItem *item = walk->data; + priv->work_id = 0; - next = g_list_next (walk); + for (item = priv->work.head, prev = NULL; item; prev = item, item = next) { + next = item->next; - g_debug ("main-loop %p: peek work queue item %p seq %d", this, item, item ? item->seq : -1); + g_debug ("main-loop %p: peek work queue item %p seq %d, prev %p next %p", this, item->obj, item->seq, prev, next); if (item->seq != SPA_ID_INVALID) continue; - g_debug ("main-loop %p: process work item %p", this, item); + g_debug ("main-loop %p: process work item %p", this, item->obj); + if (priv->work.tail == item) + priv->work.tail = prev; + if (prev == NULL) + priv->work.head = next; + else + prev->next = next; + if (item->func) - item->func (item->data, item->res, item->id); + item->func (item->obj, item->data, item->res, item->id); if (item->notify) item->notify (item->data); - - g_queue_delete_link (&priv->work, walk); g_slice_free (WorkItem, item); + item = prev; } - - priv->work_id = 0; return FALSE; } @@ -246,6 +360,7 @@ pinos_main_loop_defer (PinosMainLoop *loop, item->func = func; item->data = data; item->notify = notify; + item->next = NULL; if (SPA_RESULT_IS_ASYNC (res)) { item->seq = SPA_RESULT_ASYNC_SEQ (res); @@ -257,7 +372,7 @@ pinos_main_loop_defer (PinosMainLoop *loop, have_work = TRUE; g_debug ("main-loop %p: defer object %p", loop, obj); } - g_queue_push_tail (&priv->work, item); + SPA_QUEUE_PUSH_TAIL (&priv->work, WorkItem, next, item); if (priv->work_id == 0 && have_work) priv->work_id = g_idle_add ((GSourceFunc) process_work_queue, loop); @@ -270,19 +385,18 @@ pinos_main_loop_defer_cancel (PinosMainLoop *loop, gpointer obj, gulong id) { - GList *walk; PinosMainLoopPrivate *priv; gboolean have_work = FALSE; + WorkItem *item; g_return_if_fail (PINOS_IS_MAIN_LOOP (loop)); priv = loop->priv; - for (walk = priv->work.head; walk; walk = g_list_next (walk)) { - WorkItem *i = walk->data; - if ((id == 0 || i->id == id) && (obj == NULL || i->obj == obj)) { - g_debug ("main-loop %p: cancel defer %d for object %p", loop, i->seq, i->obj); - i->seq = SPA_ID_INVALID; - i->func = NULL; + for (item = priv->work.head; item; item = item->next) { + if ((id == 0 || item->id == id) && (obj == NULL || item->obj == obj)) { + g_debug ("main-loop %p: cancel defer %d for object %p", loop, item->seq, item->obj); + item->seq = SPA_ID_INVALID; + item->func = NULL; have_work = TRUE; } } @@ -296,7 +410,7 @@ pinos_main_loop_defer_complete (PinosMainLoop *loop, uint32_t seq, SpaResult res) { - GList *walk; + WorkItem *item; PinosMainLoopPrivate *priv; gboolean have_work = FALSE; @@ -305,16 +419,17 @@ pinos_main_loop_defer_complete (PinosMainLoop *loop, g_debug ("main-loop %p: async complete %d %d for object %p", loop, seq, res, obj); - for (walk = priv->work.head; walk; walk = g_list_next (walk)) { - WorkItem *i = walk->data; - - if (i->obj == obj && i->seq == seq) { + for (item = priv->work.head; item; item = item->next) { + if (item->obj == obj && item->seq == seq) { g_debug ("main-loop %p: found defered %d for object %p", loop, seq, obj); - i->seq = SPA_ID_INVALID; - i->res = res; + item->seq = SPA_ID_INVALID; + item->res = res; have_work = TRUE; } } + if (!have_work) + g_debug ("main-loop %p: no defered %d found for object %p", loop, seq, obj); + if (priv->work_id == 0 && have_work) priv->work_id = g_idle_add ((GSourceFunc) process_work_queue, loop); } diff --git a/pinos/server/main-loop.h b/pinos/server/main-loop.h index 6d9b732e0..6a2f71aa8 100644 --- a/pinos/server/main-loop.h +++ b/pinos/server/main-loop.h @@ -25,6 +25,7 @@ G_BEGIN_DECLS #include +#include typedef struct _PinosMainLoop PinosMainLoop; typedef struct _PinosMainLoopClass PinosMainLoopClass; @@ -61,7 +62,11 @@ struct _PinosMainLoopClass { GObjectClass parent_class; }; -typedef void (*PinosDeferFunc) (gpointer data, +typedef void (*PinosEventFunc) (SpaNodeEvent *event, + void *user_data); + +typedef void (*PinosDeferFunc) (gpointer obj, + gpointer data, SpaResult res, gulong id); diff --git a/pinos/server/node.c b/pinos/server/node.c index d7a3772af..a2c0dbaa4 100644 --- a/pinos/server/node.c +++ b/pinos/server/node.c @@ -687,7 +687,7 @@ pinos_node_constructed (GObject * obj) this, SPA_RESULT_RETURN_ASYNC (0), (PinosDeferFunc) init_complete, - this, + NULL, NULL); } node_register_object (this); @@ -1222,15 +1222,22 @@ remove_idle_timeout (PinosNode *node) } } -typedef struct { - PinosNode *node; - PinosNodeState state; -} StateData; - static void -on_state_complete (StateData *data) +on_state_complete (PinosNode *node, + gpointer data, + SpaResult res) { - pinos_node_update_state (data->node, data->state); + PinosNodeState state = GPOINTER_TO_INT (data); + + if (SPA_RESULT_IS_ERROR (res)) { + GError *error = NULL; + g_set_error (&error, + PINOS_ERROR, + PINOS_ERROR_NODE_STATE, + "error changing node state: %d", res); + pinos_node_report_error (node, error); + } else + pinos_node_update_state (node, state); } /** @@ -1248,7 +1255,6 @@ pinos_node_set_state (PinosNode *node, { PinosNodePrivate *priv; SpaResult res = SPA_RESULT_OK; - StateData data; g_return_val_if_fail (PINOS_IS_NODE (node), SPA_RESULT_INVALID_ARGUMENTS); priv = node->priv; @@ -1283,15 +1289,12 @@ pinos_node_set_state (PinosNode *node, if (SPA_RESULT_IS_ERROR (res)) return res; - data.node = node; - data.state = state; - pinos_main_loop_defer (priv->main_loop, node, res, (PinosDeferFunc) on_state_complete, - g_memdup (&data, sizeof (StateData)), - g_free); + GINT_TO_POINTER (state), + NULL); return res; } diff --git a/spa/include/spa/poll.h b/spa/include/spa/poll.h index 58cd0380c..df65152b2 100644 --- a/spa/include/spa/poll.h +++ b/spa/include/spa/poll.h @@ -61,7 +61,7 @@ typedef struct { unsigned int n_fds; } SpaPollNotifyData; -typedef int (*SpaPollNotify) (SpaPollNotifyData *data); +typedef SpaResult (*SpaPollNotify) (SpaPollNotifyData *data); /** * SpaPollItem: @@ -86,6 +86,14 @@ typedef struct { void *user_data; } SpaPollItem; + +typedef SpaResult (*SpaPollInvokeFunc) (SpaPoll *poll, + uint32_t seq, + size_t size, + void *data, + void *user_data); + + /** * SpaPoll: * @@ -122,11 +130,20 @@ struct _SpaPoll { SpaResult (*remove_item) (SpaPoll *poll, SpaPollItem *item); + + + SpaResult (*invoke) (SpaPoll *poll, + SpaPollInvokeFunc func, + uint32_t seq, + size_t size, + void *data, + void *user_data); }; #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__) +#define spa_poll_invoke(n,...) (n)->invoke((n),__VA_ARGS__) #ifdef __cplusplus } /* extern "C" */ diff --git a/spa/plugins/v4l2/v4l2-source.c b/spa/plugins/v4l2/v4l2-source.c index f68cc19a6..026c0e600 100644 --- a/spa/plugins/v4l2/v4l2-source.c +++ b/spa/plugins/v4l2/v4l2-source.c @@ -86,6 +86,7 @@ typedef struct { typedef struct { SpaLog *log; + SpaPoll *main_loop; SpaPoll *data_loop; bool export_buf; @@ -134,6 +135,8 @@ struct _SpaV4l2Source { SpaIDMap *map; SpaLog *log; + uint32_t seq; + SpaV4l2SourceProps props[2]; SpaNodeEventCallback event_cb; @@ -221,6 +224,73 @@ spa_v4l2_source_node_set_props (SpaNode *node, return res; } +static SpaResult +do_command_complete (SpaPoll *poll, + uint32_t seq, + size_t size, + void *data, + void *user_data) +{ + SpaV4l2Source *this = user_data; + SpaNodeEvent event; + + event.type = SPA_NODE_EVENT_TYPE_ASYNC_COMPLETE; + event.size = size; + event.data = data; + this->event_cb (&this->node, &event, this->user_data); + + return SPA_RESULT_OK; +} + +static SpaResult +do_start (SpaPoll *poll, + uint32_t seq, + size_t size, + void *data, + void *user_data) +{ + SpaV4l2Source *this = user_data; + SpaResult res; + SpaNodeEventAsyncComplete ac; + + res = spa_v4l2_start (this); + + ac.seq = seq; + ac.res = res; + spa_poll_invoke (this->state[0].main_loop, + do_command_complete, + seq, + sizeof (ac), + &ac, + this); + return res; +} + +static SpaResult +do_pause (SpaPoll *poll, + uint32_t seq, + size_t size, + void *data, + void *user_data) +{ + SpaV4l2Source *this = user_data; + SpaResult res; + SpaNodeEventAsyncComplete ac; + + res = spa_v4l2_pause (this); + + ac.seq = seq; + ac.res = res; + spa_poll_invoke (this->state[0].main_loop, + do_command_complete, + seq, + sizeof (ac), + &ac, + this); + return res; +} + + static SpaResult spa_v4l2_source_node_send_command (SpaNode *node, SpaNodeCommand *command) @@ -246,7 +316,12 @@ spa_v4l2_source_node_send_command (SpaNode *node, if (state->n_buffers == 0) return SPA_RESULT_NO_BUFFERS; - return spa_v4l2_start (this); + return spa_poll_invoke (this->state[0].data_loop, + do_start, + ++this->seq, + 0, + NULL, + this); } case SPA_NODE_COMMAND_PAUSE: { @@ -258,7 +333,12 @@ spa_v4l2_source_node_send_command (SpaNode *node, if (state->n_buffers == 0) return SPA_RESULT_NO_BUFFERS; - return spa_v4l2_pause (this); + return spa_poll_invoke (this->state[0].data_loop, + do_pause, + ++this->seq, + 0, + NULL, + this); } case SPA_NODE_COMMAND_FLUSH: case SPA_NODE_COMMAND_DRAIN: @@ -841,6 +921,8 @@ 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__MainLoop) == 0) + this->state[0].main_loop = support[i].data; else if (strcmp (support[i].uri, SPA_POLL__DataLoop) == 0) this->state[0].data_loop = support[i].data; } @@ -848,6 +930,10 @@ v4l2_source_init (const SpaHandleFactory *factory, spa_log_error (this->log, "an id-map is needed"); return SPA_RESULT_ERROR; } + if (this->state[0].main_loop == NULL) { + spa_log_error (this->log, "a main_loop 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;