add data probe

src/org/freedesktop/gstreamer/Pad.java

@@ -391,6 +391,17 @@ public static interface EVENT_PROBE {
         public PadProbeReturn eventReceived(Pad pad, Event event);
     }
 
+    /**
+     * Signal emitted when new data is available on the {@link Pad}
+     * 
+     * @see #addDataProbe(DATA_PROBE) 
+     * @see #removeDataProbe(DATA_PROBE) 
+     */
+    public static interface DATA_PROBE {
+        public PadProbeReturn dataReceived(Pad pad, Buffer buffer);
+    }
+
+
     /**
      * Add a listener for the <code>have-data</code> signal on this {@link Pad}
      * 
@@ -518,6 +529,36 @@ protected void disconnect() {
     public void removeEventProbe(EVENT_PROBE listener) {
         removeCallback(EVENT_PROBE.class, listener);
     }
+
+
+    public synchronized void addDataProbe(final DATA_PROBE listener) {
+
+        final GstPadAPI.PadProbeCallback probe = new GstPadAPI.PadProbeCallback() {
+            public PadProbeReturn callback(Pad pad, GstPadProbeInfo probeInfo, Pointer user_data) {
+                if ((probeInfo.padProbeType & GstPadAPI.GST_PAD_PROBE_TYPE_BUFFER) != 0) {
+                    Buffer buffer = GSTPAD_API.gst_pad_probe_info_get_buffer(probeInfo);
+                    return listener.dataReceived(pad, buffer);
+                }
+
+            	//We have to negate the return value to keep consistency with gstreamer's API
+                return PadProbeReturn.OK;
+            }
+        };
+
+        GCallback cb = new GCallback(GSTPAD_API.gst_pad_add_probe(this, GstPadAPI.GST_PAD_PROBE_TYPE_BUFFER, probe, null, null), probe) {
+            @Override
+            protected void disconnect() {
+                GSTPAD_API.gst_pad_remove_probe(Pad.this, id);                
+            }
+        };        
+
+        addCallback(DATA_PROBE.class, listener, cb);       
+    }
+
+    public void removeDataProbe(DATA_PROBE listener) {
+        removeCallback(DATA_PROBE.class, listener);
+    }
+
 
     /**
      * Sends the event to this pad. 
@@ -612,4 +653,29 @@ public FlowReturn getRange(long offset, int size, Buffer[] buffer) {
     public FlowReturn pullRange(long offset, int size, Buffer[] buffer) {
     	return GSTPAD_API.gst_pad_pull_range(this, offset, size, buffer);
     }
+
+
+
+    /**
+     * Pushes a buffer to the peer of pad .
+     * This function will call installed block probes before triggering any 
+     * installed data probes.
+     * 
+     * The function proceeds calling gst_pad_chain() on the peer pad and returns 
+     * the value from that function. If pad has no peer, GST_FLOW_NOT_LINKED 
+     * will be returned.
+     * 
+     * In all cases, success or failure, the caller loses its reference to 
+     * buffer after calling this function.
+     *
+     * @param buffer
+     *       the GstBuffer to push returns GST_FLOW_ERROR if not. [transfer full]
+     * @return 
+     *       a GstFlowReturn from the peer pad.
+     * 
+     * MT safe.
+     */
+    public FlowReturn push(final Buffer buffer) {
+        return GSTPAD_API.gst_pad_push(this, buffer);
+    }
 }

src/org/freedesktop/gstreamer/lowlevel/GstPadAPI.java

@@ -158,6 +158,8 @@ NativeLong gst_pad_add_probe(Pad pad, int mask, PadProbeCallback callback,
 
     Event gst_pad_probe_info_get_event(GstPadProbeInfo probeInfo);
 
+    Buffer gst_pad_probe_info_get_buffer(GstPadProbeInfo probeInfo);
+
 //    NativeLong /* gulong */ gst_pad_add_data_probe(Pad pad, PadDataProbe handler, Pointer data);
 //
 //    void gst_pad_remove_data_probe(Pad pad, NativeLong handler_id);

test/org/freedesktop/gstreamer/PadTest.java

@@ -121,4 +121,42 @@ public PadProbeReturn eventReceived(Pad pad, Event event) {
         sink.sendEvent(ev2);
         assertNotSame("event_prober.probeEvent() should not have been called", ev2, e.get());
     }
+
+
+    @Test
+    public void addDataProbe() {
+
+        Element elem = ElementFactory.make("identity", "src");
+        TagList taglist = new TagList();
+        Buffer buf = new Buffer(3);
+        Buffer buf2 = new Buffer(2);
+        final AtomicReference<Buffer> b = new AtomicReference<Buffer>();
+
+        Pad src = elem.getStaticPad("src");
+
+        Pad.DATA_PROBE data_probe = new Pad.DATA_PROBE() {
+
+            @Override
+            public PadProbeReturn dataReceived(Pad pad, Buffer buffer) {
+                b.set(buffer);
+                return PadProbeReturn.OK;
+            }
+        };
+
+        elem.play();
+
+        // add a dataprobe
+        src.addDataProbe(data_probe);
+
+        // push data
+        FlowReturn res = src.push(buf);
+        assertEquals("event_prober.probeEvent() was not called", buf, b.get());
+
+        // remove the dataprobe
+        src.removeDataProbe(data_probe);       
+
+        // push data
+        res = src.push(buf2);        
+        assertNotSame("event_prober.probeEvent() should not have been called", buf2, b.get());
+    }
 }