// 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 java.io.*;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.WritableByteChannel;
import java.nio.file.*;
import java.text.DecimalFormat;
import java.time.*;
import java.time.format.DateTimeFormatter;
import javax.sound.sampled.*;
import net.avadeaux.klipspringer.codec.*;

public class DeviceFeeder {
    public static class Play {
        private final Device.Player player;
        private long sentFrames;

        private Play(Device.Player player) {
            this.player = player;
        }

        private synchronized void addBytes(long n) {
            sentFrames += n / player.format().fs();
        }

        public synchronized double time() throws IOException {
            return (double) (sentFrames - player.bufferedFrames()) / player.format().rate();
        }

        public void close() throws IOException {
            player.close();
        }
    }

    private static class Buffer {
        // Shared lock on buffer, constructed inside sync block.
        private class RLock implements AutoCloseable {
            final long writeNo = Buffer.this.writeNo;
            public void close() {
                synchronized (Buffer.this) {
                    readers--;
                    Buffer.this.notifyAll();
                }
            }
        }

        // Exclusive lock on buffer, constructed inside sync block.
        private class WLock implements AutoCloseable {
            public void close() {
                synchronized (Buffer.this) {
                    writer = false;
                    Buffer.this.notifyAll();
                }
            }
        }

        final ByteBuffer rbb;
        private int readers = 0;                // number of shared locks held
        private boolean writer = false;         // exclusive lock held
        private long writeNo = -1;              // timestamp of most recent exclusive lock

        Buffer(int capacity, ByteOrder order) {
            rbb = ByteBuffer.allocateDirect(capacity);
            rbb.order(order);
        }

        synchronized RLock getR(long minNo) {
            while (writer || minNo > writeNo && writeNo > -2) {
                try { wait(); } catch (InterruptedException e) { }
            }
            readers++;
            return new RLock();
        }

        synchronized void invalidate() {
            writeNo = -2;
            notifyAll();
        }

        synchronized WLock getW(long writeNo) {
            while (writer || readers > 0) {
                try { wait(); } catch (InterruptedException e) { }
            }
            writer = true;
            this.writeNo = writeNo;
            notifyAll();
            return new WLock();
        }
    }

    private static class Recorder implements Device.Player {
        private final PcmFormat format;
        private final WritableByteChannel channel;
        private long written = 0;

        Recorder(PcmFormat format, WritableByteChannel channel) {
            this.format = format;
            this.channel = channel;
        }

        public PcmFormat format() { return format; }

        public synchronized boolean write(ByteBuffer data) throws IOException {
            int n = data.remaining();
            channel.write(data);
            written += n - data.remaining();
            return !data.hasRemaining() && channel.isOpen();
        }

        public synchronized void close() throws IOException { channel.close(); }

        synchronized long bytePosition() { return written; }
    }

    private final static DateTimeFormatter dtf = DateTimeFormatter.ofPattern("uuuuMMdd'T'kkmmss.SSS").withZone(ZoneId.systemDefault());
    private final static DecimalFormat minutef = new DecimalFormat("#.##");

    private final Buffer[] buf = new Buffer[8];
    private final int bufSizeBits;
    private final Device.Tuner.Factory inFact;
    private final Path recPath;

    private PcmFormat requested = null;
    private PcmFormat active = null; // the most recent successfully requested format
    private int maxMinutes = 30;
    private Recorder recorder = null;
    private Instant recStartTime = null;
    private int outputs = 0;
    private double recLimit = 30 * 60;

    public DeviceFeeder(String inDevice,
                        PcmFormat defaultFormat,
                        double ibufsec,
                        Path recPath,
                        long minBufSamples)
        throws IOException
    {
        this.inFact = DeviceList.inputFactory(inDevice, ibufsec, 0);
        this.recPath = recPath;
        bufSizeBits = 64 - Long.numberOfLeadingZeros((minBufSamples-1)/buf.length);
        setFormat(defaultFormat);
    }

    private synchronized PcmFormat getActive() { return active; }
    private synchronized int getBufNo(long serial) { return (int) (serial % buf.length); }
    private synchronized double getRecLimit() { return recLimit; }
    public synchronized void setRecLimit(double seconds) { recLimit = seconds; }

    public void runFeed() throws Exception {
        long writeNo = 0;
        Device.Tuner in = null;
        boolean exhausted = false;
        while (!exhausted) {
            // If no outputs, release input, wait and try again.
            synchronized (this) {
                if (outputs == 0) {
                    if (in != null) { in.close(); in = null; }
                    try { wait(); } catch (InterruptedException e) { }
                    continue;
                }
            }

            // If input is down or new format requested, make a new input.
            synchronized (this) {
                if (in == null || !active.matches(requested)) {
                    if (in != null) { in.close(); in = null; }
                    try {
                        in = inFact.open(requested);
                    } catch (Exception ex) {
                        // Failure, attempt returning to working format.
                        if (System.getProperty("klipspringer.debug") != null) { ex.printStackTrace(); }
                        if (active == null || active.matches(requested)) { throw ex; }
                        requested = active;
                        notifyAll();
                        continue;
                    }
                    // Success, set active and allocate buffers to go with new format.
                    for (int i = 0; i < buf.length; i++) {
                        if (buf[i] != null) { buf[i].invalidate(); }
                        buf[i] = new Buffer(requested.getFrameSize() << bufSizeBits, requested.order());
                    }
                    active = requested;
                    notifyAll();
                }
            }

            // Read a batch.
            Buffer b = buf[getBufNo(writeNo)];
            try (Buffer.WLock lock = b.getW(writeNo++)) {
                b.rbb.clear();
                exhausted = !in.read(b.rbb);
            }
        }
    }

    // Called when a new output is created, returns the format it should use.
    private synchronized PcmFormat incrOutputs() {
        outputs++;
        notifyAll();
        while (active == null || !active.matches(requested)) {
            try { wait(); } catch (InterruptedException e) { }
        }
        return active;
    }

    // Called when an output ceases to function.
    private synchronized void decrOutputs() { outputs--; }

    private Play runFetch(Device.Player out) {
        Play state = new Play(out);
        new Thread("Fetch stream "+out) {
            public void run() {
                long readNo = 0, processed = 0;
                ByteBuffer[] wbb = new ByteBuffer[buf.length];
                PcmFormat inFormat = getActive();
                PcmBuffer fbb = inFormat.signed() != out.format().signed()
                    || inFormat.ss() != out.format().ss()
                    || inFormat.bigend() != out.format().bigend()
                    || inFormat.bips() != 8*inFormat.ss() && out.layout() != Device.BitLayout.LSB
                    ? PcmBuffer.of(ByteBuffer.allocateDirect(out.format().fs() << bufSizeBits), out.format(), out.layout())
                    : null;     // data can be copied verbatim
                for (int i = 0; i < buf.length; i++) {
                    wbb[i] = buf[i].rbb.duplicate();
                    wbb[i].order(buf[i].rbb.order());
                }
                while (true) {
                    long maxBytes = out == recorder ? (long) (getRecLimit()*out.format().getSampleRate()*out.format().getFrameSize()) : Long.MAX_VALUE;
                    if (processed >= maxBytes) { break; }
                    int i = getBufNo(readNo);
                    Buffer b = buf[i];
                    try (Buffer.RLock r = b.getR(readNo)) {
                        if (r.writeNo < 0 || !inFormat.matches(getActive())) { break; } // format change
                        if (r.writeNo > readNo) {
                            if (readNo == 0) { readNo = r.writeNo; } // first in this thread
                            else { System.err.println("stream outpaced"); break; }
                        }
                        int n = (int) Math.min(b.rbb.capacity(), maxBytes - processed);
                        wbb[i].clear();
                        ByteBuffer bb = wbb[i];
                        if (fbb != null) { // player wants data reformatted
                            bb = fbb.byteBuffer();
                            bb.clear();
                            fbb.put(wbb[i], inFormat);
                            bb.flip();
                        }
                        if (!out.write(bb)) { break; }
                        processed += n;
                        state.addBytes(n);
                    } catch (Exception ex) {
                        if (System.getProperty("klipspringer.debug") != null) { ex.printStackTrace(); }
                        break;
                    }
                    readNo++;
                }
                decrOutputs();
                try {
                    out.close();
                } catch (Exception ex) {
                    if (System.getProperty("klipspringer.debug") != null) { ex.printStackTrace(); }
                }
                if (out == recorder) {
                    synchronized (DeviceFeeder.this) {
                        recorder = null;
                        recStartTime = null;
                    }
                }
            }
        }.start();
        return state;
    }

    public Play fetchStream(Device.Player.Factory streamFact) throws IOException {
        PcmFormat fmt = streamFact.selectFormat(incrOutputs());
        Device.Player player = streamFact.open(fmt);
        return runFetch(player);
    }

    void foo() throws IOException { }

    public Play startRecord() throws IOException {
        PcmFormat fmt;
        synchronized (this) {
            if (recStartTime != null) { throw new IllegalStateException("record already in progress"); }
            fmt = incrOutputs();
            recStartTime = Instant.now();
        }
        Files.createDirectories(recPath);
        String fnam = "kliprec@"+dtf.format(recStartTime)+Track.rawParamString(fmt)+".raw";
        RandomAccessFile file = new RandomAccessFile(new File(recPath.toFile(), fnam), "rw");
        file.setLength(0);
        Recorder rec = new Recorder(fmt, file.getChannel());
        synchronized (this) { recorder = rec; }
        return runFetch(rec);
    }

    public void stopRecord() throws IOException {
        Device.Player player;
        synchronized (this) {
            // Check if recorder is null rather than if recStartTime is, because if recStartTime is
            // set but not recorder, closing the recorder output from here is impossible.
            player = recorder;
            if (player == null) {
                if (System.getProperty("klipspringer.debug") != null) {
                    throw new IllegalStateException("no record in progress");
                }
                return;
            }
        }
        player.close();
    }

    public void setFormat(PcmFormat fmt) throws UnsupportedFormatException {
        fmt = inFact.selectFormat(fmt);
        synchronized (this) {
            if (recStartTime != null) { throw new IllegalStateException("cannot set format while recording"); }
            if (requested == null
                || fmt.rate() != requested.rate()
                || fmt.bips() != requested.bips()
                || fmt.channels() != requested.channels())
                { requested = fmt; }
        }
    }

    public void deviceStatus(Writer w) throws IOException {
        Recorder rec;
        PcmFormat fmt;
        double limSecs;
        synchronized (this) {
            rec = recorder;
            fmt = active;
            limSecs = recLimit;
        }
        if (fmt != null) {
            w.append("\"channels\":").append(PlayState.channelsString(fmt.channels()))
                .append(",\"bits_per_sample\":").append(Integer.toString(fmt.bips()))
                .append(",\"sample_rate\":\"").append(Integer.toString(fmt.rate())).append("&nbsp;Hz\"")
                .append(",\"rec_channels\":"+fmt.channels())
                .append(",\"rec_bits\":\""+Track.rawBipsString(fmt)+"\"")
                .append(",\"rec_rate\":"+(int) fmt.rate())
                .append(",");

        }
        w.append("\"rec_limit\":"+minutef.format(limSecs/60));
        if (rec != null) {
            w.append(",\"rec_time\":").append(Double.toString(rec.bytePosition()/fmt.fs()/fmt.rate()));
        }
    }
}

Version: v4.3.2.2 (2026-05-16T17:03:34+02:00)
Raw file
Source code overview
Klipspringer home