- A bit of clean up to PulseOut

- Added volume support to PulseOut
- Tweaked Player and Transport to more accurately reflect current playback

src/contrib/pulseout/PulseOut.cpp

@@ -1,7 +1,7 @@
 #include "PulseOut.h"
 #include <iostream>
 
-#define BUFFER_COUNT 48 /* seems to work well? */
+#define BUFFER_COUNT 32 /* seems to work well? */
 
 using namespace musik::core::audio;
 
@@ -44,22 +44,7 @@ static inline long long bufferLengthMicroSeconds(IBuffer* buffer) {
     return (samples * 1000000) / rate;
 }
 
-size_t PulseOut::CountBuffersWithProvider(IBufferProvider* provider) {
-    boost::recursive_mutex::scoped_lock bufferLock(this->mutex);
-
-    size_t count = 0;
-    auto it = this->buffers.begin();
-    while (it != this->buffers.end()) {
-        if ((*it)->provider == provider) {
-            ++count;
-        }
-        ++it;
-    }
-
-    return count;
-}
-
-void PulseOut::NotifyBufferCompleted(BufferContext* context) {
+static inline void notifyBufferCompleted(PulseOut::BufferContext* context) {
     IBufferProvider* provider = context->provider;
     IBuffer* buffer = context->buffer;
     provider->OnBufferProcessed(buffer);
@@ -90,6 +75,67 @@ PulseOut::~PulseOut() {
 }
 
 
+void PulseOut::Destroy() {
+    delete this;
+}
+
+void PulseOut::Pause() {
+    this->SetPaused(true);
+}
+
+void PulseOut::Resume() {
+    this->SetPaused(false);
+}
+
+void PulseOut::SetVolume(double volume) {
+    this->volume = volume;
+
+    boost::recursive_mutex::scoped_lock bufferLock(this->mutex);
+
+    if (this->pulseStream) {
+        pa_cvolume vol;
+        pa_cvolume_set(&vol, 2, pa_sw_volume_from_linear(volume));
+
+        MainLoopLock loopLock(this->pulseMainLoop);
+        pa_context_set_sink_input_volume(
+            this->pulseContext,
+            pa_stream_get_index(this->pulseStream),
+            &vol,
+            NULL,
+            NULL);
+    }
+}
+
+void PulseOut::Stop() {
+    std::deque<std::shared_ptr<BufferContext> > toNotify;
+
+    {
+        boost::recursive_mutex::scoped_lock bufferLock(this->mutex);
+
+        if (this->pulseStream) {
+            MainLoopLock loopLock(this->pulseMainLoop);
+
+            /* notify outside of the critical section */
+            std::swap(this->buffers, toNotify);
+
+            waitForCompletion(
+                pa_stream_flush(
+                    this->pulseStream,
+                    &PulseOut::OnPulseStreamSuccessCallback,
+                    this),
+                this->pulseMainLoop);
+        }
+    }
+
+    /* all buffers are dead. notify the IBufferProvider */
+    auto it = toNotify.begin();
+    while (it != toNotify.end()) {
+        this->bufferQueueLength -= bufferLengthMicroSeconds((*it)->buffer);
+        notifyBufferCompleted((*it).get());
+        ++it;
+    }
+}
+
 bool PulseOut::Play(IBuffer *buffer, IBufferProvider *provider) {
     if (!this->pulseStream ||
         this->pulseStreamFormat.rate != buffer->SampleRate() ||
@@ -122,6 +168,7 @@ bool PulseOut::Play(IBuffer *buffer, IBufferProvider *provider) {
 
         long long bufferLength = bufferLengthMicroSeconds(buffer);
         //std::cerr << "PulseOut: bufferLength " << bufferLength << std::endl;
+        
         pa_usec_t currentTime;
         pa_stream_get_time(this->pulseStream, &currentTime);
         context->endTime = bufferLength + currentTime + this->bufferQueueLength;
@@ -143,7 +190,7 @@ bool PulseOut::Play(IBuffer *buffer, IBufferProvider *provider) {
 
     if (error) {
         std::cerr << "PulseOut: FAILED!! this should not happen." << error << std::endl;
-        this->NotifyBufferCompleted(context.get());
+        notifyBufferCompleted(context.get());
     }
     else {
         // std::cerr << "PulseOut: buffer enqueued!" << std::endl;
@@ -194,7 +241,7 @@ void PulseOut::ThreadProc() {
             /* actually notify here */
             auto it = toNotify.begin();
             while (it != toNotify.end()) {
-                this->NotifyBufferCompleted((*it).get());
+                notifyBufferCompleted((*it).get());
                 ++it;
             }
 
@@ -207,18 +254,6 @@ void PulseOut::ThreadProc() {
     std::cerr << "PulseOut: timing thread finished\n";
 }
 
-void PulseOut::Destroy() {
-    delete this;
-}
-
-void PulseOut::Pause() {
-    this->SetPaused(true);
-}
-
-void PulseOut::Resume() {
-    this->SetPaused(false);
-}
-
 void PulseOut::SetPaused(bool paused) {
     if (this->pulseStream) {
         MainLoopLock loopLock(this->pulseMainLoop);
@@ -232,39 +267,6 @@ void PulseOut::SetPaused(bool paused) {
     }
 }
 
-void PulseOut::SetVolume(double volume) {
-}
-
-void PulseOut::Stop() {
-    std::deque<std::shared_ptr<BufferContext> > toNotify;
-
-    {
-        boost::recursive_mutex::scoped_lock bufferLock(this->mutex);
-
-        if (this->pulseStream) {
-            MainLoopLock loopLock(this->pulseMainLoop);
-
-            /* notify outside of the critical section */
-            std::swap(this->buffers, toNotify);
-
-            waitForCompletion(
-                pa_stream_flush(
-                    this->pulseStream,
-                    &PulseOut::OnPulseStreamSuccessCallback,
-                    this),
-                this->pulseMainLoop);
-        }
-    }
-
-    /* all buffers are dead. notify the IBufferProvider */
-    auto it = toNotify.begin();
-    while (it != toNotify.end()) {
-        this->bufferQueueLength -= bufferLengthMicroSeconds((*it)->buffer);
-        this->NotifyBufferCompleted((*it).get());
-        ++it;
-    }
-}
-
 void PulseOut::OnPulseContextStateChanged(pa_context *context, void *data) {
     PulseOut* out = static_cast<PulseOut*>(data);
     const pa_context_state_t state = pa_context_get_state(context);
@@ -423,6 +425,7 @@ bool PulseOut::InitPulseStream(size_t rate, size_t channels) {
     }
 
     if (ready) {
+        this->SetVolume(this->volume);
         this->Resume();
     }
 
@@ -457,3 +460,19 @@ void PulseOut::DeinitPulse() {
         this->pulseMainLoop = NULL;
     }
 }
+
+
+size_t PulseOut::CountBuffersWithProvider(IBufferProvider* provider) {
+    boost::recursive_mutex::scoped_lock bufferLock(this->mutex);
+
+    size_t count = 0;
+    auto it = this->buffers.begin();
+    while (it != this->buffers.end()) {
+        if ((*it)->provider == provider) {
+            ++count;
+        }
+        ++it;
+    }
+
+    return count;
+}

src/contrib/pulseout/PulseOut.h

@@ -10,6 +10,13 @@
 
 class PulseOut : public musik::core::audio::IOutput {
     public:
+        struct BufferContext {
+            PulseOut *output;
+            musik::core::audio::IBuffer *buffer;
+            musik::core::audio::IBufferProvider *provider;
+            long long endTime;
+        };
+
         PulseOut();
         virtual ~PulseOut();
 
@@ -24,13 +31,6 @@ class PulseOut : public musik::core::audio::IOutput {
             musik::core::audio::IBufferProvider *provider);
 
     private:
-        struct BufferContext {
-            PulseOut *output;
-            musik::core::audio::IBuffer *buffer;
-            musik::core::audio::IBufferProvider *provider;
-            long long endTime;
-        };
-
         static void OnPulseContextStateChanged(pa_context *c, void *data);
         static void OnPulseStreamStateChanged(pa_stream *s, void *data);
         static void OnPulseStreamSuccessCallback(pa_stream *s, int success, void *data);
@@ -38,7 +38,6 @@ class PulseOut : public musik::core::audio::IOutput {
 
         void ThreadProc(); /* ugh shoot me */
 
-        void NotifyBufferCompleted(BufferContext *context);
         size_t CountBuffersWithProvider(musik::core::audio::IBufferProvider *provider);
 
         void InitPulse();

src/core/audio/Player.cpp

@@ -70,6 +70,7 @@ Player::Player(const std::string &url, OutputPtr output)
 , url(url)
 , currentPosition(0)
 , output(output)
+, notifiedStarted(false)
 , setPosition(-1) {
     musik::debug::info(TAG, "new instance created");
 
@@ -142,8 +143,6 @@ void Player::ThreadLoop() {
         bool finished = false;
         BufferPtr buffer;
 
-        bool startedPlaying = false;
-
         while (!finished && !this->Exited()) {
             /* see if we've been asked to seek since the last sample was
             played. if we have, clear our output buffer and seek the
@@ -187,11 +186,6 @@ void Player::ThreadLoop() {
             the output device. */
             if (buffer) {
                 if (this->output->Play(buffer.get(), this)) {
-                    if (!startedPlaying) {
-                        startedPlaying = true;
-                        this->PlaybackStarted(this);
-                    }
-
                     /* success! the buffer was accepted by the output.*/
                     boost::mutex::scoped_lock lock(this->mutex);
 
@@ -272,6 +266,9 @@ bool Player::Exited() {
 }
 
 void Player::OnBufferProcessed(IBuffer *buffer) {
+    bool started = false;
+
+    {
         boost::mutex::scoped_lock lock(this->mutex);
 
         /* removes the specified buffer from the list of locked buffers, and also
@@ -294,7 +291,21 @@ void Player::OnBufferProcessed(IBuffer *buffer) {
                 try to enqueue another sample. the thread loop blocks on this condition */
                 this->writeToOutputCondition.notify_all();
 
-            return;
+                it = this->lockedBuffers.end(); /* bail out of the loop */
             }
         }
+
+        if (!this->notifiedStarted) {
+            this->notifiedStarted = true;
+            started = true;
+        }
+    }
+
+    /* we notify our listeners that we've started playing only after the first
+    buffer has been consumed. this is because sometimes we precache buffers
+    and send them to the output before they are actually processed by the
+    output device */
+    if (started) {
+        this->PlaybackStarted(this);
+    }
 }

src/core/audio/Player.h

@@ -117,6 +117,7 @@ namespace musik { namespace core { namespace audio {
             double currentPosition;
             double setPosition;
             int state;
+            bool notifiedStarted;
     };
 
 } } }

src/core/playback/Transport.cpp

@@ -124,7 +124,7 @@ void Transport::StartWithPlayer(Player* newPlayer) {
 
             musik::debug::info(TAG, "play()");
 
-            this->active.push_front(newPlayer);
+            this->active.push_back(newPlayer);
             this->output->Resume();
             newPlayer->Play();
         }
@@ -140,6 +140,10 @@ void Transport::Stop() {
 void Transport::Stop(bool suppressStopEvent, bool stopOutput) {
     musik::debug::info(TAG, "stop");
 
+    /* if we stop the output, we kill all of the Players immediately.
+    otherwise, we let them finish naturally; RemoveActive() will take
+    care of disposing of them */
+    if (stopOutput) {
         std::list<Player*> toDelete;
 
         {
@@ -154,8 +158,6 @@ void Transport::Stop(bool suppressStopEvent, bool stopOutput) {
         std::for_each(toDelete.begin(), toDelete.end(), stopPlayer);
         DEFER(&Transport::DeletePlayers, toDelete);
 
-
-    if (stopOutput) {
         /* stopping the transport will stop any buffers that are currently in
         flight. this makes the sound end immediately. */
         this->output->Stop();

src/musikbox/app/window/TransportWindow.cpp

@@ -167,7 +167,7 @@ void TransportWindow::Update() {
     int secondsCurrent = (int) round(transport.Position());
     int secondsTotal = boost::lexical_cast<int>(duration);
 
-    std::string currentTime = text::Duration(secondsCurrent);
+    std::string currentTime = text::Duration(std::min(secondsCurrent, secondsTotal));
     std::string totalTime = text::Duration(secondsTotal);
 
     size_t timerWidth =