audrey/src/playbin.vala

class Playbin : Object {
    // dynamic: undocumented vala feature
    // lets us access the about-to-finish signal
    private dynamic Gst.Element playbin;

    private SourceFunc? async_done_callback;

    // cubic: recommended for media player volume sliders?
    public double volume {
        get {
            return ((Gst.Audio.StreamVolume) this.playbin).get_volume (Gst.Audio.StreamVolumeFormat.CUBIC);
        }
        set {
            ((Gst.Audio.StreamVolume) this.playbin).set_volume (Gst.Audio.StreamVolumeFormat.CUBIC, value);
        }
    }

    public bool mute {
        get { return playbin.mute; }
        set { playbin.mute = value; }
    }

    public signal void set_position (int64 position);
    public int64 duration { get; private set; }

    public signal void stream_started ();
    public signal void stream_over ();

    construct {
        this.playbin = Gst.ElementFactory.make ("playbin3", null);
        assert (this.playbin != null);

        //dynamic Gst.Element souphttpsrc = ((Gst.Bin) this.playbin).get_by_name ("souphttpsrc");
        //assert (souphttpsrc != null);
        //souphttpsrc.user_agent = "Wavelet/0.1.0 (Linux)"; // WAVELET_VERSION

        // regularly update position
        Timeout.add (500, () => {
            int64 new_position;
            if (this.playbin.query_position (Gst.Format.TIME, out new_position)) {
                this.set_position (new_position < this.duration ? new_position : this.duration);
            }

            // keep rerunning
            return true;
        });

        this.playbin.get_bus ().add_watch (Priority.DEFAULT, (bus, message) => {
            // message.type actually seems to be flags
            if (Gst.MessageType.ERROR in message.type) {
                Error err;
                string? debug;
                message.parse_error (out err, out debug);
                warning ("gst playbin bus error: %s", err.message);
            }

            if (Gst.MessageType.ASYNC_DONE in message.type) {
                assert (this.async_done_callback != null);
                var cb = (owned) this.async_done_callback;
                assert (this.async_done_callback == null); // sanity check
                cb ();
            }

            if (Gst.MessageType.STREAM_START in message.type) {
                int64 new_duration;
                assert (this.playbin.query_duration (Gst.Format.TIME, out new_duration));
                this.duration = new_duration;

                string? next_uri = null;

                this.next_uri_lock.lock ();
                next_uri = this.next_uri;

                if (next_uri != (string) this.playbin.current_uri) {
                    this.next_uri_lock.unlock ();
                    // WHOOPS! didn't actually switch to the track the play queue wanted
                    // we can still fix this though
                    assert (next_uri != null);
                    this.playbin.set_state (Gst.State.READY);
                    this.playbin.uri = next_uri;
                    this.playbin.set_state (Gst.State.PLAYING);
                    // no one will ever know
                } else {
                    this.next_uri = null;
                    this.next_uri_lock.unlock ();

                    this.stream_started ();
                }
            }

            if (Gst.MessageType.EOS in message.type) {
                string next_uri;

                this.next_uri_lock.lock ();
                next_uri = this.next_uri;
                this.next_uri_lock.unlock ();

                if (next_uri == null) {
                    // no next track was arranged, we're done
                    this.stream_over ();
                }
            }

            return true;
        });

        this.playbin.about_to_finish.connect (this.on_about_to_finish);
    }

    private async void set_state (Gst.State state) {
        assert (this.async_done_callback == null);

        switch (this.playbin.set_state (state)) {
        case Gst.StateChangeReturn.SUCCESS:
            break;

        case Gst.StateChangeReturn.ASYNC:
            this.async_done_callback = this.set_state.callback;
            yield;
            break;

        default:
            assert (false);
            break;
        }
    }

    public async bool seek (int64 position) {
        // don't actually seek if an operation is pending
        if (this.async_done_callback != null) {
            return false;
        }

        // ASSUMPTION: this can only work asynchronously
        // (will wait for an ASYNC_DONE in the bus)
        assert (this.playbin.seek_simple (Gst.Format.TIME, Gst.SeekFlags.FLUSH | Gst.SeekFlags.KEY_UNIT, position));
        this.async_done_callback = this.seek.callback;
        yield;

        return true;
    }

    private string? play_now_queued;
    // returns true if the "play now" request wasn't overriden by a later call
    public async bool play_now (string uri) {
        if (this.async_done_callback != null) {
            // an operation was already pending
            // last writer wins here
            this.play_now_queued = uri;
            // idle spinning is probably fine
            while (this.async_done_callback != null) {
                Idle.add (this.play_now.callback);
                yield;
                // only keep spinning if we can still win
                if (this.play_now_queued != uri) {
                    return false;
                }
            }
            this.play_now_queued = null;
        }

        // pretend this track was locked in by about-to-finish before
        this.next_uri_lock.lock ();
        this.next_uri = uri;
        this.next_uri_lock.unlock ();

        yield this.set_state (Gst.State.READY);
        this.playbin.uri = uri;
        yield this.set_state (Gst.State.PLAYING);

        if (this.play_now_queued != null) {
            // another "play now" was queued while we were busy
            // defer to it
            return false;
        }

        return true;
    }

    Mutex next_uri_lock;
    string? next_uri;

    public void set_next_uri (string? next_uri) {
        this.next_uri_lock.lock ();
        this.next_uri = next_uri;
        this.next_uri_lock.unlock ();
    }

    // called when uri can be switched for gapless playback
    // need async queue because this might be called from a gstreamer thread
    private void on_about_to_finish (dynamic Gst.Element playbin) {
        this.next_uri_lock.lock ();
        string? next_uri = this.next_uri;
        this.next_uri_lock.unlock ();

        if (next_uri != null) {
            playbin.uri = next_uri;
        }
    }

    public void pause () {
        this.playbin.set_state (Gst.State.PAUSED);
    }

    public void play () {
        this.playbin.set_state (Gst.State.PLAYING);
    }
}