[gst-devel] split/reassemble a stream into fixed-size buffers

[Antoine Pitrou]

Hi Andy,

Le lundi 26 février 2007 à 12:04 +0100, Andy Wingo a écrit :
> On Thu, 2007-02-22 at 13:56 +0100, Antoine Pitrou wrote:
> > In my application I have to end up with buffers of a fixed size (these
> > are PCMU audio buffers), for reasons I don't have control over.
> ...
> > Is there a built-in element to actually recompose a stream into
> > fixed-size buffers, or do I have to write a custom element using a
> > gst.Adapter ?
> 
> Your suspicions are correct :) You have to write a custom element, and
> using gstadapter is a good idea.

Thanks for your answer !
So I did it in Python, but the problem is gst.Adapter throws some
(seemingly random) error messages about refcount errors:

(nosetests:13339): GStreamer-CRITICAL **: gst_mini_object_unref: assertion `mini_object->refcount > 0' failed

The workaround I found is to always build a copy of the incoming buffer,
but of course it negates the advantage of using gst.Adapter which is to
not make a copy every time. The code for my element follows, if I remove
the "buf = buf.copy()" line in the chain function I get those warning
messages. Is there anything wrong in my code?

(I'm not that much bothered by the inefficiency, since I'm doing it in
Python anyway, but I'd like to learn and know whether I'm doing things
wrong :-))

Regards

Antoine.



class Chunker(gst.Element):
    """
    This element recomposes a stream into a stream of fixed-size buffers,
    splitting and reassembling them as necessary.
    """

    _sinkpadtemplate = gst.PadTemplate("sink",
                                       gst.PAD_SINK,
                                       gst.PAD_ALWAYS,
                                       gst.caps_new_any())

    _srcpadtemplate = gst.PadTemplate("src",
                                      gst.PAD_SRC,
                                      gst.PAD_ALWAYS,
                                      gst.caps_new_any())

    def __init__(self, chunk_size):
        self.__gobject_init__()

        if chunk_size <= 0:
            raise ValueError("bad chunk size %s" % chunk_size)
        self.chunk_size = chunk_size

        self.sink_pad = gst.Pad(self._sinkpadtemplate, "sink")
        self.add_pad(self.sink_pad)
        self.src_pad = gst.Pad(self._srcpadtemplate, "src")
        self.add_pad(self.src_pad)

        self.sink_pad.set_chain_function(self.chain_function)
        self.sink_pad.set_event_function(self.event_function)

        self.adapter = gst.Adapter()

    def event_function(self, pad, event):
        if event.type in (gst.EVENT_NEWSEGMENT, gst.EVENT_EOS, gst.EVENT_FLUSH_STOP):
            self._clear()
        return pad.event_default(event)

    def chain_function(self, pad, buf):
        buf = buf.copy()
        self.adapter.push(buf)
        while self.adapter.available() >= self.chunk_size:
            ret = self.src_pad.push(self.adapter.take_buffer(self.chunk_size))
            if ret != gst.FLOW_OK:
                return ret
        return gst.FLOW_OK

    def _clear(self):
        self.adapter.clear()


gobject.type_register(Chunker)