Wrap timed-pop-filtered into a synchronizable event

gstreamer/bus.rkt

@@ -0,0 +1,45 @@
+#lang racket/base
+
+(require ffi/unsafe/introspection
+         racket/contract
+         racket/place
+         "gst.rkt")
+
+(provide (contract-out [make-bus-channel
+                        (->* (gobject?)
+                             ((listof symbol?)
+                              #:timeout exact-nonnegative-integer?)
+                             evt?)]))
+
+(define bus% (gst 'Bus))
+
+(define message% (gst 'Message))
+
+(define clock-time-none ((gst 'CLOCK_TIME_NONE)))
+
+(define (make-bus-channel bus [filters '(any)]
+                          #:timeout [timeout clock-time-none])
+  (define bus-pipe
+    (place chan
+           (let*-values ([(bus-ptr timeout filters)
+                          (apply values (place-channel-get chan))]
+                         [(bus) (gobject-cast bus-ptr bus%)])
+             (let loop ()
+               (define msg
+                 (send bus timed-pop-filtered timeout filters))
+               (place-channel-put chan (and msg
+                                            (gtype-instance-pointer msg)))
+               (loop)))))
+  (place-channel-put bus-pipe (list (gtype-instance-pointer bus)
+                                    timeout
+                                    filters))
+  ;; (define bus-pipe (make-channel))
+  ;; (thread
+  ;;  (let loop ()
+  ;;    (define msg
+  ;;      (send bus timed-pop-filtered timeout filters))
+  ;;    (channel-put bus-pipe (and msg
+  ;;                               (gtype-instance-pointer msg)))
+  ;;    (loop)))
+  (wrap-evt bus-pipe (lambda (ptr) (and ptr
+                                   (gstruct-cast ptr message%)))))

gstreamer/gst.rkt

@@ -0,0 +1,7 @@
+#lang racket/base
+
+(require ffi/unsafe/introspection)
+
+(provide gst)
+
+(define gst (introspection 'Gst))

gstreamer/main.rkt

@@ -2,11 +2,11 @@
 
 (require ffi/unsafe
          ffi/unsafe/introspection
-         racket/place)
+         racket/place
+         "gst.rkt"
+         "bus.rkt")
 
-(provide gst)
-
-(define gst (introspection 'Gst))
+(provide main)
 
 (let-values ([(initialized? argc argv) ((gst 'init_check) 0 #f)])
   (if initialized?
@@ -41,15 +41,9 @@
 (define (main)
   (send playbin set-state 'playing)
   (define pipe
-    (place chan
-           (define bus (gobject-cast (place-channel-get chan) (gst 'Bus)))
-           (let loop ()
-             (define msg
-               (send bus timed-pop-filtered (* 100 millisecond) '(eos error state-changed duration-changed)))
-             (and msg
-                  (place-channel-put chan (get-field type msg)))
-             (loop))))
-  (place-channel-put pipe (gtype-instance-pointer (send playbin get-bus)))
+    (make-bus-channel (send playbin get-bus)))
   (thread (lambda () (let loop ()
-                  (println (place-channel-get pipe))
-                  (loop)))))
+                  (define msg (sync pipe))
+                  (println (get-field type msg))
+                  (unless (memf (lambda (x) (memq x '(eos error))) (get-field type msg))
+                    (loop))))))