// Copyright 2014-2025 Jesper Larsson
//
// This file is part of Klipspringer, <https://klipspringer.avadeaux.net/>
//
// Klipspringer is free software: you can redistribute it and/or modify it under the terms of the
// GNU General Public License as published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// Klipspringer is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without
// even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
// General Public License for more details.
//
// You should have received a copy of the GNU General Public License along with Klipspringer. If
// not, see <https://www.gnu.org/licenses/>.
package net.avadeaux.klipspringer;
import static net.avadeaux.klipspringer.PlayState.Status.*;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import javax.sound.sampled.*;
import net.avadeaux.klipspringer.codec.*;
/** Buffers up PCM data and sends it on in real time to a player. There are two points with this
* internal buffering: to avoid hearing a short silence between tracks in case loading the next
* track is slow, and to allow crossfading with track transitions.
*/
public class TrackConsumer implements PcmWriter {
private static class Play {
final Track track;
final long skipFrames;
final Device.Player.Factory outFact;
final PcmFormat format;
final Track.Limits limits;
Play(Track track, long skipFrames, Device.Player.Factory outFact, PcmFormat format, Track.Limits limits) {
this.track = track;
this.skipFrames = skipFrames;
this.outFact = outFact;
this.format = format;
this.limits = limits;
}
}
// ---------------------------------------------------------------------------------------------
// Consumer thread runnable, with its own synchronization monitor.
private class ConsumeRunner implements Runnable {
private boolean busy = false; // conusumer section in progress
private boolean exit = false; // exit has been called, terminate thread
private boolean purged = false; // waiting for non-busy state should fail
private Queue<Play> queue = new LinkedList<Play>();
// Called when play starts from a new position (new track sequence).
synchronized void seek(Track track, long skipFrames, Device.Player.Factory outFact, PcmFormat format, Track.Limits limits) {
queue.clear();
if (skipFrames * format.fs() < limits.outStart) { // unless seeking into fade-out
queue.add(new Play(track, skipFrames, outFact, format, limits));
}
notifyAll();
}
// Called when play should continue with the next track being decoded.
synchronized void queue(Track track, Device.Player.Factory outFact, PcmFormat format, Track.Limits limits) {
queue.add(new Play(track, 0, outFact, format, limits));
notifyAll();
}
public void run() {
consumerThread: while (true) {
Play play = null;
synchronized (this) { // get play record from queue
while ((play = queue.poll()) == null) {
long before = System.currentTimeMillis();
boolean streaming = currentOut != null && currentOut instanceof AudioStream && ((AudioStream) currentOut).streaming();
try { wait(timeoutMillis); } catch (InterruptedException e) { }
if (!streaming && System.currentTimeMillis() - before >= timeoutMillis) {
if (System.getProperty("klipspringer.debug") != null) { System.err.println("Consume timed out"); }
break consumerThread;
}
}
busy = true;
}
Device.Player out = getOut();
try {
if (out != null && !out.format().matchesQuality(play.track.format())) {
out.drain();
setOut(out = null);
}
if (out == null) {
if (!inputStarting(play.track, play.limits)) { continue; }
state.consumerStatus(STREAM_SWITCHING, null, 0);
out = play.outFact.open(play.format);
setOut(out);
} else {
state.consumerStatus(PLAYING, play.track, play.skipFrames/play.track.format().getSampleRate());
}
consumingFrame = play.skipFrames;
boolean ok = consumeTrack(play.track, out, play.format, play.limits);
if (ok && play.track.index == tracks.length-1) {
out.drain();
return;
}
} catch (IOException ex) {
state.consumerStatus(STOPPED, null, 0);
if (System.getProperty("klipspringer.debug") != null) {
System.err.println("Consume aborted: "+ex);
if (!(ex instanceof java.nio.channels.ClosedChannelException)) { ex.printStackTrace(); }
}
} catch (RuntimeException ex) {
stop();
throw ex;
} finally {
synchronized (this) {
busy = false;
notifyAll();
}
}
}
stop();
}
// Waits until the consumer thread is in a state where it is not consuming a track, and it
// is thus safe to start feeding a new track sequence. Returns true if idle state was
// reached, false if it was given up because of purge (which means that the new track
// sequence has been prematurely stopped).
synchronized boolean awaitIdle() {
while (busy && !purged) {
try { wait(); } catch (InterruptedException e) { }
}
purged = false;
return !busy;
}
// Make awaitIdle return false immediately.
synchronized void purge() {
purged = true;
notifyAll();
}
synchronized void clearPurged() { purged = false; }
}
// ---------------------------------------------------------------------------------------------
// TrackConsumer members and constructor
private final Track.List tracks; // the track list to be consumed
private final PlayState state; // where to send status notification
private final ConsumeRunner crun; // runner for consumer thread
private final ByteBuffer ibuf; // underlying buffer
private final ByteBuffer obuf; // mirror of ibuf used on consumer side
private final long timeoutMillis; // how long to wait for something to be queued
private boolean stopped = false; // tells consumer to abandon track and feeder to refuse data
private boolean paused = false; // pause mode
private Track feeding = null; // track currently being fed
private Track.Limits flim = null; // limits of feeding
private boolean buflock = false; // set when accessing buffer outside synchronized
private PcmBuffer fbuf; // wraps ibuf
private int bufn = 0; // bytes ready for consumer in ibuf
private int bufx = 0; // unused byte due to alignment or wrap-around
private int fadn = 0; // bytes in fade-out section not ready to play
private int fadx = 0; // unused bytes in fade-out section
private long gotBytes = 0; // total amount received by write method
private long consumingFrame; // set in run before calling consumeTrack, which updates it
private boolean monomix = false; // monomix audio before writing
// Administered by the consumer thread, shared with stop, pause, and unpause.
private Device.Player currentOut = null;
public TrackConsumer(Track.List tracks, PlayState state, int bufferBytes, long timeoutMillis) {
this.tracks = tracks;
this.state = state;
crun = new ConsumeRunner();
ibuf = ByteBuffer.allocateDirect(bufferBytes);
obuf = ibuf.duplicate();
this.timeoutMillis = timeoutMillis;
}
public void close() { stop(); }
// Called from the consumer thread.
private synchronized Device.Player getOut() { return currentOut; }
private synchronized void setOut(Device.Player out) { currentOut = out; }
// Adjusts buffer position to be good for reading/writing at least one frame. If the position is
// not aligned for the format, it is advanced a few bytes. If it is too close to the end of the
// buffer, it is reset to zero. Returns the number of unused bytes skipped over.
private static int adjustPosition(ByteBuffer buf, PcmFormat fmt) {
int p = buf.position();
int x = -p & fmt.alignment()-1; // padding
int c = buf.capacity();
if (p+x + fmt.fs() > c) { // not enough room for frame
buf.position(0); // go to beginning of buf
return c - p;
} else {
buf.position(p+x); // add padding
return x;
}
}
// ---------------------------------------------------------------------------------------------
// Feeder methods
public void runConsumerThread() {
crun.run();
}
/** Feed one batch of PCM data. */
public boolean write(ByteBuffer data) throws IOException {
int dfs = feeding.format().fs(), bfs = fbuf.format().fs();
long leave = gotBytes + data.remaining()/dfs*bfs - flim.end; // bytes not to play
if (leave > 0) { data.limit(data.limit() - (int) leave); }
int m = data.remaining()/dfs; // frames to write in ibuf
synchronized (this) {
if (stopped) { return false; }
while (buflock || m*(bfs+1) > ibuf.capacity()-bufn-fadn-bufx) { // locked or not enough room
if (!buflock && bufn == 0 && m*(bfs+1) > ibuf.capacity()-fadn-bufx) {
System.err.println("Deadlock in feeding "+feeding);
throw new IllegalStateException("Deadlock due to too small internal PCM buffer");
}
try { wait(); } catch (InterruptedException e) { }
if (stopped) { return false; }
continue;
}
buflock = true; // keep consumer off buffer
}
for (; m > 0; m = data.remaining()/dfs) { // something left to write
m = Math.min(m, ibuf.remaining()/bfs);
int x = adjustPosition(ibuf, fbuf.format()); // padding
bufx += x;
if (gotBytes + m*bfs > flim.outStart) { fadx += x; } // padding in fade-out
if (gotBytes < flim.inEnd) { // in fade-in section
m = (int) Math.min(m, (flim.inEnd - gotBytes)/bfs); // stop at end of fade-in
PcmBuffer fdata = PcmBuffer.of(data, feeding.format(), Device.BitLayout.LSB);
MixSpec.Fade fade = feeding.fade();
double dt = 1.0 / ((flim.inEnd-flim.start)/bfs - 1); // time increment
double t = (gotBytes - flim.start)/bfs * dt; // point in time
int bss = fbuf.format().ss();
int channels = feeding.format().channels();
int p = ibuf.position();
for (int i = 0, c = 0; i < m*channels; i++) {
fbuf.putSample(p+i*bss, fade.mix(fbuf.getSample(p+i*bss), fdata.getSample(), t));
if (++c == channels) { c = 0; t += dt; }
}
fdata.close();
ibuf.position(p + m*bfs);
bufn += m*bfs;
} else { // use bulk transfer
int e = data.limit(); // remember
data.limit(data.position() + m*dfs); // transfer m frames
fbuf.put(data, feeding.format()); // transfer data into ibuf
data.limit(e); // back to remembered limit
if (gotBytes + m*bfs > flim.outStart) { // in fade-out section
int nonfadn = (int) Math.max(0, flim.outStart - gotBytes);
bufn += nonfadn;
fadn += m*bfs - nonfadn;
} else { // not in fade-out
bufn += m*bfs;
}
}
gotBytes += m*bfs;
}
synchronized (this) {
buflock = false; // consumer access to buffer
notifyAll();
return gotBytes < flim.end;
}
}
/** Tells the consumer that any previous call to {@link #stop()} should not apply to the track
* starting next. Should be called before {@link #continuePlay(Track, Device.Player.Factory)},
* so that it does not not immediately return false.
*/
void readyPlay() { crun.clearPurged(); }
/** Sets off playing. Returns true if successful, false if interrupted by {@link #stop()} since
* the last call to {@link #readyPlay()}.
*/
public boolean startPlay(Track fromTrack, long framesIntoTrack, Device.Player.Factory outFact, boolean monomix)
throws IOException
{
if (!crun.awaitIdle()) { return false; }
synchronized (this) {
if (stopped || !feeding.format().matchesQuality(fromTrack.format())) {
if (fbuf != null) { fbuf.close(); }
fbuf = PcmBuffer.of(ibuf, outFact.selectFormat(fromTrack.format()), outFact.layout(fromTrack.format()));
}
feeding = fromTrack;
flim = feeding.limits(fbuf.format().fs(), 0);
stopped = false;
gotBytes = framesIntoTrack*fbuf.format().fs();
this.monomix = monomix;
state.monomixing(monomix);
// Reset buffer.
ibuf.clear();
obuf.clear();
bufn = bufx = fadn = fadx = 0;
crun.seek(fromTrack, framesIntoTrack, outFact, fbuf.format(), flim);
}
return true;
}
/** Continues playing with next track without any seek. */
public boolean continuePlay(Track track, Device.Player.Factory outFact) throws IOException {
synchronized (this) {
if (!feeding.format().matchesQuality(track.format())) {
if (outFact instanceof AudioStream.Factory) {
// Stream cannot switch format.
currentOut.drain();
return false;
}
fbuf = PcmBuffer.of(ibuf, outFact.selectFormat(track.format()), outFact.layout(track.format()));
}
feeding = track;
flim = feeding.limits(fbuf.format().fs(), flim.outStart);
}
// Retreat to start of fade-out.
gotBytes -= fadn;
int p = ibuf.position(), c = ibuf.capacity();
ibuf.position((c + p - (fadn+fadx)) % c);
bufx -= fadx;
fadx = fadn = 0;
crun.queue(track, outFact, fbuf.format(), flim);
return true;
}
// ---------------------------------------------------------------------------------------------
// Control interface methods
/** Sets pause mode to active, with no effect if it is already active. */
public synchronized void pause() throws IOException {
PlayState.ResumeException ex = null;
if (!paused) {
if (currentOut != null) {
if (currentOut.canPause()) {
currentOut.pause();
} else {
ex = state.resumeException();
stop();
}
}
paused = true;
notifyAll();
if (ex != null) { throw ex; }
}
}
/** Sets pause mode to inactive, with no effect if it is already inactive. */
public synchronized void unpause() throws IOException {
if (paused) {
if (currentOut != null) { currentOut.unpause(); }
paused = false;
notifyAll();
}
}
/** Stops consumer from playing and feeder from accepting data. When the consumer side detects
* the stop signal, it sets state to STOPPED and drops out of the consumption loop, returning
* to run to pick up another track. On the feeder side, any write operation will return false
* while in stopped state.
*/
public synchronized void stop() {
crun.purge();
if (currentOut != null) {
try {
currentOut.close();
} catch (Exception ex) {
if (System.getProperty("klipspringer.debug") != null) { ex.printStackTrace(); }
} finally {
currentOut = null;
}
}
stopped = true;
notifyAll();
}
// ---------------------------------------------------------------------------------------------
// Consumer methods
// Waits until input starts arriving, or it becomes known that no input will arrive.
private synchronized boolean inputStarting(Track consuming, Track.Limits lim) {
while (!stopped && (buflock || bufn == 0 && gotBytes < lim.end)) {
try { wait(); } catch (InterruptedException e) { }
}
return !stopped && bufn > 0;
}
// Consumption loop, plays until end of track is reached of stop is called.
private boolean consumeTrack(Track consuming,
Device.Player out,
PcmFormat fmt,
Track.Limits lim)
throws IOException
{
long sentBytes = lim.start + consumingFrame * fmt.fs();
double trackTime = consumingFrame/fmt.getSampleRate();
obuf.order(fmt.order());
while (true) {
synchronized (this) {
if (stopped) {
state.consumerStatus(STOPPED, null, 0);
return false;
}
if (paused) {
state.consumerStatus(PAUSED, consuming, trackTime - out.bufferedFrames()/fmt.getSampleRate());
try { wait(); } catch (InterruptedException e) { }
continue;
}
state.consumerStatus(PLAYING, consuming, trackTime - out.bufferedFrames()/fmt.getSampleRate());
if (sentBytes == lim.outStart) { return true; } // track completed
if (buflock || bufn == 0) { // wait for feeder
try { wait(); } catch (InterruptedException e) { }
continue;
}
buflock = true; // keep feeder off buffer
}
bufx -= adjustPosition(obuf, fmt); // skip padding
int p = obuf.position();
try { // holding buflock
int n = (int) Math.min(bufn, lim.outStart - sentBytes);
obuf.limit(Math.min(p+n, obuf.capacity()-bufx));
if (monomix) { fbuf.monomix(obuf.position(), obuf.remaining()/fmt.fs()); }
boolean ok = out.write(obuf);
if (!ok && !paused) { return false; } // write returned false
} finally {
obuf.limit(obuf.capacity());
int n = obuf.position() - p; // bytes written
bufn -= n;
sentBytes += n;
consumingFrame += n/fmt.fs();
trackTime += n/fmt.fs()/fmt.getSampleRate();
synchronized (this) {
buflock = false; // feeder access to buffer
notifyAll();
}
}
}
}
}
Version: v4.3.2.2 (2026-05-16T17:03:34+02:00)
Raw file
Source code overview
Klipspringer home