GstAggregator doesn't consume buffers

[Alex A]

Hello,
I'm trying to write gstreamer plugin for `RGBA bitmap -> DVB Subtitles`
encoding. First of all I wrote simple stub: element just emulate encoding by
producing static dvb subtitles from file on every input buffer.
If run this pipeline:

gst-launch gst-launch-1.0 souphttpsrc location=http://localhost/live-stream.ts is-live=1 ! tsdemux name=dem mpegtsmux name=mux \
dem. ! application/x-teletext ! queue max-size-buffers=0 max-size-bytes=0 max-size-time=0 name=squeue ! teletextdec page=888 ! video/x-raw ! dvbsubenc ! mux.sink_1 \
mux. ! filesink location=out.ts

then result is fine: DVB-Inspector sees subtitles and it's valid.
But if add another stream to muxer (e.g. audio or video):

gst-launch-1.0 souphttpsrc location=http://localhost/live-stream.ts is-live=1 ! tsdemux name=dem mpegtsmux name=mux \
dem. ! video/x-h264 ! queue max-size-buffers=0 max-size-bytes=0 max-size-time=0 name=vqueue ! h264parse ! mux.sink_0 \
dem. ! application/x-teletext ! teletextdec page=888 ! dvbsubenc ! queue max-size-buffers=0 max-size-bytes=0 max-size-time=0 name=squeue ! mux.sink_1 \
mux. ! filesink location=out.ts

then the resulting file will not contain subtitles. Even in the Program Map
Table there are no entry for it.

Logs of `queue` between `dvbsubenc` and `mpegtsmux`:

0:00:00.816812187   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:00.816860078   853 0x55e8051e7190 LOG                    queue gstqueue.c:539:update_time_level:<squeue> update src time
0:00:00.816888470   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:00.000000000, src +0:00:00.000000000
0:00:00.816931152   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:539:update_time_level:<squeue> update src time
0:00:00.816997921   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:00.000000000, src +0:00:00.000000000
0:00:04.408493799   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:04.408521925   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:03.514554213, src +0:00:00.000000000
0:00:04.408646997   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:539:update_time_level:<squeue> update src time
0:00:04.408763052   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:03.514554213, src +0:00:03.514554213
0:00:10.161072943   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:10.161132247   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:09.286613009, src +0:00:03.514554213
0:00:10.161189567   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:539:update_time_level:<squeue> update src time
0:00:10.161230089   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:09.286613009, src +0:00:09.286613009
0:00:10.292462209   853 0x55e8051e7190 LOG                    queue gstqueue.c:1064:gst_queue_handle_sink_query:<squeue> queuing query 0x7fcf68002590 (allocation)
0:00:10.293333882   853 0x55e8051e7190 DEBUG                  queue gstqueue.c:580:apply_segment:<squeue> configured SEGMENT time segment start=0:00:00.000000000, offset=0:00:00.000000000, stop=99:99:99.999999999, rate=1.000000, applied_rate=1.000000, flags=0x00, time=0:00:00.000000000, base=0:00:00.000000000, position 0:00:00.000000000, duration 99:99:99.999999999
0:00:10.293375229   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:10.293419398   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:00.000000000, src +0:00:09.286613009
0:00:10.293466298   853 0x55e8051e7190 DEBUG                  queue gstqueue.c:580:apply_segment:<squeue> configured SEGMENT time segment start=0:00:00.000000000, offset=0:00:00.000000000, stop=99:99:99.999999999, rate=1.000000, applied_rate=1.000000, flags=0x00, time=0:00:00.000000000, base=0:00:00.000000000, position 0:00:00.000000000, duration 99:99:99.999999999
0:00:10.293551157   853 0x55e8051e7190 LOG                    queue gstqueue.c:539:update_time_level:<squeue> update src time
0:00:10.293632466   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:00.000000000, src +0:00:00.000000000
0:00:10.295299959   853 0x55e8051e7190 LOG                    queue gstqueue.c:635:apply_buffer:<squeue> sink position updated to 0:00:00.000000000
0:00:10.295328730   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:10.295354059   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:00.000000000, src +0:00:00.000000000
0:00:10.295462767   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:635:apply_buffer:<squeue> src position updated to 0:00:00.000000000
0:00:10.295505976   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:539:update_time_level:<squeue> update src time
0:00:10.295555260   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:00.000000000, src +0:00:00.000000000
0:00:14.335817600   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:14.335913758   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:13.454008457, src +0:00:00.000000000
0:00:14.335989388   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:539:update_time_level:<squeue> update src time
0:00:14.336141285   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:13.454008457, src +0:00:13.454008457
0:00:15.319799842   853 0x55e8051e7190 LOG                    queue gstqueue.c:635:apply_buffer:<squeue> sink position updated to 0:00:13.454008457
0:00:15.319863188   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:15.319888409   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:13.454008457, src +0:00:13.454008457
0:00:15.319971354   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:635:apply_buffer:<squeue> src position updated to 0:00:13.454008457
0:00:15.320013580   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:539:update_time_level:<squeue> update src time
0:00:15.320111136   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:13.454008457, src +0:00:13.454008457
0:00:18.308842013   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:18.308898804   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:17.468358437, src +0:00:13.454008457
0:00:18.308963483   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:539:update_time_level:<squeue> update src time
0:00:18.309028759   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:17.468358437, src +0:00:17.468358437
0:00:20.258750852   853 0x55e8051e7190 LOG                    queue gstqueue.c:635:apply_buffer:<squeue> sink position updated to 0:00:17.468358437
0:00:20.258784938   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:20.258810684   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:17.468358437, src +0:00:17.468358437
0:00:20.258898746   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:635:apply_buffer:<squeue> src position updated to 0:00:17.468358437
0:00:20.258939754   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:539:update_time_level:<squeue> update src time
0:00:20.258973426   853 0x55e8050d4ed0 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:17.468358437, src +0:00:17.468358437
0:00:24.479322296   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:24.479430972   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:23.582504406, src +0:00:17.468358437
0:00:25.296684171   853 0x55e8051e7190 LOG                    queue gstqueue.c:635:apply_buffer:<squeue> sink position updated to 0:00:23.582504406
0:00:25.296716960   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:25.296748639   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:23.582504406, src +0:00:17.468358437
0:00:28.556264311   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:28.556307322   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:27.684730766, src +0:00:17.468358437
0:00:30.268321319   853 0x55e8051e7190 LOG                    queue gstqueue.c:635:apply_buffer:<squeue> sink position updated to 0:00:27.684730766
0:00:30.268359055   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:30.268394402   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:27.684730766, src +0:00:17.468358437
0:00:34.501528505   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:34.501844773   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:33.599100023, src +0:00:17.468358437
0:00:35.291422014   853 0x55e8051e7190 LOG                    queue gstqueue.c:635:apply_buffer:<squeue> sink position updated to 0:00:33.599100023
0:00:35.291448868   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:35.291470922   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:33.599100023, src +0:00:17.468358437
0:00:38.472689712   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:38.472746774   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:37.613727456, src +0:00:17.468358437
0:00:40.287136216   853 0x55e8051e7190 LOG                    queue gstqueue.c:635:apply_buffer:<squeue> sink position updated to 0:00:37.613727456
0:00:40.287172395   853 0x55e8051e7190 LOG                    queue gstqueue.c:530:update_time_level:<squeue> update sink time
0:00:40.287200342   853 0x55e8051e7190 LOG                    queue gstqueue.c:548:update_time_level:<squeue> sink +0:00:37.613727456, src +0:00:17.468358437

On a src pad, time is only updated for first few buffers. It looks like
mpegtsmux (more precisely, GstAggregator) blocks `queue:src` thread and
`queue` can't push data. Information about that this stream is sparse, as
well as gap events, are supplied by the demuxer (pad with teletext) and just
forwarded by other elements to `mpegtsmux`.

In GstAggregator logs:

0:00:08.413923203   912 0x55cf6a40c5e0 DEBUG             aggregator gstaggregator.c:1624:gst_aggregator_default_sink_event:<mux:sink_1> Eating event: segment event: 0x7f8a4035c340, time 99:99:99.999999999, seq-num 61, GstEventSegment, segment=(GstSegment)"GstSegment, flags=(GstSegmentFlags)GST_SEGMENT_FLAG_NONE, rate=(double)1, applied-rate=(double)1, format=(GstFormat)time, base=(guint64)0, offset=(guint64)0, start=(guint64)0, stop=(guint64)18446744073709551615, time=(guint64)0, position=(guint64)0, duration=(guint64)18446744073709551615;";
0:00:08.413949065   912 0x55cf6a40c5e0 LOG               aggregator gstaggregator.c:850:gst_aggregator_do_events_and_queries:<mux:sink_1> Signaling buffer consumed from thread 0x55cf6a40c5e0
0:00:08.413993045   912 0x55cf6a40c5e0 LOG               aggregator gstaggregator.c:493:gst_aggregator_check_pads_ready:<mux:sink_1> pad not ready to be aggregated yet
0:00:08.414790452   912 0x55cf6a2f9ed0 DEBUG             aggregator gstaggregator.c:2888:gst_aggregator_pad_chain_internal:<mux:sink_1> Done chaining
0:00:08.418916670   912 0x7f8a50002e80 DEBUG             aggregator gstaggregator.c:2099:gst_aggregator_event_forward_func:<mux:sink_1> return of event push is 1

"Done chaining" prints only for first few buffers, and then:

0:00:18.395623508   912 0x55cf6a2f9ed0 DEBUG             aggregator gstaggregator.c:2830:gst_aggregator_pad_chain_internal:<mux:sink_1> Waiting for buffer to be consumed
0:00:18.395670089   912 0x55cf6a2f9ed0 LOG               aggregator gstaggregator.c:2833:gst_aggregator_pad_chain_internal:<mux:sink_1> Waiting for buffer to be consumed thread 0x55cf6a2f9ed0
0:00:18.397272236   912 0x55cf6a40c5e0 LOG               aggregator gstaggregator.c:493:gst_aggregator_check_pads_ready:<mux:sink_1> pad not ready to be aggregated yet

Line "pad not ready to be consumed yet" are repeated endlessly. And after
that, the time on the queue src pad ceases to be updated.

I am using the latest version of Gstreamer compiled from source code. Can anyone explain what is happening and what am I doing wrong?
Code and file with subtitles:
gstdvbsubenc.h
gstdvbsubenc.c
dvb-dump.bin