gst/gstobject.override: don't try and unref objects that are already finalizing

Original commit message from CVS:

* gst/gstobject.override:
don't try and unref objects that are already finalizing
* gst/gstpad.override:
fix up the set_chainfunc method
* testsuite/common.py:
* testsuite/test_pad.py:
add linked/unlinked tests with no/true/false bufferprobes

ChangeLog

@@ -1,3 +1,13 @@
+2005-10-05  Thomas Vander Stichele  <thomas at apestaart dot org>
+
+	* gst/gstobject.override:
+	  don't try and unref objects that are already finalizing
+	* gst/gstpad.override:
+	  fix up the set_chainfunc method
+	* testsuite/common.py:
+	* testsuite/test_pad.py:
+	  add linked/unlinked tests with no/true/false bufferprobes
+
 2005-10-05  Thomas Vander Stichele  <thomas at apestaart dot org>
 
 	* codegen/argtypes.py:

gst/gstobject.override

@@ -123,11 +123,18 @@ _wrap_gst_object_tp_dealloc(PyGObject *self)
 
     /* if we're a GstObject, we want to monkeypatch the GObject.tp_dealloc's
      * g_object_unref and "replace" it with a gst_object_unref */
-    if (! GST_IS_OBJECT (obj))
+    if (GST_IS_OBJECT (obj)) {
+        GST_DEBUG_OBJECT (obj, "gst.Object.tp_dealloc, go rc %d, gsto rc %d",
+            obj->ref_count, GST_OBJECT_REFCOUNT_VALUE (obj));
+	if (GST_OBJECT_REFCOUNT_VALUE (obj) == 0) {
+            /* already being finalized, can't resurrect object */
+            obj = NULL;
+            self->obj = NULL;
+        } else {
+          g_object_ref (obj);
+        }
+    } else {
         obj = NULL;
-    else {
-        GST_DEBUG_OBJECT (obj, "gst.Object.tp_dealloc");
-        g_object_ref (obj);
     }
 
     PyGObject_Type.tp_dealloc((PyObject *) self);

gst/gstpad.override

@@ -254,24 +254,20 @@ handle_chain_function_exception (GValue *ret, guint n, const GValue *params)
 }
 
 static GstFlowReturn
-call_chain_function(GstPad *pad, GstBuffer *data)
+call_chain_function(GstPad *pad, GstBuffer *buffer)
 {
     GClosure *closure;
     GValue ret = { 0, };
     GValue args[2] = { { 0, }, { 0, } };
     GstFlowReturn    flow;
     
-    g_value_init (&ret, G_TYPE_ENUM);
+    g_value_init (&ret, GST_TYPE_FLOW_RETURN);
     g_value_set_enum (&ret, GST_FLOW_ERROR);
     g_value_init (&args[0], GST_TYPE_PAD);
-    if (GST_IS_BUFFER (data)) {
+    g_value_init (&args[1], GST_TYPE_BUFFER);
-      g_value_init (&args[1], GST_TYPE_BUFFER);
-    } else if (GST_IS_EVENT (data)) {
-      g_value_init (&args[1], GST_TYPE_EVENT);
-    }
 
     g_value_set_object (&args[0], pad);
-    g_value_take_boxed (&args[1], data);
+    gst_value_set_mini_object (&args[1], GST_MINI_OBJECT (buffer));
     closure = pad_private(pad)->chain_function;
     
     g_closure_invoke (closure, &ret, 2, args, NULL);
@@ -281,6 +277,9 @@ call_chain_function(GstPad *pad, GstBuffer *data)
     g_value_unset (&args[0]);
     g_value_unset (&args[1]);
 
+    /* a chain function takes over the ref of the buffer handed to it;
+     * so we should unref after calling the pythonic chain func */
+    gst_buffer_unref (buffer);
     return flow;
 }
 
@@ -291,6 +290,7 @@ _wrap_gst_pad_set_chain_function(PyGObject *self,
 {
     SET_PAD_CLOSURE (self, args, kwargs, chain_function)
 }
+
 %%
 override gst_pad_set_event_function kwargs
 

testsuite/common.py

@@ -145,6 +145,7 @@ class TestCase(unittest.TestCase):
             new.extend([o for o in objs if o not in self._tracked[c]])
 
         self.failIf(new, new)
+        #self.failIf(new, ["%r:%d" % (type(o), id(o)) for o in new])
         del self._tracked
 
     def setUp(self):

testsuite/test_pad.py

@@ -39,6 +39,109 @@ class PadTemplateTest(TestCase):
         self.assertEquals(sys.getrefcount(template), 3)
         #self.assertEquals(template.__gstrefcount__, 1)
 
+class PadPushUnlinkedTest(TestCase):
+    def setUp(self):
+        self.gctrack()
+        self.src = gst.Pad("src", gst.PAD_SRC)
+        self.sink = gst.Pad("sink", gst.PAD_SINK)
+
+    def tearDown(self):
+        self.assertEquals(sys.getrefcount(self.src), 3)
+        self.assertEquals(self.src.__gstrefcount__, 1)
+        del self.src
+        self.assertEquals(sys.getrefcount(self.sink), 3)
+        self.assertEquals(self.sink.__gstrefcount__, 1)
+        del self.sink
+        self.gccollect()
+        self.gcverify()
+
+    def testNoProbe(self):
+        self.buffer = gst.Buffer()
+        self.assertEquals(self.buffer.__grefcount__, 1)
+        self.assertEquals(self.src.push(self.buffer), gst.FLOW_NOT_LINKED)
+        # pushing it takes a ref in the python wrapper to keep buffer
+        # alive afterwards; but the core unrefs the ref it receives
+        self.assertEquals(self.buffer.__grefcount__, 1)
+
+    def testFalseProbe(self):
+        id = self.src.add_buffer_probe(self._probe_handler, False)
+        self.buffer = gst.Buffer()
+        self.assertEquals(self.buffer.__grefcount__, 1)
+        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
+        self.assertEquals(self.buffer.__grefcount__, 1)
+        self.src.remove_buffer_probe(id)
+
+    def testTrueProbe(self):
+        id = self.src.add_buffer_probe(self._probe_handler, True)
+        self.buffer = gst.Buffer()
+        self.assertEquals(self.buffer.__grefcount__, 1)
+        self.assertEquals(self.src.push(self.buffer), gst.FLOW_NOT_LINKED)
+        self.assertEquals(self.buffer.__grefcount__, 1)
+        self.src.remove_buffer_probe(id)
+
+    def _probe_handler(self, pad, buffer, ret):
+        return ret
+
+class PadPushLinkedTest(TestCase):
+    def setUp(self):
+        self.gctrack()
+        self.src = gst.Pad("src", gst.PAD_SRC)
+        self.sink = gst.Pad("sink", gst.PAD_SINK)
+        caps = gst.caps_from_string("foo/bar")
+        self.src.set_caps(caps)
+        self.sink.set_caps(caps)
+        self.sink.set_chain_function(self._chain_func)
+        self.src.link(self.sink)
+        self.buffers = []
+
+    def tearDown(self):
+        self.assertEquals(sys.getrefcount(self.src), 3)
+        self.assertEquals(self.src.__gstrefcount__, 1)
+        del self.src
+        self.assertEquals(sys.getrefcount(self.sink), 3)
+        self.assertEquals(self.sink.__gstrefcount__, 1)
+        del self.sink
+        self.gccollect()
+        self.gcverify()
+
+    def _chain_func(self, pad, buffer):
+        self.buffers.append(buffer)
+
+        return gst.FLOW_OK
+
+    def testNoProbe(self):
+        self.buffer = gst.Buffer()
+        self.assertEquals(self.buffer.__grefcount__, 1)
+        gst.debug('pushing buffer on linked pad, no probe')
+        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
+        gst.debug('pushed buffer on linked pad, no probe')
+        # pushing it takes a ref in the python wrapper to keep buffer
+        # alive afterwards; fakesink will get the buffer
+        self.assertEquals(self.buffer.__grefcount__, 1)
+        self.assertEquals(len(self.buffers), 1)
+
+    def testFalseProbe(self):
+        id = self.src.add_buffer_probe(self._probe_handler, False)
+        self.buffer = gst.Buffer()
+        self.assertEquals(self.buffer.__grefcount__, 1)
+        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
+        self.assertEquals(self.buffer.__grefcount__, 1)
+        self.src.remove_buffer_probe(id)
+        self.assertEquals(len(self.buffers), 0)
+
+    def testTrueProbe(self):
+        id = self.src.add_buffer_probe(self._probe_handler, True)
+        self.buffer = gst.Buffer()
+        self.assertEquals(self.buffer.__grefcount__, 1)
+        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
+        self.assertEquals(self.buffer.__grefcount__, 1)
+        self.src.remove_buffer_probe(id)
+        self.assertEquals(len(self.buffers), 1)
+
+    def _probe_handler(self, pad, buffer, ret):
+        return ret
+        
+ 
 class PadTest(TestCase):
     def setUp(self):
         self.gctrack()
@@ -86,7 +189,7 @@ class PadPipelineTest(TestCase):
 #        assert self.srcpad.query(gst.QUERY_POSITION, gst.FORMAT_TIME) == 0
 
 
-class PadProbeTest(TestCase):
+class PadProbePipeTest(TestCase):
     def setUp(self):
         self.gctrack()
         self.pipeline = gst.Pipeline()
@@ -101,6 +204,8 @@ class PadProbeTest(TestCase):
         self.pipeline.add(self.fakesrc, self.fakesink)
         self.assertEquals(self.fakesrc.__gstrefcount__, 2) # added
         self.assertEquals(sys.getrefcount(self.fakesrc), 3)
+        self.assertEquals(self.fakesink.__gstrefcount__, 2) # added
+        self.assertEquals(sys.getrefcount(self.fakesink), 3)
 
         self.fakesrc.link(self.fakesink)
 
@@ -108,48 +213,64 @@ class PadProbeTest(TestCase):
         self.assertEquals(sys.getrefcount(self.pipeline), 3)
         self.assertEquals(self.fakesrc.__gstrefcount__, 2)
         self.assertEquals(sys.getrefcount(self.fakesrc), 3)
+        self.assertEquals(self.fakesink.__gstrefcount__, 2)
+        self.assertEquals(sys.getrefcount(self.fakesink), 3)
 
     def tearDown(self):
         self.assertEquals(self.pipeline.__gstrefcount__, 1)
         self.assertEquals(sys.getrefcount(self.pipeline), 3)
         self.assertEquals(self.fakesrc.__gstrefcount__, 2)
         self.assertEquals(sys.getrefcount(self.fakesrc), 3)
+        self.assertEquals(self.fakesink.__gstrefcount__, 2)
+        self.assertEquals(sys.getrefcount(self.fakesink), 3)
         gst.debug('deleting pipeline')
         del self.pipeline
         self.gccollect()
 
         self.assertEquals(self.fakesrc.__gstrefcount__, 1) # parent gone
+        self.assertEquals(self.fakesink.__gstrefcount__, 1) # parent gone
         self.assertEquals(sys.getrefcount(self.fakesrc), 3)
+        self.assertEquals(sys.getrefcount(self.fakesink), 3)
         gst.debug('deleting fakesrc')
         del self.fakesrc
         self.gccollect()
+        gst.debug('deleting fakesink')
         del self.fakesink
         self.gccollect()
 
         self.gcverify()
         
-    def testFakeSrcProbeOnce(self):
+    def testFakeSrcProbeOnceKeep(self):
         self.fakesrc.set_property('num-buffers', 1)
 
+        self.fakesink.set_property('signal-handoffs', True)
+        self.fakesink.connect('handoff', self._handoff_callback_fakesink)
+
         pad = self.fakesrc.get_pad('src')
         id = pad.add_buffer_probe(self._probe_callback_fakesrc)
         self._got_fakesrc_buffer = 0
+        self._got_fakesink_buffer = 0
         self.pipeline.set_state(gst.STATE_PLAYING)
         while not self._got_fakesrc_buffer:
             pass
+        while not self._got_fakesink_buffer:
+            pass
+
+        self.assertEquals(self._got_fakesink_buffer, 1)
+        pad.remove_buffer_probe(id)
 
         self.pipeline.set_state(gst.STATE_NULL)
-        pad.remove_buffer_probe (id)
 
     def testFakeSrcProbeMany(self):
         self.fakesrc.set_property('num-buffers', 1000)
 
         pad = self.fakesrc.get_pad('src')
-        pad.add_buffer_probe(self._probe_callback_fakesrc)
+        id = pad.add_buffer_probe(self._probe_callback_fakesrc)
         self._got_fakesrc_buffer = 0
         self.pipeline.set_state(gst.STATE_PLAYING)
         while not self._got_fakesrc_buffer == 1000:
             pass
+        pad.remove_buffer_probe(id)
 
         self.pipeline.set_state(gst.STATE_NULL)
 
@@ -157,6 +278,13 @@ class PadProbeTest(TestCase):
         self.failUnless(isinstance(pad, gst.Pad))
         self.failUnless(isinstance(buffer, gst.Buffer))
         self._got_fakesrc_buffer += 1
+        return True
+
+    def _handoff_callback_fakesink(self, sink, buffer, pad):
+        self.failUnless(isinstance(buffer, gst.Buffer))
+        self.failUnless(isinstance(pad, gst.Pad))
+        self._got_fakesink_buffer += 1
+        return True
 
     def testRemovingProbe(self):
         self.fakesrc.set_property('num-buffers', 10)