Streaming audio data now works in a similar way to buffered data.

It is possible to loop streaming sources, as well as play
them again after they finished playing. It is possible to stop()
and then play() a streaming source to start playing from the
beginning. Essentially the lifecycle of a streaming audio
is now completely controlled by the user, in the same fashion
as buffered data.
In addition, AudioNode status updates (when an audio stops playing)
now occur every frame rendered, as opposed to every 50 milliseconds.

jme3-core/src/main/java/com/jme3/audio/AudioStream.java

@@ -39,36 +39,37 @@ import java.util.logging.Level;
 import java.util.logging.Logger;
 
 /**
- * <code>AudioStream</code> is an implementation of AudioData that
- * acquires the audio from an InputStream. Audio can be streamed
- * from network, hard drive etc. It is assumed the data coming
- * from the input stream is uncompressed.
+ * <code>AudioStream</code> is an implementation of AudioData that acquires the
+ * audio from an InputStream. Audio can be streamed from network, hard drive
+ * etc. It is assumed the data coming from the input stream is uncompressed.
  *
  * @author Kirill Vainer
  */
-public class AudioStream extends AudioData implements Closeable{
+public class AudioStream extends AudioData implements Closeable {
 
     private final static Logger logger = Logger.getLogger(AudioStream.class.getName());
     protected InputStream in;
     protected float duration = -1f;
     protected boolean open = false;
+    protected boolean eof = false;
     protected int[] ids;
-    
-    public AudioStream(){
-        super();        
+
+    public AudioStream() {
+        super();
     }
-    
-    protected AudioStream(int[] ids){
+
+    protected AudioStream(int[] ids) {
         // Pass some dummy ID so handle
         // doesn't get created.
-        super(-1);      
+        super(-1);
         // This is what gets destroyed in reality
         this.ids = ids;
     }
 
-    public void updateData(InputStream in, float duration){
-        if (id != -1 || this.in != null)
+    public void updateData(InputStream in, float duration) {
+        if (id != -1 || this.in != null) {
             throw new IllegalStateException("Data already set!");
+        }
 
         this.in = in;
         this.duration = duration;
@@ -76,22 +77,27 @@ public class AudioStream extends AudioData implements Closeable{
     }
 
     /**
-     * Reads samples from the stream. The format of the data
-     * depends on the getSampleRate(), getChannels(), getBitsPerSample()
-     * values.
+     * Reads samples from the stream. The format of the data depends on the
+     * getSampleRate(), getChannels(), getBitsPerSample() values.
      *
      * @param buf Buffer where to read the samples
      * @param offset The offset in the buffer where to read samples
      * @param length The length inside the buffer where to read samples
      * @return number of bytes read.
      */
-    public int readSamples(byte[] buf, int offset, int length){
-        if (!open)
+    public int readSamples(byte[] buf, int offset, int length) {
+        if (!open || eof) {
             return -1;
+        }
 
-        try{
-            return in.read(buf, offset, length);
-        }catch (IOException ex){
+        try {
+            int totalRead = in.read(buf, offset, length);
+            if (totalRead < 0) {
+                eof = true;
+            }
+            return totalRead;
+        } catch (IOException ex) {
+            eof = true;
             return -1;
         }
     }
@@ -103,41 +109,41 @@ public class AudioStream extends AudioData implements Closeable{
      * @param buf Buffer where to read the samples
      * @return number of bytes read.
      */
-    public int readSamples(byte[] buf){
+    public int readSamples(byte[] buf) {
         return readSamples(buf, 0, buf.length);
     }
 
-    public float getDuration(){
+    public float getDuration() {
         return duration;
     }
 
     @Override
-    public int getId(){
+    public int getId() {
         throw new RuntimeException("Don't use getId() on streams");
     }
 
     @Override
-    public void setId(int id){
+    public void setId(int id) {
         throw new RuntimeException("Don't use setId() on streams");
     }
 
-    public void initIds(int count){
+    public void initIds(int count) {
         ids = new int[count];
     }
 
-    public int getId(int index){
+    public int getId(int index) {
         return ids[index];
     }
 
-    public void setId(int index, int id){
+    public void setId(int index, int id) {
         ids[index] = id;
     }
 
-    public int[] getIds(){
+    public int[] getIds() {
         return ids;
     }
 
-    public void setIds(int[] ids){
+    public void setIds(int[] ids) {
         this.ids = ids;
     }
 
@@ -155,9 +161,7 @@ public class AudioStream extends AudioData implements Closeable{
 
     @Override
     public void deleteObject(Object rendererObject) {
-        // It seems that the audio renderer is already doing a good
-        // job at deleting audio streams when they finish playing.
-//        ((AudioRenderer)rendererObject).deleteAudioData(this);
+        ((AudioRenderer) rendererObject).deleteAudioData(this);
     }
 
     @Override
@@ -165,42 +169,42 @@ public class AudioStream extends AudioData implements Closeable{
         return new AudioStream(ids);
     }
     
-    /**
-     * @return Whether the stream is open or not. Reading from a closed
-     * stream will always return eof.
-     */
-    public boolean isOpen(){
-        return open;
+    public boolean isEOF() {
+        return eof;
     }
-
+    
     /**
-     * Closes the stream, releasing all data relating to it. Reading
-     * from the stream will return eof.
+     * Closes the stream, releasing all data relating to it. 
+     * Reading from the stream will return eof.
+     *
      * @throws IOException
      */
     public void close() {
-        if (in != null && open){
-            try{
+        if (in != null && open) {
+            try {
                 in.close();
-            }catch (IOException ex){
+            } catch (IOException ex) {
             }
             open = false;
-        }else{
+        } else {
             throw new RuntimeException("AudioStream is already closed!");
         }
     }
 
-  
-    public void setTime(float time){
-        if(in instanceof SeekableStream){
-            ((SeekableStream)in).setTime(time);
-        }else{
-            logger.log(Level.WARNING,"Cannot use setTime on a stream that is not seekable. You must load the file with the streamCache option set to true");
+    public void setTime(float time) {
+        if (in instanceof SeekableStream) {
+            ((SeekableStream) in).setTime(time);
+            eof = false;
+        } else {
+            logger.log(Level.WARNING, 
+                    "Cannot use setTime on a stream that "
+                    + "is not seekable. You must load the file "
+                    + "with the streamCache option set to true");
         }
     }
 
     @Override
     public long getUniqueId() {
-        return ((long)OBJTYPE_AUDIOSTREAM << 32) | ((long)ids[0]);
+        return ((long) OBJTYPE_AUDIOSTREAM << 32) | ((long) ids[0]);
     }
 }

jme3-core/src/main/java/com/jme3/audio/openal/ALAudioRenderer.java

@@ -72,9 +72,10 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
     private int auxSends = 0;
     private int reverbFx = -1;
     private int reverbFxSlot = -1;
-    // Update audio 20 times per second
+    
+    // Fill streaming sources every 50 ms
     private static final float UPDATE_RATE = 0.05f;
-    private final Thread audioThread = new Thread(this, "jME3 Audio Thread");
+    private final Thread decoderThread = new Thread(this, "jME3 Audio Decoding Thread");
     private final AtomicBoolean threadLock = new AtomicBoolean(false);
 
     private final AL al;
@@ -88,23 +89,26 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
     }
 
     public void initialize() {
-        if (!audioThread.isAlive()) {
-            audioThread.setDaemon(true);
-            audioThread.setPriority(Thread.NORM_PRIORITY + 1);
-            audioThread.start();
+        if (!decoderThread.isAlive()) {
+            // Set high priority to avoid buffer starvation.
+            decoderThread.setDaemon(true);
+            decoderThread.setPriority(Thread.NORM_PRIORITY + 1);
+            decoderThread.start();
         } else {
             throw new IllegalStateException("Initialize already called");
         }
     }
 
     private void checkDead() {
-        if (audioThread.getState() == Thread.State.TERMINATED) {
-            throw new IllegalStateException("Audio thread is terminated");
+        if (decoderThread.getState() == Thread.State.TERMINATED) {
+            throw new IllegalStateException("Decoding thread is terminated");
         }
     }
 
     public void run() {
-        initInThread();
+        initInDecoderThread();
+        
+        // Notify render thread that OAL context is available.
         synchronized (threadLock) {
             threadLock.set(true);
             threadLock.notifyAll();
@@ -120,7 +124,7 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
             }
 
             synchronized (threadLock) {
-                updateInThread(UPDATE_RATE);
+                updateInDecoderThread(UPDATE_RATE);
             }
 
             long endTime = System.nanoTime();
@@ -139,11 +143,11 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
         }
 
         synchronized (threadLock) {
-            cleanupInThread();
+            cleanupInDecoderThread();
         }
     }
 
-    public void initInThread() {
+    public void initInDecoderThread() {
         try {
             if (!alc.isCreated()) {
                 alc.createALC();
@@ -225,7 +229,7 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
         }
     }
 
-    public void cleanupInThread() {
+    public void cleanupInDecoderThread() {
         if (audioDisabled) {
             alc.destroyALC();
             return;
@@ -264,10 +268,10 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
 
     public void cleanup() {
         // kill audio thread
-        if (audioThread.isAlive()) {
-            audioThread.interrupt();
+        if (decoderThread.isAlive()) {
+            decoderThread.interrupt();
             try {
-                audioThread.join();
+                decoderThread.join();
             } catch (InterruptedException ex) {
             }
         }
@@ -453,10 +457,8 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
                     }
                     break;
                 case Looping:
-                    if (src.isLooping()) {
-                        if (!(src.getAudioData() instanceof AudioStream)) {
-                            al.alSourcei(id, AL_LOOPING, AL_TRUE);
-                        }
+                    if (src.isLooping() && !(src.getAudioData() instanceof AudioStream)) {
+                        al.alSourcei(id, AL_LOOPING, AL_TRUE);
                     } else {
                         al.alSourcei(id, AL_LOOPING, AL_FALSE);
                     }
@@ -509,7 +511,7 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
             }
         }
 
-        if (forceNonLoop) {
+        if (forceNonLoop || src.getAudioData() instanceof AudioStream) {
             al.alSourcei(id, AL_LOOPING, AL_FALSE);
         } else {
             al.alSourcei(id, AL_LOOPING, src.isLooping() ? AL_TRUE : AL_FALSE);
@@ -661,45 +663,71 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
         return true;
     }
 
-    private boolean fillStreamingSource(int sourceId, AudioStream stream) {
-        if (!stream.isOpen()) {
-            return false;
-        }
-
-        boolean active = true;
+    private boolean fillStreamingSource(int sourceId, AudioStream stream, boolean looping) {
+        boolean success = false;
         int processed = al.alGetSourcei(sourceId, AL_BUFFERS_PROCESSED);
-
-//        while((processed--) != 0){
-        if (processed > 0) {
+        
+        for (int i = 0; i < processed; i++) {
             int buffer;
 
             ib.position(0).limit(1);
             al.alSourceUnqueueBuffers(sourceId, 1, ib);
             buffer = ib.get(0);
 
-            active = fillBuffer(stream, buffer);
-
-            ib.position(0).limit(1);
-            ib.put(0, buffer);
-            al.alSourceQueueBuffers(sourceId, 1, ib);
-        }
-
-        if (!active && stream.isOpen()) {
-            stream.close();
+            boolean active = fillBuffer(stream, buffer);
+            
+            if (!active && !stream.isEOF()) {
+                throw new AssertionError();
+            }
+            
+            if (!active && looping) {
+                stream.setTime(0);
+                active = fillBuffer(stream, buffer);
+            }
+            
+            if (active) {
+                ib.position(0).limit(1);
+                ib.put(0, buffer);
+                al.alSourceQueueBuffers(sourceId, 1, ib);
+                // At least one buffer enqueued = success.
+                success = true;
+            } else {
+                // No more data left to process.
+                break;
+            }
         }
 
-        return active;
+        return success;
     }
 
-    private boolean attachStreamToSource(int sourceId, AudioStream stream) {
-        boolean active = true;
+    private boolean attachStreamToSource(int sourceId, AudioStream stream, boolean looping) {
+        boolean success = false;
+        
+        // Reset the stream. Typically happens if it finished playing on 
+        // its own and got reclaimed. 
+        // Note that AudioNode.stop() already resets the stream
+        // since it might not be in EOF when stopped.
+        if (stream.isEOF()) {
+            stream.setTime(0);
+        }
+        
         for (int id : stream.getIds()) {
-            active = fillBuffer(stream, id);
-            ib.position(0).limit(1);
-            ib.put(id).flip();
-            al.alSourceQueueBuffers(sourceId, 1, ib);
+            boolean active = fillBuffer(stream, id);
+            if (!active && !stream.isEOF()) {
+                throw new AssertionError();
+            }
+            if (!active && looping) {
+                stream.setTime(0);
+                active = fillBuffer(stream, id);
+            }
+            if (active) {
+                ib.position(0).limit(1);
+                ib.put(id).flip();
+                al.alSourceQueueBuffers(sourceId, 1, ib);
+                success = true;
+            }
         }
-        return active;
+        return success;
     }
 
     private boolean attachBufferToSource(int sourceId, AudioBuffer buffer) {
@@ -707,11 +735,11 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
         return true;
     }
 
-    private boolean attachAudioToSource(int sourceId, AudioData data) {
+    private boolean attachAudioToSource(int sourceId, AudioData data, boolean looping) {
         if (data instanceof AudioBuffer) {
             return attachBufferToSource(sourceId, (AudioBuffer) data);
         } else if (data instanceof AudioStream) {
-            return attachStreamToSource(sourceId, (AudioStream) data);
+            return attachStreamToSource(sourceId, (AudioStream) data, looping);
         }
         throw new UnsupportedOperationException();
     }
@@ -723,15 +751,9 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
 
             int sourceId = channels[index];
             al.alSourceStop(sourceId);
-
-            if (src.getAudioData() instanceof AudioStream) {
-                AudioStream str = (AudioStream) src.getAudioData();
-                ib.position(0).limit(STREAMING_BUFFER_COUNT);
-                ib.put(str.getIds()).flip();
-                al.alSourceUnqueueBuffers(sourceId, STREAMING_BUFFER_COUNT, ib);
-            } else if (src.getAudioData() instanceof AudioBuffer) {
-                al.alSourcei(sourceId, AL_BUFFER, 0);
-            }
+            
+            // For streaming sources, this will clear all queued buffers.
+            al.alSourcei(sourceId, AL_BUFFER, 0);
 
             if (src.getDryFilter() != null && supportEfx) {
                 // detach filter
@@ -747,71 +769,118 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
             chanSrcs[index] = null;
         }
     }
+    
+    private AudioSource.Status convertStatus(int oalStatus) {
+        switch (oalStatus) {
+            case AL_INITIAL:
+            case AL_STOPPED:
+                return Status.Stopped;
+            case AL_PAUSED:
+                return Status.Paused;
+            case AL_PLAYING:
+                return Status.Playing;
+            default:
+                throw new UnsupportedOperationException("Unrecognized OAL state: " + oalStatus);
+        }
+    }
 
     public void update(float tpf) {
-        // does nothing
+        synchronized (threadLock) {
+            updateInRenderThread(tpf);
+        }
     }
 
-    public void updateInThread(float tpf) {
+    public void updateInRenderThread(float tpf) {
         if (audioDisabled) {
             return;
         }
-
+        
         for (int i = 0; i < channels.length; i++) {
             AudioSource src = chanSrcs[i];
+            
             if (src == null) {
                 continue;
             }
 
             int sourceId = channels[i];
-
-            // is the source bound to this channel
-            // if false, it's an instanced playback
             boolean boundSource = i == src.getChannel();
-
-            // source's data is streaming
-            boolean streaming = src.getAudioData() instanceof AudioStream;
-
-            // only buffered sources can be bound
-            assert (boundSource && streaming) || (!streaming);
-
-            int state = al.alGetSourcei(sourceId, AL_SOURCE_STATE);
-            boolean wantPlaying = src.getStatus() == Status.Playing;
-            boolean stopped = state == AL_STOPPED;
-
-            if (streaming && wantPlaying) {
-                AudioStream stream = (AudioStream) src.getAudioData();
-                if (stream.isOpen()) {
-                    fillStreamingSource(sourceId, stream);
-                    if (stopped) {
-                        al.alSourcePlay(sourceId);
+            boolean reclaimChannel = false;
+            
+            Status oalStatus = convertStatus(al.alGetSourcei(sourceId, AL_SOURCE_STATE));
+            Status jmeStatus = src.getStatus();
+            
+            // Check if we need to sync JME status with OAL status.
+            if (oalStatus != jmeStatus) {
+                if (oalStatus == Status.Stopped && jmeStatus == Status.Playing) {
+                    // Maybe we need to reclaim the channel.
+                    if (src.getAudioData() instanceof AudioStream) {
+                        AudioStream stream = (AudioStream) src.getAudioData();
+                        
+                        if (stream.isEOF() && !src.isLooping()) {
+                            // Stream finished playing
+                            reclaimChannel = true;
+                        } else {
+                            // Stream still has data. 
+                            // Buffer starvation occured.
+                            // Audio decoder thread will fill the data 
+                            // and start the channel again.
+                        }
+                    } else {
+                        // Buffer finished playing.
+                        reclaimChannel = true;
                     }
-                } else {
-                    if (stopped) {
-                        // became inactive
-                        src.setStatus(Status.Stopped);
-                        src.setChannel(-1);
+                    
+                    if (reclaimChannel) {
+                        if (boundSource) {
+                            src.setStatus(Status.Stopped);
+                            src.setChannel(-1);
+                        }
                         clearChannel(i);
                         freeChannel(i);
-
-                        // And free the audio since it cannot be
-                        // played again anyway.
-                        deleteAudioData(stream);
                     }
+                } else {
+                    // jME3 state does not match OAL state.
+                    throw new AssertionError();
                 }
-            } else if (!streaming) {
-                boolean paused = state == AL_PAUSED;
+            } else {
+                // Stopped channel was not cleared correctly.
+                if (oalStatus == Status.Stopped) {
+                    throw new AssertionError();
+                }
+            }
+        }
+    }
+    
+    public void updateInDecoderThread(float tpf) {
+        if (audioDisabled) {
+            return;
+        }
 
-                // make sure OAL pause state & source state coincide
-                assert (src.getStatus() == Status.Paused && paused) || (!paused);
+        for (int i = 0; i < channels.length; i++) {
+            AudioSource src = chanSrcs[i];
+            
+            if (src == null || !(src.getAudioData() instanceof AudioStream)) {
+                continue;
+            }
 
-                if (stopped) {
-                    if (boundSource) {
-                        src.setStatus(Status.Stopped);
-                        src.setChannel(-1);
-                    }
-                    clearChannel(i);
-                    freeChannel(i);
+            int sourceId = channels[i];
+            AudioStream stream = (AudioStream) src.getAudioData();
+
+            Status oalStatus = convertStatus(al.alGetSourcei(sourceId, AL_SOURCE_STATE));
+            Status jmeStatus = src.getStatus();
+
+            // Keep filling data (even if we are stopped / paused)
+            boolean buffersWereFilled = fillStreamingSource(sourceId, stream, src.isLooping());
+
+            if (buffersWereFilled) {
+                if (oalStatus == Status.Stopped && jmeStatus == Status.Playing) {
+                    // The source got stopped due to buffer starvation.
+                    // Start it again.
+                    logger.log(Level.WARNING, "Buffer starvation "
+                                            + "occurred while playing stream");
+                    al.alSourcePlay(sourceId);
+                } else {
+                    // Buffers were filled, stream continues to play.
                 }
             }
         }
@@ -877,7 +946,7 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
             if (src.getAudioData() instanceof AudioStream) {
                 throw new UnsupportedOperationException(
                         "Cannot play instances "
-                        + "of audio streams. Use playSource() instead.");
+                        + "of audio streams. Use play() instead.");
             }
 
             if (src.getAudioData().isUpdateNeeded()) {
@@ -896,7 +965,7 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
 
             // set parameters, like position and max distance
             setSourceParams(sourceId, src, true);
-            attachAudioToSource(sourceId, src.getAudioData());
+            attachAudioToSource(sourceId, src.getAudioData(), false);
             chanSrcs[index] = src;
 
             // play the channel
@@ -917,12 +986,11 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
                 return;
             }
 
-            //assert src.getStatus() == Status.Stopped || src.getChannel() == -1;
-
             if (src.getStatus() == Status.Playing) {
                 return;
             } else if (src.getStatus() == Status.Stopped) {
-
+                assert src.getChannel() != -1;
+                
                 // allocate channel to this source
                 int index = newChannel();
                 if (index == -1) {
@@ -939,7 +1007,7 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
 
                 chanSrcs[index] = src;
                 setSourceParams(channels[index], src, false);
-                attachAudioToSource(channels[index], data);
+                attachAudioToSource(channels[index], data, src.isLooping());
             }
 
             al.alSourcePlay(channels[src.getChannel()]);
@@ -989,16 +1057,9 @@ public class ALAudioRenderer implements AudioRenderer, Runnable {
                 src.setChannel(-1);
                 clearChannel(chan);
                 freeChannel(chan);
-
+                
                 if (src.getAudioData() instanceof AudioStream) {
-                    AudioStream stream = (AudioStream) src.getAudioData();
-                    if (stream.isOpen()) {
-                        stream.close();
-                    }
-
-                    // And free the audio since it cannot be
-                    // played again anyway.
-                    deleteAudioData(src.getAudioData());
+                    ((AudioStream)src.getAudioData()).setTime(0);
                 }
             }
         }