Fix behaviour with PadProbeReturn.REMOVE for new and old probe types; tidy up tests and documentation.

neilcsmith-net · neilcsmith-net · commit dd6213f7f838 · 2020-11-02T16:36:32.000Z
diff --git a/src/org/freedesktop/gstreamer/Pad.java b/src/org/freedesktop/gstreamer/Pad.java
@@ -431,6 +431,12 @@ public PadProbeReturn callback(Pad pad, GstPadProbeInfo probeInfo, Pointer user_
                 PadProbeInfo info = new PadProbeInfo(probeInfo);
                 PadProbeReturn ret = callback.probeCallback(pad, info);
                 info.invalidate();
+                if (ret == PadProbeReturn.REMOVE) {
+                    // don't want handle to try and remove in GCallback::disconnect
+                    // @TODO move to Map<PROBE, NativeLong> of probes over callback
+                    handle.probes.remove(probeInfo.id);
+                    removeCallback(PROBE.class, callback);
+                }
                 return ret;
             }
         };
@@ -468,16 +474,20 @@ public void addEventProbe(final EVENT_PROBE listener) {
     synchronized void addEventProbe(final EVENT_PROBE listener, final int mask) {
         final GstPadAPI.PadProbeCallback probe = new GstPadAPI.PadProbeCallback() {
             public PadProbeReturn callback(Pad pad, GstPadProbeInfo probeInfo, Pointer user_data) {
-//        	    System.out.println("CALLBACK " + probeInfo.padProbeType);
                 if ((probeInfo.padProbeType & mask) != 0) {
                     Event event = null;
                     if ((probeInfo.padProbeType & EVENT_HAS_INFO_MASK) != 0) {
                         event = GSTPAD_API.gst_pad_probe_info_get_event(probeInfo);
                     }
-                    return listener.eventReceived(pad, event);
+                    PadProbeReturn ret = listener.eventReceived(pad, event);
+                    if (ret == PadProbeReturn.REMOVE) {
+                        // don't want handle to try and remove in GCallback::disconnect
+                        handle.probes.remove(probeInfo.id);
+                        removeCallback(EVENT_PROBE.class, listener);
+                    }
+                    return ret;
                 }
 
-                //We have to negate the return value to keep consistency with gstreamer's API
                 return PadProbeReturn.OK;
             }
         };
@@ -508,10 +518,15 @@ public synchronized void addDataProbe(final DATA_PROBE listener) {
             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);
+                    PadProbeReturn ret = listener.dataReceived(pad, buffer);
+                    if (ret == PadProbeReturn.REMOVE) {
+                        // don't want handle to try and remove in GCallback::disconnect
+                        handle.probes.remove(probeInfo.id);
+                        removeCallback(DATA_PROBE.class, listener);
+                    }
+                    return ret;
                 }
 
-                //We have to negate the return value to keep consistency with gstreamer's API
                 return PadProbeReturn.OK;
             }
         };
@@ -742,25 +757,27 @@ public static interface PROBE {
     }
 
     /**
-     * Signal emitted when an event passes through this <tt>Pad</tt>.
+     * Probe for listening when an event passes through this Pad.
      *
      * @see #addEventProbe(EVENT_PROBE)
      * @see #removeEventProbe(EVENT_PROBE)
      */
     public static interface EVENT_PROBE {
 
         public PadProbeReturn eventReceived(Pad pad, Event event);
+        
     }
 
     /**
-     * Signal emitted when new data is available on the {@link Pad}
+     * Probe for listening when new data is available on the Pad.
      *
      * @see #addDataProbe(DATA_PROBE)
      * @see #removeDataProbe(DATA_PROBE)
      */
     public static interface DATA_PROBE {
 
         public PadProbeReturn dataReceived(Pad pad, Buffer buffer);
+        
     }
 
     private static class Handle extends GstObject.Handle {
diff --git a/test/org/freedesktop/gstreamer/PadProbeTypeTest.java b/test/org/freedesktop/gstreamer/PadProbeTypeTest.java
@@ -21,17 +21,10 @@
 import java.util.Set;
 import org.freedesktop.gstreamer.glib.NativeFlags;
 import org.freedesktop.gstreamer.lowlevel.GstPadAPI;
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
 import org.junit.Test;
 
 import static org.junit.Assert.*;
 
-/**
- *
- */
 public class PadProbeTypeTest {
     
 
diff --git a/test/org/freedesktop/gstreamer/PadTest.java b/test/org/freedesktop/gstreamer/PadTest.java
@@ -1,5 +1,5 @@
 /* 
- * Copyright (c) 2019 Neil C Smith
+ * Copyright (c) 2020 Neil C Smith
  * Copyright (c) 2007 Wayne Meissner
  * 
  * This file is part of gstreamer-java.
@@ -27,6 +27,7 @@
 import static org.junit.Assert.assertTrue;
 
 import java.lang.ref.WeakReference;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import org.freedesktop.gstreamer.event.FlushStopEvent;
 
@@ -35,7 +36,6 @@
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 
 /**
@@ -127,6 +127,40 @@ public PadProbeReturn eventReceived(Pad pad, Event event) {
         assertNotSame("event_prober.probeEvent() should not have been called", ev2, e.get());
     }
     
+    @Test
+    public void addEventProbe_Remove() {
+        Element elem = ElementFactory.make("identity", "src");
+        Event ev = new TagEvent(new TagList());
+
+        Pad sink = elem.getStaticPad("sink");
+
+        final AtomicReference<Event> e = new AtomicReference<Event>();
+
+        Pad.EVENT_PROBE event_probe = new Pad.EVENT_PROBE() {
+
+            public PadProbeReturn eventReceived(Pad pad, Event event) {
+                e.set(event);
+                return PadProbeReturn.REMOVE;
+            }
+        };
+
+        sink.setActive(true);
+        sink.sendEvent(new FlushStopEvent());
+
+        sink.addEventProbe(event_probe);
+        sink.sendEvent(ev);
+        assertEquals("event_prober.probeEvent() was not called", ev, e.get());
+
+        Event ev2 = new TagEvent(new TagList());
+        sink.sendEvent(ev2);
+        assertNotSame("event_prober.probeEvent() should not have been called", ev2, e.get());
+        
+        WeakReference<Pad.EVENT_PROBE> probeRef = new WeakReference<>(event_probe);
+        event_probe = null;
+        assertTrue("Removed probe not collected", GCTracker.waitGC(probeRef));
+        
+    }
+    
     @Test
     public void addProbe_Event() {
         Element elem = ElementFactory.make("identity", "src");
@@ -156,6 +190,42 @@ public void addProbe_Event() {
         sink.sendEvent(ev2);
         assertNotSame("Probe (Event) should not have been called", ev2, e.get());
     }
+    
+    @Test
+    public void addProbe_EventRemove() {
+        Element elem = ElementFactory.make("identity", "src");
+        Event ev = new TagEvent(new TagList());
+
+        Pad sink = elem.getStaticPad("sink");
+
+        final AtomicReference<Event> e = new AtomicReference<>();
+
+        Pad.PROBE probe = (Pad pad, PadProbeInfo info) -> {
+            assertTrue("Info type does not include event downstream",
+                    info.getType().contains(PadProbeType.EVENT_DOWNSTREAM));
+            e.set(info.getEvent());
+            return PadProbeReturn.REMOVE;
+        };
+
+        sink.setActive(true);
+        sink.sendEvent(new FlushStopEvent());
+
+        sink.addProbe(PadProbeType.EVENT_BOTH, probe);
+        sink.sendEvent(ev);
+        assertEquals("Probe (Event) was not called", ev, e.get());
+
+        Event ev2 = new TagEvent(new TagList());
+        sink.sendEvent(ev2);
+        assertNotSame("Probe (Event) should not have been called", ev2, e.get());
+        
+        WeakReference<Pad.PROBE> probeRef = new WeakReference<>(probe);
+        probe = null;
+        assertTrue("Removed probe not collected", GCTracker.waitGC(probeRef));
+        
+        Event ev3 = new TagEvent(new TagList());
+        sink.sendEvent(ev3);
+        assertNotSame("Probe (Event) should not have been called", ev3, e.get());
+    }
 
     @Test
     public void addDataProbe() {
@@ -229,4 +299,29 @@ public void addProbe_Data() {
         assertNotSame("Probe (Data) should not have been called", buf2, b.get());
     }
     
+    @Test
+    public void addProbe_Idle() {
+
+        Element elem = ElementFactory.make("identity", "src");
+        final AtomicBoolean called = new AtomicBoolean();
+
+        Pad src = elem.getStaticPad("src");
+
+        Pad.PROBE probe = (Pad pad, PadProbeInfo info) -> {
+            called.set(true);
+            return PadProbeReturn.REMOVE;
+        };
+        
+        src.addProbe(PadProbeType.IDLE, probe);
+        
+        assertTrue("Idle probe not called", called.get());
+        
+        WeakReference<Pad.PROBE> probeRef = new WeakReference<>(probe);
+        
+        probe = null;
+        
+        assertTrue("Idle probe not collected", GCTracker.waitGC(probeRef));
+        
+    }
+    
 }
