diff --git a/spa/include/spa/event.h b/spa/include/spa/event.h
index 7a156ff0b..8d9c72604 100644
--- a/spa/include/spa/event.h
+++ b/spa/include/spa/event.h
@@ -75,12 +75,14 @@ struct _SpaEvent {
  * @events: events to watch for
  * @revents: result events
  * @callback: callback called when there was activity on @fd
+ * @user_data: user data to pass to @callback
  */
 typedef struct {
   int   fd;
   short events;
   short revents;
   SpaNotify callback;
+  void     *user_data;
 } SpaEventPoll;
 
 #ifdef __cplusplus
diff --git a/spa/meson.build b/spa/meson.build
index 105aafc8f..40e6894c8 100644
--- a/spa/meson.build
+++ b/spa/meson.build
@@ -5,6 +5,7 @@ v4l2_dep = dependency('libv4l2')
 xv_dep = dependency('x11')
 sdl_dep = dependency('sdl2')
 dl_lib = find_library('dl', required : true)
+pthread_lib = find_library('pthread', required : true)
 
 inc = include_directories('include')
 
diff --git a/spa/plugins/v4l2/v4l2-source.c b/spa/plugins/v4l2/v4l2-source.c
index 912470caa..cb3808cab 100644
--- a/spa/plugins/v4l2/v4l2-source.c
+++ b/spa/plugins/v4l2/v4l2-source.c
@@ -22,7 +22,6 @@
 #include <sys/stat.h>
 #include <fcntl.h>
 
-#include <pthread.h>
 #include <linux/videodev2.h>
 
 #include <spa/node.h>
@@ -47,11 +46,6 @@ reset_v4l2_source_props (SpaV4l2SourceProps *props)
 
 #define MAX_BUFFERS     256
 
-typedef struct {
-  SpaEventPoll poll;
-  SpaV4l2Source *source;
-} V4l2EventPoll;
-
 typedef struct _V4l2Buffer V4l2Buffer;
 
 struct _V4l2Buffer {
@@ -72,11 +66,10 @@ typedef struct {
   struct v4l2_format fmt;
   enum v4l2_buf_type type;
   struct v4l2_requestbuffers reqbuf;
-  pthread_t thread;
-  bool running;
   V4l2Buffer buffers[MAX_BUFFERS];
   V4l2Buffer *ready;
   uint32_t ready_count;
+  SpaEventPoll poll;
 } SpaV4l2State;
 
 struct _SpaV4l2Source {
@@ -94,7 +87,6 @@ struct _SpaV4l2Source {
 
   SpaPortInfo info;
   SpaPortStatus status;
-
 };
 
 #include "v4l2-utils.c"
diff --git a/spa/plugins/v4l2/v4l2-utils.c b/spa/plugins/v4l2/v4l2-utils.c
index 6c2d08ec3..c47acffb4 100644
--- a/spa/plugins/v4l2/v4l2-utils.c
+++ b/spa/plugins/v4l2/v4l2-utils.c
@@ -356,8 +356,7 @@ mmap_read (SpaV4l2Source *this)
 static void
 v4l2_on_fd_events (void *user_data)
 {
-  V4l2EventPoll *poll = user_data;
-  SpaV4l2Source *this = poll->source;
+  SpaV4l2Source *this = user_data;
   SpaEvent event;
 
   mmap_read (this);
@@ -371,45 +370,6 @@ v4l2_on_fd_events (void *user_data)
   this->event_cb (&this->handle, &event, this->user_data);
 }
 
-static void *
-v4l2_loop (void *user_data)
-{
-  SpaV4l2Source *this = user_data;
-  SpaV4l2State *state = &this->state;
-  struct pollfd fds[1];
-  SpaEvent event;
-  int r;
-
-  event.notify = NULL;
-  event.type = SPA_EVENT_TYPE_CAN_PULL_OUTPUT;
-  event.port_id = 0;
-  event.size = 0;
-  event.data = NULL;
-
-  fds[0].fd = state->fd;
-  fds[0].events = POLLIN | POLLPRI | POLLERR;
-  fds[0].revents = 0;
-
-  while (state->running) {
-    r = poll (fds, 1, 2000);
-    if (r < 0) {
-      if (errno == EINTR)
-        continue;
-      break;
-    }
-    if (r == 0) {
-      fprintf (stderr, "select timeout\n");
-      break;
-    }
-
-    if (mmap_read (this) < 0)
-     break;
-
-    event.refcount = 1;
-    this->event_cb (&this->handle, &event, this->user_data);
-  }
-  return NULL;
-}
 static void
 v4l2_buffer_free (void *data)
 {
@@ -540,10 +500,8 @@ static int
 spa_v4l2_start (SpaV4l2Source *this)
 {
   SpaV4l2State *state = &this->state;
-  int err;
   enum v4l2_buf_type type;
   SpaEvent event;
-  V4l2EventPoll poll;
 
   if (spa_v4l2_open (this) < 0)
     return -1;
@@ -562,14 +520,14 @@ spa_v4l2_start (SpaV4l2Source *this)
   event.notify = NULL;
   event.type = SPA_EVENT_TYPE_ADD_POLL;
   event.port_id = 0;
-  event.data = &poll.poll;
-  event.size = sizeof (poll.poll);
+  event.data = &state->poll;
+  event.size = sizeof (state->poll);
 
-  poll.poll.fd = state->fd;
-  poll.poll.events = POLLIN | POLLPRI | POLLERR;
-  poll.poll.revents = 0;
-  poll.poll.callback = v4l2_on_fd_events;
-  poll.source = this;
+  state->poll.fd = state->fd;
+  state->poll.events = POLLIN | POLLPRI | POLLERR;
+  state->poll.revents = 0;
+  state->poll.callback = v4l2_on_fd_events;
+  state->poll.user_data = this;
   this->event_cb (&this->handle, &event, this->user_data);
 
   type = V4L2_BUF_TYPE_VIDEO_CAPTURE;
@@ -577,13 +535,7 @@ spa_v4l2_start (SpaV4l2Source *this)
     perror ("VIDIOC_STREAMON");
     return -1;
   }
-
-  state->running = true;
-  if ((err = pthread_create (&state->thread, NULL, v4l2_loop, this)) != 0) {
-    printf ("can't create thread: %d %s", err, strerror (err));
-    state->running = false;
-  }
-  return err;
+  return 0;
 }
 
 static int
@@ -591,11 +543,7 @@ spa_v4l2_stop (SpaV4l2Source *this)
 {
   SpaV4l2State *state = &this->state;
   enum v4l2_buf_type type;
-
-  if (state->running) {
-    state->running = false;
-    pthread_join (state->thread, NULL);
-  }
+  SpaEvent event;
 
   type = V4L2_BUF_TYPE_VIDEO_CAPTURE;
   if (xioctl (state->fd, VIDIOC_STREAMOFF, &type) < 0) {
@@ -603,6 +551,15 @@ spa_v4l2_stop (SpaV4l2Source *this)
     return -1;
   }
 
+  event.refcount = 1;
+  event.notify = NULL;
+  event.type = SPA_EVENT_TYPE_REMOVE_POLL;
+  event.port_id = 0;
+  event.data = &state->poll;
+  event.size = sizeof (state->poll);
+  this->event_cb (&this->handle, &event, this->user_data);
+
   spa_v4l2_close (this);
+
   return 0;
 }
diff --git a/spa/tests/meson.build b/spa/tests/meson.build
index d7da85d14..0c5ba4364 100644
--- a/spa/tests/meson.build
+++ b/spa/tests/meson.build
@@ -5,5 +5,5 @@ executable('test-mixer', 'test-mixer.c',
 
 executable('test-v4l2', 'test-v4l2.c',
            include_directories : inc,
-           dependencies : [dl_lib, sdl_dep],
+           dependencies : [dl_lib, sdl_dep, pthread_lib],
            install : false)
diff --git a/spa/tests/test-v4l2.c b/spa/tests/test-v4l2.c
index a809002ae..6bd840db0 100644
--- a/spa/tests/test-v4l2.c
+++ b/spa/tests/test-v4l2.c
@@ -22,6 +22,9 @@
 #include <stdlib.h>
 #include <unistd.h>
 #include <dlfcn.h>
+#include <poll.h>
+#include <pthread.h>
+#include <errno.h>
 
 #include <SDL2/SDL.h>
 
@@ -34,6 +37,9 @@ typedef struct {
   SDL_Renderer *renderer;
   SDL_Window *window;
   SDL_Texture *texture;
+  bool running;
+  pthread_t thread;
+  SpaEventPoll poll;
 } AppData;
 
 static SpaResult
@@ -121,6 +127,9 @@ on_source_event (SpaHandle *handle, SpaEvent *event, void *user_data)
       }
       break;
     }
+    case SPA_EVENT_TYPE_ADD_POLL:
+      memcpy (&data->poll, event->data, sizeof (SpaEventPoll));
+      break;
     default:
       printf ("got event %d\n", event->type);
       break;
@@ -187,17 +196,52 @@ negotiate_formats (AppData *data)
   return SPA_RESULT_OK;
 }
 
+static void *
+loop (void *user_data)
+{
+  AppData *data = user_data;
+  struct pollfd fds[1];
+  int r;
+
+  fds[0].fd = data->poll.fd;
+  fds[0].events = data->poll.events;
+  fds[0].revents = 0;
+
+  printf ("enter thread\n");
+  while (data->running) {
+    r = poll (fds, 1, 2000);
+    if (r < 0) {
+      if (errno == EINTR)
+        continue;
+      break;
+    }
+    if (r == 0) {
+      fprintf (stderr, "select timeout\n");
+      break;
+    }
+    data->poll.callback (data->poll.user_data);
+  }
+  printf ("leave thread\n");
+  return NULL;
+}
+
 static void
 run_async_source (AppData *data)
 {
   SpaResult res;
   SpaCommand cmd;
   bool done = false;
+  int err;
 
   cmd.type = SPA_COMMAND_START;
   if ((res = data->source_node->send_command (data->source, &cmd)) < 0)
     printf ("got error %d\n", res);
 
+  data->running = true;
+  if ((err = pthread_create (&data->thread, NULL, loop, data)) != 0) {
+    printf ("can't create thread: %d %s", err, strerror (err));
+    data->running = false;
+  }
 
   while (!done) {
     SDL_Event event;
@@ -214,7 +258,11 @@ run_async_source (AppData *data)
           break;
       }
     }
+  }
 
+  if (data->running) {
+    data->running = false;
+    pthread_join (data->thread, NULL);
   }
 
   cmd.type = SPA_COMMAND_STOP;