Switch to CountDownLatch in SampleTester and BufferProbeTester.

Author: neilcsmith-net

test/org/freedesktop/gstreamer/BufferProbeTester.java

@@ -23,18 +23,18 @@
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.fail;
 
-import java.util.ArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
 
-import org.freedesktop.gstreamer.glib.GError;
-
 /**
  * Utility class for unit testing API that operates on a Buffer.
  * <p>
  * Call {@link BufferTester#test(Consumer)} and pass a callback which will
- * perform the test on a Buffer it is supplied. The callback runs on the
- * AppSink.NEW_SAMPLE thread. The sample is produced by a simple, ephemeral
- * pipeline that is fed by a video test source.
+ * perform the test on a Buffer it is supplied. The callback runs in a Pad data
+ * probe. The buffer is produced by a simple, ephemeral pipeline that is fed by
+ * a video test source.
  */
 public class BufferProbeTester {
 
@@ -49,29 +49,26 @@ public static void test(Consumer<Buffer> callback, String pipelineDescription) {
     public static void test(Consumer<Buffer> callback, String pipelineDescription, int skipFrames) {
         assertNotNull("Pipeline description can not be null", pipelineDescription);
         assertFalse("Pipeline description can not be empty", pipelineDescription.isEmpty());
-        ArrayList<GError> errors = new ArrayList<>();
-        Bin bin = Gst.parseBinFromDescription(pipelineDescription, false, errors);
-        assertNotNull("Unable to create Bin from pipeline description: ", bin);
+        Pipeline pipe = (Pipeline) Gst.parseLaunch(pipelineDescription);
+        assertNotNull("Unable to create Pipeline from pipeline description: ", pipe);
 
-        Element sink = bin.getElementByName("sink");
+        Element sink = pipe.getElementByName("sink");
         Pad pad = sink.getStaticPad("sink");
         BufferProbe probe = new BufferProbe(callback, skipFrames);
         pad.addDataProbe(probe);
 
-        bin.play();
+        pipe.play();
 
         // Wait for the sample to arrive and for the client supplied test function to
         // complete
         try {
-            synchronized (probe) {
-                probe.wait();
-            }
-        } catch (InterruptedException e) {
-            fail("Unexpected interruption waiting for sample");
+            probe.await(5000);
+        } catch (Exception ex) {
+            fail("Unexpected exception waiting for buffer\n" + ex);
+        } finally {
+            pipe.stop();
         }
 
-        bin.stop();
-
         // If the test threw an exception on the sample listener thread, throw it here
         // (on the main thread)
         if (probe.exception != null) {
@@ -81,8 +78,10 @@ public static void test(Consumer<Buffer> callback, String pipelineDescription, i
 
     private static class BufferProbe implements Pad.DATA_PROBE {
 
-        private Consumer<Buffer> callback;
         private final int skipFrames;
+        private final CountDownLatch latch;
+        private final Consumer<Buffer> callback;
+
         private Throwable exception;
         private int counter = 0;
 
@@ -93,11 +92,12 @@ private static class BufferProbe implements Pad.DATA_PROBE {
         BufferProbe(Consumer<Buffer> callback, int skip) {
             this.callback = callback;
             skipFrames = skip;
+            latch = new CountDownLatch(1);
         }
 
         @Override
         public PadProbeReturn dataReceived(Pad pad, Buffer buffer) {
-            if (callback != null) {
+            if (latch.getCount() > 0) {
                 if (counter < skipFrames) {
                     counter++;
                     return PadProbeReturn.OK;
@@ -109,15 +109,18 @@ public PadProbeReturn dataReceived(Pad pad, Buffer buffer) {
                     } catch (Throwable exc) {
                         exception = exc;
                     }
-                    callback = null;
                 } finally {
-                    synchronized (this) {
-                        notify();
-                    }
+                    latch.countDown();
                 }
             }
             return PadProbeReturn.OK;
         }
+
+        void await(long millis) throws InterruptedException, TimeoutException {
+            if (!latch.await(millis, TimeUnit.MILLISECONDS)) {
+                throw new TimeoutException();
+            }
+        }
     }
 
 }

test/org/freedesktop/gstreamer/SampleTest.java

@@ -17,16 +17,14 @@
  * You should have received a copy of the GNU Lesser General Public License
  * along with gstreamer-java.  If not, see <http://www.gnu.org/licenses/>.
  */
-
 package org.freedesktop.gstreamer;
 
-import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.*;
 
 import org.freedesktop.gstreamer.glib.Natives;
 import org.freedesktop.gstreamer.util.TestAssumptions;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 
 public class SampleTest {
@@ -36,59 +34,71 @@ public SampleTest() {
 
     @BeforeClass
     public static void setUpClass() throws Exception {
-    	Gst.init(Gst.getVersion(), "SampleTest");
+        Gst.init(Gst.getVersion(), "SampleTest");
     }
-    
+
     @AfterClass
     public static void tearDownClass() throws Exception {
         Gst.deinit();
     }
 
     @Test
-    @Ignore
-    // @TODO for some reason this test is unreliable on GitHub Actions.
     public void testGetCaps() {
-    	SampleTester.test((Sample sample) -> {
-    		Caps caps = sample.getCaps();
-    		Structure struct = caps.getStructure(0);
-    		String name = struct.getName();
-    		assertEquals("video/x-raw", name);
-    	});
+        SampleTester.test((Sample sample) -> {
+            Caps caps = sample.getCaps();
+            Structure struct = caps.getStructure(0);
+            String name = struct.getName();
+            assertEquals("video/x-raw", name);
+        });
     }
 
     @Test
     public void testGetBuffer() {
-    	SampleTester.test((Sample sample) -> {
-    		Buffer buffer = sample.getBuffer();
-    		assertEquals(1, buffer.getMemoryCount());
-    	});
+        SampleTester.test((Sample sample) -> {
+            Buffer buffer = sample.getBuffer();
+            assertEquals(1, buffer.getMemoryCount());
+        });
+    }
+
+    @Test
+    public void testSetBuffer() {
+        // since gst 1.16, the sample is recycled and keep a reference on the last buffer received
+        TestAssumptions.requireGstVersion(1, 16);
+
+        SampleTester.test((Sample sample) -> {
+
+            Buffer buffer = sample.getBuffer();
+
+            int refCount = buffer.getRefCount();
+
+            assertEquals(2, sample.getRefCount());
+
+            // make sample writable
+            Natives.unref(sample);
+
+            // force sample to release the buffer
+            sample.setBuffer(null);
+
+            Natives.ref(sample);
+
+            assertEquals(2, sample.getRefCount());
+
+            assertEquals(refCount - 1, buffer.getRefCount());
+        });
     }
-    
+
     @Test
-    public void testSetBuffer() {    	
-    	// since gst 1.16, the sample is recycled and keep a reference on the last buffer received
-		TestAssumptions.requireGstVersion(1, 16);
-
-    	SampleTester.test((Sample sample) -> {
-    		
-    		Buffer buffer = sample.getBuffer();
-
-    		int refCount = buffer.getRefCount();
-    		    		
-    		assertEquals(2, sample.getRefCount());
-
-    		// make sample writable
-    		Natives.unref(sample);
-    		
-    		// force sample to release the buffer
-    		sample.setBuffer(null);
-    		
-    		Natives.ref(sample);
-    		
-    		assertEquals(2, sample.getRefCount());
-    		
-    		assertEquals(refCount-1, buffer.getRefCount());
-    	});
+    public void testSampleTester() {
+        try {
+            SampleTester.test(sample -> {
+                throw new IllegalStateException();
+            });
+        } catch (Throwable t) {
+            assertTrue(t instanceof AssertionError);
+            assertTrue(t.getCause() instanceof IllegalStateException);
+            return;
+        }
+        fail("No exception thrown");
     }
-    
+
 }