// 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.codec.spi;
import java.nio.ByteBuffer;
import java.io.IOException;
import java.util.concurrent.ThreadFactory;
import javax.sound.sampled.AudioFileFormat;
import javax.sound.sampled.AudioFormat;
import javax.sound.sampled.AudioInputStream;
import javax.sound.sampled.UnsupportedAudioFileException;
import net.avadeaux.klipspringer.codec.*;
/**
* Wrapper for {@link AudioDecoder} to make it readable as an {@link AudioInputStream}
*/
public class DecoderAudioInputStream extends AudioInputStream {
// Factory for feeder threads.
private static ThreadFactory threadFactory = new ThreadFactory() {
public Thread newThread(Runnable r) {
Thread thread = new Thread(new Runnable() {
public void run() {
try {
r.run();
} catch (Throwable thr) {
System.err.println("Uncaught in DecoderAudioInputStream feeder thread: "+thr);
thr.printStackTrace();
System.exit(1);
}
}
}, "DecoderAudioInputStream feeder");
thread.setDaemon(true);
return thread;
}
};
/**
* Sets global thread factory for feeder threads, returning the previous factory. This allows,
* for instance, controlling what happens on uncaught throwables (such as Error, because
* RuntimeExceptions are caught and transferred to the client thread). By default, an uncaught
* throwable prints an error and exits the process.
*/
public static ThreadFactory setThreadFactory(ThreadFactory fact) {
ThreadFactory prev = threadFactory;
threadFactory = fact;
return prev;
}
/** Factory class to delegate object creation to subclass. */
interface DecoderFactory {
AudioDecoder createAudioDecoder(String fileName, AudioDecoder.Target target) throws IOException;
AudioFileFormat.Type type(String fileName);
}
// A target of this class is created in the factory method, and the target is what creates the
// stream (an instance of the surrounding class) that it sends blocks on as they arrive.
private static class ISTarget implements AudioDecoder.Target {
final DecoderFactory fact;
DecoderAudioInputStream stream = null;
ISTarget(DecoderFactory fact) { this.fact = fact; }
public void metadata(PcmFormat format, long totalSamples) {
if (stream == null) { stream = new DecoderAudioInputStream(format, totalSamples); }
}
public boolean write(ByteBuffer buffer) { return stream.write(buffer); }
public void close() throws IOException { }
}
private enum State {
NEW, // no call to available or read yet
READING, // either read or available called
SEEKING, // requested restart decoding from new position
ERROR, // exception in the feeder thread
CLOSED // closed by consumer thread or input exhausted
}
private State state = State.NEW;
// Exception caught in feeder is kept in one of these, and state set to ERROR.
private IOException feedIOEx = null;
private RuntimeException feedRTEx = null;
// Stores the actual buffer sent to Target
private ByteBuffer buffer = null;
private long atSample = 0; // number of the next sample that read returns
private long fromSample = 0; // position to start decoder from
private long markedSample = 0; // position set by mark()
static DecoderAudioInputStream getInstance(String fileName, DecoderFactory fact) throws IOException, UnsupportedAudioFileException {
ISTarget target = new ISTarget(fact);
AudioDecoder firstDecoder;
try {
firstDecoder = fact.createAudioDecoder(fileName, target);
firstDecoder.decodeMetadata();
} catch (IOException ex) {
throw new UnsupportedAudioFileException();
}
threadFactory.newThread(new Runnable() {
public void run() {
AudioDecoder decoder = firstDecoder;
while (decoder != null && target.stream.decodeAllOf(decoder)) {
decoder = target.stream.newDecoder(fact, fileName, target);
}
}
}).start();
return target.stream;
}
DecoderAudioInputStream(AudioFormat fmt, long lengthSamples) {
super(null, fmt, lengthSamples);
}
// Feeder thread method to create a new decoder (for seeking) in stream context.
private synchronized AudioDecoder newDecoder(DecoderFactory fact, String fileName, AudioDecoder.Target target) {
try { return fact.createAudioDecoder(fileName, target); }
catch (IOException ex) { feedIOEx = ex; state = State.ERROR; }
catch (RuntimeException ex) { feedRTEx = ex; state = State.ERROR; }
return null;
}
// Wrapper for decoder.close that handles (unlikely) exceptions.
private synchronized void close(AudioDecoder decoder) {
try {
decoder.close();
}
catch (IOException ex) { if (state != State.ERROR) { feedIOEx = ex; state = State.ERROR; } }
catch (RuntimeException ex) { if (state != State.ERROR) { feedRTEx = ex; state = State.ERROR; } }
}
// Feeder thread method to run decoder.decodeAll in stream context. Returns true if a new
// decoder is to be created because decoding was aborted due to seek, false if stream is closed.
private synchronized boolean decodeAllOf(AudioDecoder decoder) {
while (state == State.NEW) { // postpone decoding until someone asks for data
try { wait(); } catch (InterruptedException e) { }
}
switch (state) {
case SEEKING: // entering new decoder after seek request
state = State.READING;
case READING: // ready to decodeAll
try {
decoder.decodeAll(atSample = fromSample);
if (state == State.READING) { // exhausted without interruption
state = State.CLOSED;
}
}
catch (IOException ex) { feedIOEx = ex; state = State.ERROR; }
catch (RuntimeException ex) { feedRTEx = ex; state = State.ERROR; }
finally { // decoding stopped, normally or abnormally
notifyAll();
close(decoder);
}
break;
case CLOSED: // the stream was closed before anything was read
close(decoder);
break;
default: // should not happen
throw new IllegalStateException();
}
return state == State.SEEKING; // true to process seek
}
// Feeder thread method to transfer a decoded block to the stream. Blocks until the client has
// read the whole block.
private synchronized boolean write(ByteBuffer buffer) {
this.buffer = buffer;
notifyAll();
while (state == State.READING && buffer.hasRemaining()) {
try { wait(); } catch (InterruptedException e) { }
}
this.buffer = null;
return state == State.READING;
}
// Called on the client side to throw an exception caught in the feeder thread.
private synchronized void rethrow() throws IOException {
if (feedIOEx != null) { throw feedIOEx; }
if (feedRTEx != null) { throw feedRTEx; }
}
/** Closes the stream, causing decoding to be aborted and the feeder thread to finish. */
public synchronized void close() throws IOException {
switch (state) {
case NEW:
case READING:
case SEEKING:
state = State.CLOSED;
notifyAll();
break;
case ERROR:
rethrow();
case CLOSED:
break;
default: // should not be possible
throw new IllegalStateException();
}
}
// Common start of read methods: check for errors and block until input is available.
private synchronized void readUp() throws IOException {
if (state == State.ERROR) { rethrow(); }
if (state == State.NEW) {
state = State.READING;
notifyAll();
}
while (state == State.READING && (buffer == null || !buffer.hasRemaining()) || state == State.SEEKING) {
try { wait(); } catch (InterruptedException e) { }
}
if (state == State.ERROR) { rethrow(); }
}
/** See {@link AudioInputStream#read()}. */
public synchronized int read() throws IOException {
// Docs say this can only be used if frame size is one byte.
if (getFormat().getFrameSize() != 1) { throw new IOException("Attempt to read partial frame"); }
readUp();
if (state != State.READING) { throw new IOException("Attempt to read past end"); }
if (buffer.remaining() == 1) { notifyAll(); }
atSample++;
return buffer.get();
}
/** See {@link AudioInputStream#read(byte[], int, int)}. */
public synchronized int read(byte[] b, int off, int n) throws IOException {
readUp();
if (state != State.READING) { return -1; }
if (n < buffer.remaining()) {
int f = n / getFormat().getFrameSize();
n = f * getFormat().getFrameSize(); // round down to whole frames
buffer.get(b, off, n);
atSample += f;
} else {
n = buffer.remaining();
buffer.get(b, off, n);
atSample += n / getFormat().getFrameSize();
notifyAll(); // let feeder supply next block
}
return n;
}
/** See {@link AudioInputStream#read(byte[])}. */
public synchronized int read(byte[] b) throws IOException {
return read(b, 0, b.length);
}
/** Skips ahead the given number of bytes in the stream. If any data have been read from from
* the stream using the currently active decoder, the decoding process is aborted and a new
* decoder is created starting from the desired position. This method can be combined with
* reset() in order to seek to an arbitrary position in the file. Returns the number of bytes
* actually skipped, which is at most to the end of the stream, and rounded down to the nearest
* integral number of frames.
*/
public synchronized long skip(long n) throws IOException {
long skipSamples = Math.max(n, 0) / getFormat().getFrameSize();
switch (state) {
case NEW: // no decodeAll called yet
case SEEKING: // decodeAll of new decoder not called yet
skipSamples = Math.min(skipSamples, getFrameLength() - fromSample);
fromSample += skipSamples;
break;
case READING: // need to abort decoder and restart with a new one
skipSamples = Math.min(skipSamples, getFrameLength() - atSample);
if (skipSamples > 0) { // ... unless skipping zero samples, which is ignored
fromSample = skipSamples;
state = State.SEEKING; // block read will return false to abort decoding
notifyAll();
}
break;
case ERROR:
rethrow();
case CLOSED:
return 0;
default:
throw new IllegalStateException("Unhandled state: "+state);
}
return skipSamples * getFormat().getFrameSize();
}
/** See {@link AudioInputStream#available()}. */
public synchronized int available() throws IOException {
if (state == State.NEW) {
// In NEW state, there is no available data, because the feader thread waits for the
// state to change before it goes into decoding, but it would be confusing to just
// return 0 immediately, because then the client would probably choose not to read, and
// get stuck. Therefore, change to READING state and allow the feeder to start before
// moving on. Since available() is not supposed to be blocking, wait at most one
// millisecond, which should be plenty for the feeder to come in unless the input is
// blocked. If it is blocked, returning zero is correct behavior.
state = State.READING;
notifyAll();
try { wait(1); } catch (InterruptedException e) { }
}
if (state == State.ERROR) { rethrow(); }
return buffer == null || state != State.READING ? 0 : buffer.remaining();
}
/** Returns true. */
public boolean markSupported() { return true; }
/** Remembers the current position for a later reset. The specified read limit is ignored. By
* default, the beginning of the input is marked.
*/
public synchronized void mark(int ignoredReadLimit) {
markedSample = atSample;
}
/** Resets the stream to the marked position, or to the beginning if mark() has not been
* called. If any data have been read from from the stream using the currently active decoder,
* the decoding process is aborted and a new decoder is created starting from the desired
* position. This method can be combined with skip() in order to seek to an arbitrary position
* in the file.
*/
public synchronized void reset() throws IOException {
switch (state) {
case NEW: // no decodeAll called yet
case SEEKING: // decodeAll of new decoder not called yet
fromSample = markedSample;
break;
case READING: // need to abort decoder and restart with a new one
if (atSample != markedSample) { // ... unless skipping zero samples, which is ignored
fromSample = markedSample;
state = State.SEEKING; // block read will return false to abort decoding
notifyAll();
}
break;
case ERROR:
rethrow();
case CLOSED:
throw new IOException("Stream is closed");
default:
throw new IllegalStateException("Unhandled state: "+state);
}
}
}
Version: v4.3.2.2 (2026-05-16T17:03:34+02:00)
Raw file
Source code overview
Klipspringer home