[rygel] core: Pause pipeline when buffer grows too much

[Zeeshan Ali Khattak <zeeshanak src gnome org>]

commit 69ab246a87fd8b0f9acab949cbd4e97ed50e4bfa
Author: Zeeshan Ali (Khattak) <zeeshanak gnome org>
Date:   Mon May 3 01:29:37 2010 +0300

    core: Pause pipeline when buffer grows too much
    
    There is no way of knowing that an HTTP client has paused the playback but
    we do know it when its not reading from socket anymore. So now we keep
    track of number of buffers queued to be sent to the client and if this
    number goes beyond a fixed amount, we pause the gstreamer pipeline until
    client catches up and there is only a few number of queued buffers left.

 src/rygel/rygel-live-response.vala |   28 ++++++++++++++++++++++++++++
 1 files changed, 28 insertions(+), 0 deletions(-)
---
diff --git a/src/rygel/rygel-live-response.vala b/src/rygel/rygel-live-response.vala
index bb27040..c6c24c6 100644
--- a/src/rygel/rygel-live-response.vala
+++ b/src/rygel/rygel-live-response.vala
@@ -28,6 +28,9 @@ using Gst;
 
 internal class Rygel.LiveResponse : Rygel.HTTPResponse {
     private const string SINK_NAME = "fakesink";
+    // High and low threshold for number of buffered chunks
+    private const uint MAX_BUFFERED_CHUNKS = 32;
+    private const uint MIN_BUFFERED_CHUNKS = 4;
 
     private Pipeline pipeline;
 
@@ -35,6 +38,9 @@ internal class Rygel.LiveResponse : Rygel.HTTPResponse {
 
     private SourceFunc run_continue;
 
+    private size_t buffered;
+    private bool out_of_sync;
+
     public LiveResponse (Soup.Server  server,
                          Soup.Message msg,
                          string       name,
@@ -47,9 +53,14 @@ internal class Rygel.LiveResponse : Rygel.HTTPResponse {
 
         this.prepare_pipeline (name, src);
         this.time_range = time_range;
+
+        this.buffered = 0;
+        this.out_of_sync = false;
     }
 
     public override async void run () {
+        this.msg.wrote_chunk.connect (this.on_wrote_chunk);
+
         // Only bother attempting to seek if the offset is greater than zero.
         if (this.time_range != null && this.time_range.start > 0) {
             this.pipeline.set_state (State.PAUSED);
@@ -64,6 +75,7 @@ internal class Rygel.LiveResponse : Rygel.HTTPResponse {
 
     public override void end (bool aborted, uint status) {
         this.pipeline.set_state (State.NULL);
+        this.msg.wrote_chunk.disconnect (this.on_wrote_chunk);
 
         if (!aborted) {
             this.msg.response_body.complete ();
@@ -151,11 +163,27 @@ internal class Rygel.LiveResponse : Rygel.HTTPResponse {
         Idle.add_full (Priority.HIGH_IDLE,
                        () => {
             this.push_data (buffer.data, buffer.size);
+            this.buffered++;
+
+            if (this.buffered > MAX_BUFFERED_CHUNKS) {
+                // Client is either not reading (Paused) or not fast enough
+                this.pipeline.set_state (State.PAUSED);
+                this.out_of_sync = true;
+            }
 
             return false;
         });
     }
 
+    private void on_wrote_chunk (Soup.Message msg) {
+        this.buffered--;
+
+        if (out_of_sync && this.buffered < MIN_BUFFERED_CHUNKS) {
+            this.pipeline.set_state (State.PLAYING);
+            this.out_of_sync = false;
+        }
+    }
+
     private bool bus_handler (Gst.Bus     bus,
                               Gst.Message message) {
         bool ret = true;