// Copyright 2014-2025 Jesper Larsson
//
// This file is part of Klipspringer, <https://klipspringer.eavadeaux.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;
import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel;
import java.nio.channels.ClosedChannelException;
import java.io.IOException;
import java.util.concurrent.Executor;
/** Common base class for writers that use external libraries for encoding data to a channel. */
public abstract class ExternalEncodedStream implements AudioStream {
/** Factory interface for subclasses to provide the constructor with encoder handle. */
interface EncoderRecordFactory {
/** Sets up the external encoder and returns a handle. */
long create(Receiver receiver, PcmFormat format, int bufferFrames) throws IOException;
}
/** Receiver of encoded data. A separate class from {@link ExternalEncodedStream} mainly because
* callback to {@link #receive(ByteBuffer, int)} may be invoked during {@link
* EncoderRecordFactory#create(Receiver, PcmFormat, int)}, and the receiver object should
* therefore be fully constructed before the encoder record is created in the stream
* constructor.
*
* To prevent deadlocks, all synchronization in {@link ExternalEncodedStream} is on the
* receiver object rather than the stream, even when protecting fields in the stream class.
*/
static class Receiver {
// Runnable for keeping a thread alive until the destination has played the full input.
private class KeepAlive implements Runnable {
private final long writtenFrames;
private final int allowUnplayedFrames;
KeepAlive(long writtenFrames, int allowUnplayedFrames) {
this.writtenFrames = writtenFrames;
this.allowUnplayedFrames = allowUnplayedFrames;
}
public void run() {
synchronized (Receiver.this) {
while (writtenFrames - playedFrames > allowUnplayedFrames) {
long before = System.currentTimeMillis();
try { Receiver.this.wait(keepAliveTimeoutMillis); } catch (InterruptedException e) { }
if (System.currentTimeMillis() - before >= keepAliveTimeoutMillis) { break; }
}
streamFinished = true;
}
}
}
private final WritableByteChannel channel; // output channel
private final long writeaheadFrames; // max unplayed written to recipient
private final long playTimeoutMillis; // time to allow recipient to call back
private final long keepAliveTimeoutMillis; // time before KeepAlive gives up
private final int holdingBytes; // amount recipient might hold unplayed
private long playedFrames = 0; // according to playedMillis callback
private long processedFrames = 0; // minimum frames assumed processed
private int heldFrames = 0; // frames possibly held unprocessed
private int heldBytes = 0; // amount of data assumed to be held
private long receivedFrames = 0; // frames output by the encoder
private boolean streamFinished = false; // set to true when KeepAlive stops
private Receiver(WritableByteChannel channel,
long writeaheadFrames,
long playTimeoutMillis,
long keepAliveTimeoutMillis,
int holdingBytes)
throws IOException
{
this.channel = channel;
this.writeaheadFrames = writeaheadFrames;
this.playTimeoutMillis = playTimeoutMillis;
this.keepAliveTimeoutMillis = keepAliveTimeoutMillis;
this.holdingBytes = holdingBytes;
}
/** Method to be called from the encoder to pass encoded data on to the channel. */
public synchronized void receive(ByteBuffer edata, int frames) throws IOException {
while (processedFrames - playedFrames > writeaheadFrames) {
long waitStart = System.currentTimeMillis();
if (playTimeoutMillis > 0) {
try { wait(playTimeoutMillis); } catch (InterruptedException e) { }
}
if (System.currentTimeMillis() - waitStart >= playTimeoutMillis) { throw new IOException("Stream timeout"); }
if (!channel.isOpen()) { throw new ClosedChannelException(); }
}
int b = heldBytes + edata.remaining();
channel.write(edata);
if (b >= holdingBytes) {
processedFrames += heldFrames; // add the samples held before this call
heldFrames = frames; // pessimistically assume new are all held
heldBytes = b % (holdingBytes+1);
} else {
heldFrames += frames; // assume everything in this call is held
heldBytes = b;
}
receivedFrames += frames;
notifyAll();
}
private synchronized void playedFrames(long playedFrames) {
this.playedFrames = playedFrames;
notifyAll();
}
private KeepAlive keepAlive(long writtenFrames, int allowUnplayedFrames) { return new KeepAlive(writtenFrames, allowUnplayedFrames); }
}
// Lock to make sure that the decoder is not deallocated by close while in the middle of writing
// or draining in another thread. (Also protects write and drain from each other, although they
// are most reasonably called in the same thread.) Can only be used inside sync block.
private class ExLock implements AutoCloseable {
ExLock() {
while (writing) {
try { receiver.wait(); } catch (InterruptedException e) { }
}
writing = true;
}
public void close() {
writing = false;
receiver.notifyAll();
}
}
private final Receiver receiver; // the receive callback object
private final PcmFormat format; // audio format
private final int bufferFrames; // limit on single write to external
private final long handle; // external encoder handle
private final Executor keepAliveExec; // for executing KeepAlive
private long writtenFrames = 0; // frames written to the encoder
private boolean closed = false; // close has been called
private boolean writing = false; // for ExLock
ExternalEncodedStream(WritableByteChannel channel,
PcmFormat format,
int bufferFrames,
double writeaheadSecs,
double playTimeoutSecs,
int holdingBytes,
EncoderRecordFactory efact,
Executor keepAliveExec,
double keepAliveTimeoutSecs)
throws IOException
{
receiver = new Receiver(channel,
(long) (writeaheadSecs * format.rate()),
(long) (playTimeoutSecs * 1000),
(long) (keepAliveTimeoutSecs * 1000),
holdingBytes);
this.format = format;
this.bufferFrames = bufferFrames;
this.keepAliveExec = keepAliveExec;
handle = efact.create(receiver, format, bufferFrames);
}
/** Gets the stream audio format. */
public final PcmFormat format() { return format; }
/** Checks if the recipient appears to still be processing data. If a keep-alive executor is
* available, this method continues checking for recipient activity even after the stream is
* closed. Otherwise, it returns false as soon as the stream is closed.
*/
public boolean streaming() {
synchronized (receiver) {
return !closed || keepAliveExec != null && receiver.streamFinished;
}
}
/** Writes data to an external buffer, whose content is then immediately encoded, and encoded
* data is sent to the receiver.
*/
public final boolean write(ByteBuffer data) throws IOException {
synchronized (receiver) {
if (closed) { return false; }
if (!data.isDirect()) { return ByteTransfer.writeDirect(data, this, format.fs()); }
try {
try (ExLock lk = new ExLock()) {
while (data.hasRemaining()) {
int p = data.position();
int n = Math.min(bufferFrames*format.fs(), data.remaining());
int m = n/format.fs();
write(handle, data, p, m);
writtenFrames += m;
data.position(p+n);
}
return true;
}
} catch (IOException e) {
close();
throw e;
} catch (RuntimeException e) {
close();
throw e;
}
}
}
public void close() throws IOException {
synchronized (receiver) {
if (!closed) {
closed = true;
try {
if (receiver.channel != null) { receiver.channel.close(); }
} catch (Exception ex) {
if (System.getProperty("klipspringer.debug") != null) { ex.printStackTrace(); }
}
receiver.notifyAll();
try (ExLock lk = new ExLock()) {
free(handle);
}
}
}
}
/** Forces the encoder to write any retained audio, and closes the stream. Although all data is
* written to the recipient before this method returns, the recipient may not have played (or
* otherwise processed) the data it has received at that point, which can continue even after
* the stream is closed. If a keep-alive executor is available, it is used for executing a wait
* loop that continues for as long as callbacks to {@link #playedMillis(long)} continue. The
* idea is that the keep-alive executor can run a wait loop in a non-daemon thread, to keep the
* process alive and stop premature termination of playback.
*/
public void drain() throws IOException {
synchronized (receiver) {
if (!closed) {
try (ExLock lk = new ExLock()) {
finish(handle);
}
close();
if (keepAliveExec != null) {
// Allow 1000 bytes not to be played, to not get stuck due to rounding etc.
keepAliveExec.execute(receiver.keepAlive(writtenFrames, (int) Math.ceil(1000.0/format.fs())));
}
}
}
}
public void playedMillis(long time) {
receiver.playedFrames(time * format.rate() / 1000);
}
/** Gets an alternative PCM writing interface that is suitable when the external encoder is set
* up to send encoded data to a file rather than streaming.
*/
PcmWriter fileEncoder(PcmFormat dataFormat, Device.BitLayout lout) throws IOException {
return new PcmWriter() {
final int bfs = format.fs(), dfs = dataFormat.fs();
final ByteBuffer buf = ByteBuffer.allocateDirect(bufferFrames*bfs);
final PcmBuffer fbuf = PcmBuffer.of(buf, format, lout);
boolean closed = false;
public boolean write(ByteBuffer data) throws IOException {
int p = data.position(), q = data.limit();
while (p < q) {
int m = Math.min((q-p)/dfs, buf.capacity()/bfs);
p += m*dfs;
data.limit(p);
buf.clear();
fbuf.put(data, dataFormat);
ExternalEncodedStream.this.write(handle, buf, 0, m);
}
return true;
}
public void close() throws IOException {
if (!closed) {
try {
finish(handle);
ExternalEncodedStream.this.close();
} finally {
closed = true;
}
}
}
};
}
/** Writes data to the externally allocated encoder. */
abstract void write(long handle, ByteBuffer data, int pos, int frames) throws IOException;
/** Sends any remaining data to the destination. Called by {@link #drain()}. */
abstract void finish(long handle) throws IOException;
/** Releases external resources. Called by {@link close()}. */
abstract void free(long handle);
}
Version: v4.3.2.2 (2026-05-16T17:03:34+02:00)
Raw file
Source code overview
Klipspringer home