<NME> MosaicExample.java
<BEF> package examples;

import com.github.kokorin.jaffree.LogLevel;
import com.github.kokorin.jaffree.StreamType;
import com.github.kokorin.jaffree.ffmpeg.FFmpeg;
import com.github.kokorin.jaffree.ffmpeg.FFmpegResult;
import com.github.kokorin.jaffree.ffmpeg.Frame;
import com.github.kokorin.jaffree.ffmpeg.FrameConsumer;
import com.github.kokorin.jaffree.ffmpeg.FrameInput;
import com.github.kokorin.jaffree.ffmpeg.FrameOutput;
import com.github.kokorin.jaffree.ffmpeg.FrameProducer;
import com.github.kokorin.jaffree.ffmpeg.Stream;
import com.github.kokorin.jaffree.ffmpeg.UrlInput;
import com.github.kokorin.jaffree.ffmpeg.UrlOutput;
import com.github.kokorin.jaffree.ffprobe.FFprobe;
import com.github.kokorin.jaffree.ffprobe.FFprobeResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.awt.Color;
import java.awt.Graphics;
import java.awt.Image;
import java.awt.image.BufferedImage;
import java.awt.image.ImageObserver;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Deque;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * Note: for some reason ffmpeg 3.3.1 fails when using amix or amerge filters (in this example)
 * Use ffmpeg 4.0 at least, amix works (while amerge still fails)
 */
public class MosaicExample {
    private final Path ffmpegBin;
    private final List<String> inputs;
    private final int sampleRate = 44100;
    private final int frameRate = 25;

    public static final Logger LOGGER = LoggerFactory.getLogger(MosaicExample.class);

    public MosaicExample(String ffmpegBin, List<String> inputs) {
        this.ffmpegBin = Paths.get(ffmpegBin);
        this.inputs = inputs;
    }

    public void execute() {
        final List<FFmpegResult> results = new CopyOnWriteArrayList<>();
        List<FrameIterator> frameIterators = new ArrayList<>();

        for (int i = 0; i < inputs.size(); i++) {
            String input = inputs.get(i);

            boolean hasAudioStream = false;
            // We must test input for audio stream
            // Other corner-cases (no video stream, multiple audio stream are not considered)
            FFprobeResult probeResult = FFprobe.atPath(ffmpegBin)
                    .setShowStreams(true)
                    .setInput(Paths.get(input))
                    .execute();
            for (com.github.kokorin.jaffree.ffprobe.Stream stream : probeResult.getStreams()) {
                if (StreamType.AUDIO == stream.getCodecType()) {
                    hasAudioStream = true;
                    break;
                }
            }

            FrameIterator frameIterator = new FrameIterator();
            frameIterators.add(frameIterator);

            final FFmpeg ffmpeg = FFmpeg.atPath(ffmpegBin)
                    .addInput(UrlInput
                            .fromUrl(input)
                            .setDuration(15_000))
                    .addOutput(FrameOutput
                            .withConsumer(frameIterator.getConsumer())
                            .setFrameRate(frameRate)
                            .addArguments("-ac", "1")
                            .addArguments("-ar", Integer.toString(sampleRate))
                    )
                    .setContextName("input" + i);

            if (!hasAudioStream) {
                ffmpeg
                        .addInput(
                                UrlInput.fromUrl("anullsrc")
                                        .setFormat("lavfi")
                        )
                        .addArgument("-shortest");
            }

            Thread ffmpegThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    FFmpegResult result = ffmpeg.execute();
                    results.add(result);

                }
            }, "Reader-" + i + "-main");
            ffmpegThread.setDaemon(true);
            ffmpegThread.start();
        }

        FrameProducer frameProducer = produceMosaic(frameIterators);

        FFmpegResult mosaicResult = FFmpeg.atPath(ffmpegBin)
                .addInput(
                        FrameInput
                                .withProducer(frameProducer)
                                .setFrameRate(frameRate)
                                .setFrameOrderingBuffer(5_000L)
                )
                .setOverwriteOutput(true)
                .setLogLevel(LogLevel.TRACE)
                .addOutput(UrlOutput
                                .toUrl("mosaic.mp4")
                        //.addMap(0)
                )
                .addArguments("-filter_complex", "amix=inputs=" + inputs.size())
                .setContextName("result")
                .execute();
    }

    private FrameProducer produceMosaic(final List<FrameIterator> frameIterators) {
        final int rows = (int) Math.round(Math.sqrt(frameIterators.size()));
        final int columns = (int) Math.ceil(1.0 * frameIterators.size() / rows);

        return new FrameProducer() {
            private final int elementWidth = 320;
            private final int elementHeight = 240;
            private final int mosaicWidth = columns * elementWidth;
            private final int mosaicHeight = rows * elementHeight;
            private final long videoFrameDuration = 1000 / frameRate;

            private final Map<Integer, Deque<Frame>> audioQueues = new HashMap<>();
            private long timecode = 0;
            private final Frame[] nextVideoFrames = new Frame[frameIterators.size()];
            private long nextVideoFrameTimecode = 0;
            private long nextAudioFrameTimecode = 0;

            @Override
            public List<Stream> produceStreams() {
                List<Stream> streams = new ArrayList<>();
                streams.add(
                        new Stream()
                                .setId(0)
                                .setType(Stream.Type.VIDEO)
                                .setTimebase(1000L)
                                .setWidth(mosaicWidth)
                                .setHeight(mosaicHeight)
                );

                // We copy all audio streams to output and merge tham with amerge filter
                for (int i = 0; i < frameIterators.size(); i++) {
                    streams.add(
                            new Stream()
                                    .setId(1 + i)
                                    .setType(Stream.Type.AUDIO)
                                    .setTimebase((long) sampleRate)
                                    .setChannels(1)
                                    .setSampleRate(sampleRate)
                    );
                }

                return streams;
            }

            @Override
            public Frame produce() {
                Frame result = null;
                if (nextVideoFrameTimecode <= timecode) {
                    if (nextVideoFrameTimecode == 0) {
                        readNextVideoFrames(nextVideoFrameTimecode);
                    }

                    result = produceVideoFrame(nextVideoFrames);
                    // We have to read video frames ahead, otherwise we will have no audio frames read
                    readNextVideoFrames(nextVideoFrameTimecode);
                } else if (nextAudioFrameTimecode <= timecode) {
                    result = produceAudioFrame();
                }

                timecode = Math.min(nextVideoFrameTimecode, nextAudioFrameTimecode);

                return result;
            }

            public Frame produceVideoFrame(Frame[] videoFrames) {
                BufferedImage mosaic = new BufferedImage(mosaicWidth, mosaicHeight, BufferedImage.TYPE_3BYTE_BGR);
                Graphics mosaicGraphics = mosaic.getGraphics();
                mosaicGraphics.setColor(new Color(0));
                mosaicGraphics.fillRect(0, 0, mosaicWidth, mosaicHeight);

                boolean atLeastHasOneElement = false;

                for (int i = 0; i < videoFrames.length; i++) {
                    Frame videoFrame = videoFrames[i];
                    if (videoFrame == null) {
                        continue;
                    }

                    atLeastHasOneElement = true;

                    BufferedImage element = videoFrame.getImage();
                    int row = i / columns;
                    int column = i % columns;

                    ImageObserver observer = new ImageObserver() {
                        @Override
                        public boolean imageUpdate(Image img, int infoflags, int x, int y, int width, int height) {
                            return false;
                        }
                    };

                    int dx1 = column * elementWidth;
                    int dy1 = row * elementHeight;
                    int dx2 = dx1 + elementWidth;
                    int dy2 = dy1 + elementHeight;

                    mosaicGraphics.drawImage(element,
                            dx1, dy1, dx2, dy2,
                            0, 0, element.getWidth(), element.getHeight(),
                            observer
                    );
                }

                if (!atLeastHasOneElement) {
                    return null;
                }

                Frame result = Frame.createVideoFrame(0, nextVideoFrameTimecode, mosaic);

                nextVideoFrameTimecode += videoFrameDuration;

                return result;
            }

            private Frame produceAudioFrame() {
                long minPts = Long.MAX_VALUE;
                long nextPts = Long.MAX_VALUE;
                int minI = -1;
                for (int i = 0; i < frameIterators.size(); i++) {
                    Deque<Frame> aQueue = audioQueues.get(i);
                    Frame frame = aQueue.peekFirst();
                    if (frame == null) {
                        continue;
                    }

                    long queuePts = frame.getPts();
                    if (queuePts < minPts) {
                        nextPts = minPts;
                        minPts = queuePts;
                        minI = i;
                        continue;
                    }
                    if (queuePts < nextPts) {
                        nextPts = queuePts;
                    }
                }

                if (minI == -1) {
                    return null;
                }

                Frame aFrame = audioQueues.get(minI).pollFirst();
                if (aFrame == null) {
                    return null;
                }
                aFrame = Frame.createAudioFrame(1 + minI, aFrame.getPts(), aFrame.getSamples());

                if (nextPts != Long.MAX_VALUE) {
                    nextAudioFrameTimecode = 1000L * nextPts / sampleRate;
                } else {
                    nextAudioFrameTimecode = nextVideoFrameTimecode;
                }

                return aFrame;
            }

            // All video input streams have forced frameRate, so we know that in every stream
            // frames will occur with the same timestamp
            private void readNextVideoFrames(long videoTs) {
                for (int i = 0; i < frameIterators.size(); i++) {
                    FrameIterator iter = frameIterators.get(i);

                    if (!iter.hasNext) {
                        nextVideoFrames[i] = null;
                    }

                    while (iter.hasNext) {
                        Frame frame = iter.next();
                        if (frame == null) {
                            break;
                        }

                        Stream stream = iter.getStream(frame.getStreamId());
                        if (stream == null) {
                            break;
                        }

                        switch (stream.getType()) {
                            case VIDEO:
                                nextVideoFrames[i] = frame;
                                break;
                            case AUDIO:
                                Deque<Frame> aQueue = audioQueues.get(i);
                                if (aQueue == null) {
                                    aQueue = new LinkedList<>();
                                    audioQueues.put(i, aQueue);
                                }
                                aQueue.addLast(frame);
                                break;
                        }
                        long frameTs = 1000L * frame.getPts() / stream.getTimebase();
                        if (frameTs >= videoTs) {
                            break;
                        }
                    }
                }
            }
        };
    }

    public static void main(String[] args) {
        Iterator<String> argIter = Arrays.asList(args).iterator();

        String ffmpegBin = null;
        List<String> inputs = new ArrayList<>();

        while (argIter.hasNext()) {
            String arg = argIter.next();

            if ("-ffmpeg_bin".equals(arg)) {
                ffmpegBin = argIter.next();
            } else {
                inputs.add(arg);
            }
        }

        if (ffmpegBin == null || inputs.isEmpty()) {
            System.err.println("Arguments: -ffmpeg_bin </path/to/ffmpeg/bin>");
            return;
        }

        new MosaicExample(ffmpegBin, inputs).execute();
    }

    public static class FrameIterator implements Iterator<Frame> {
        private volatile boolean hasNext = true;
        private volatile Frame next = null;
        private volatile List<Stream> tracks;

        private final FrameConsumer consumer = new FrameConsumer() {
            @Override
            public void consumeStreams(List<Stream> tracks) {
                FrameIterator.this.tracks = tracks;
            }

            @Override
            public void consume(Frame frame) {
                while (next != null) {
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        LOGGER.warn("Exception while supplying frame", e);
                    }
                }

                hasNext = frame != null;
                next = frame;
            }
        };

        @Override
        public boolean hasNext() {
            waitForNextFrame();
            return hasNext;
        }

        @Override
        public Frame next() {
            waitForNextFrame();
            Frame result = next;
            next = null;
            return result;
        }

        @Override
        public void remove() {
            throw new UnsupportedOperationException("remove");
        }

        public FrameConsumer getConsumer() {
            return consumer;
        }

        public List<Stream> getTracks() {
            return tracks;
        }

        public Stream getStream(int id) {
            for (Stream stream : tracks) {
                if (stream.getId() == id) {
                    return stream;
                }
            }

            return null;
        }

        private void waitForNextFrame() {
            while (hasNext && next == null) {
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    LOGGER.warn("Exception while waiting for frame", e);
                }
            }
        }
    }
}

<MSG> Refactor FrameInput & FrameOutput, improve tests

<DFF> @@ -9,6 +9,7 @@ import com.github.kokorin.jaffree.ffmpeg.FrameConsumer;
 import com.github.kokorin.jaffree.ffmpeg.FrameInput;
 import com.github.kokorin.jaffree.ffmpeg.FrameOutput;
 import com.github.kokorin.jaffree.ffmpeg.FrameProducer;
+import com.github.kokorin.jaffree.ffmpeg.ImageFormats;
 import com.github.kokorin.jaffree.ffmpeg.Stream;
 import com.github.kokorin.jaffree.ffmpeg.UrlInput;
 import com.github.kokorin.jaffree.ffmpeg.UrlOutput;
@@ -113,9 +114,8 @@ public class MosaicExample {
         FFmpegResult mosaicResult = FFmpeg.atPath(ffmpegBin)
                 .addInput(
                         FrameInput
-                                .withProducer(frameProducer)
+                                .withProducer(frameProducer, ImageFormats.BGR24, 5_000L)
                                 .setFrameRate(frameRate)
-                                .setFrameOrderingBuffer(5_000L)
                 )
                 .setOverwriteOutput(true)
                 .setLogLevel(LogLevel.TRACE)

{'commit': 'diff --git a/src/main/java/com/github/kokorin/jaffree/ffmpeg/Frame.java b/src/main/java/com/github/kokorin/jaffree/ffmpeg/Frame.java\nindex 49ecdc5..529f2c9 100644\n--- a/src/main/java/com/github/kokorin/jaffree/ffmpeg/Frame.java\n+++ b/src/main/java/com/github/kokorin/jaffree/ffmpeg/Frame.java\n@@ -40,7 +40,7 @@ public class Frame {\n      * @param image    video frame image\n      * @see Stream#getTimebase()\n      */\n-    // TODO make static mathod\n+    // TODO make static method\n     public Frame(final int streamId, final long pts, final BufferedImage image) {\n         this(streamId, pts, image, null);\n     }\ndiff --git a/src/main/java/com/github/kokorin/jaffree/ffmpeg/FrameInput.java b/src/main/java/com/github/kokorin/jaffree/ffmpeg/FrameInput.java\nindex e21956b..c5653d1 100644\n--- a/src/main/java/com/github/kokorin/jaffree/ffmpeg/FrameInput.java\n+++ b/src/main/java/com/github/kokorin/jaffree/ffmpeg/FrameInput.java\n@@ -27,7 +27,6 @@ import org.slf4j.LoggerFactory;\n import java.io.IOException;\n import java.io.OutputStream;\n import java.net.Socket;\n-import java.util.concurrent.TimeUnit;\n \n /**\n  * Allows to supply ffmpeg with audio &amp; video frames constructed in Java.\n@@ -44,54 +43,39 @@ public class FrameInput extends TcpInput<FrameInput> implements Input {\n     /**\n      * Creates {@link FrameInput} for {@link FFmpeg}.\n      *\n-     * @param producer frame producer\n+     * @param frameWriter frame writer\n+     * @see NutFrameWriter\n      */\n-    public FrameInput(final FrameProducer producer) {\n-        this(new FrameInputNegotiator(producer));\n+    protected FrameInput(final FrameWriter frameWriter, final String format) {\n+        this(new FrameInputNegotiator(frameWriter), format);\n     }\n \n-    protected FrameInput(FrameInputNegotiator negotiator) {\n+    private FrameInput(final FrameInputNegotiator negotiator, final String format) {\n         super(negotiator);\n         this.negotiator = negotiator;\n-        super.setFormat("nut");\n-    }\n-\n-    /**\n-     * Whether produced video stream should contain alpha channel.\n-     *\n-     * @param containsAlphaChannel alpha channel\n-     * @return this\n-     */\n-    // TODO rename method\n-    public FrameInput produceAlpha(final boolean containsAlphaChannel) {\n-        negotiator.setAlpha(containsAlphaChannel);\n-        return this;\n+        super.setFormat(format);\n     }\n \n     /**\n+     * {@inheritDoc}\n+     * <p>\n      * <b>It\'s strongly recommended</b> to specify videoFrameRate for video producing.\n      * <p>\n      * Otherwise conversion can be very slow (20-50 times slower) and even can result in\n      * corrupted video.\n-     *\n-     * @param frameRate video frames per second\n-     * @return this\n      */\n     @Override\n-    public FrameInput setFrameRate(final Number frameRate) {\n-        negotiator.setFrameRateSet(true);\n+    public FrameInput setFrameRate(Number frameRate) {\n         return super.setFrameRate(frameRate);\n     }\n \n     /**\n+     * {@inheritDoc}\n+     * <p>\n      * <b>It\'s strongly recommended</b> to specify videoFrameRate for video producing.\n      * <p>\n      * Otherwise conversion can be very slow (20-50 times slower) and even can result in\n      * corrupted video.\n-     *\n-     * @param streamSpecifier stream specifier\n-     * @param frameRate       video frames per second\n-     * @return this\n      */\n     @Override\n     public FrameInput setFrameRate(final String streamSpecifier, final Number frameRate) {\n@@ -99,81 +83,90 @@ public class FrameInput extends TcpInput<FrameInput> implements Input {\n         return super.setFrameRate(streamSpecifier, frameRate);\n     }\n \n+    @Override\n+    public final FrameInput setFormat(String format) {\n+        throw new JaffreeException("Format can\'t be changed");\n+    }\n+\n     /**\n-     * Sets frame ordering buffer while producing NUT video. Default is 200 ms.\n-     * <p>\n-     * Frame ordering buffer allows {@link FrameProducer} to produce frame without strict ordering\n-     * (which is required by NUT format).\n+     * Creates {@link FrameInput} with specified frame producer.\n      * <p>\n-     * Note: high values may cause OutOfMemoryError or performance degradation.\n+     * Note: frame producer should produce video frames in BGR24 format.\n      *\n-     * @param bufferTime buffer time\n-     * @param unit       time unit\n-     * @return this\n+     * @param producer frame producer\n+     * @return FrameInput\n+     * @see ImageFormats#BGR24\n      */\n-    public FrameInput setFrameOrderingBuffer(final long bufferTime, final TimeUnit unit) {\n-        return setFrameOrderingBuffer(unit.toMillis(bufferTime));\n+    public static FrameInput withProducer(final FrameProducer producer) {\n+        return withProducer(producer, ImageFormats.BGR24);\n     }\n \n     /**\n-     * Sets frame ordering buffer while producing NUT video. Default is 200 ms.\n-     * <p>\n-     * Frame ordering buffer allows {@link FrameProducer} to produce frame without strict ordering\n-     * (which is required by NUT format).\n+     * Creates {@link FrameInput} with specified frame producer with alpha channel.\n      * <p>\n-     * Note: high values may cause OutOfMemoryError or performance degradation.\n+     * Note: frame producer should produce video frames in ABGR format.\n      *\n-     * @param bufferTimeMillis buffer time in milliseconds\n-     * @return this\n+     * @param producer frame producer\n+     * @return FrameInput\n+     * @see ImageFormats#ABGR\n      */\n-    public FrameInput setFrameOrderingBuffer(final long bufferTimeMillis) {\n-        negotiator.setFrameOrderingBufferMillis(bufferTimeMillis);\n-        return this;\n+    public static FrameInput withProducerAlpha(final FrameProducer producer) {\n+        return withProducer(producer, ImageFormats.ABGR);\n     }\n \n-    @Override\n-    public final FrameInput setFormat(String format) {\n-        throw new JaffreeException("Format can\'t be changed");\n+    /**\n+     * Creates {@link FrameInput} with specified frame producer and image format.\n+     *\n+     * @param producer    frame producer\n+     * @param imageFormat video frame image format\n+     * @return FrameInput\n+     * @see ImageFormats\n+     */\n+    public static FrameInput withProducer(final FrameProducer producer, final ImageFormat imageFormat) {\n+        return withProducer(producer, imageFormat, 200);\n     }\n \n     /**\n-     * Creates {@link FrameInput} with {@link FrameProducer}.\n+     * Creates {@link FrameInput} with specified frame producer, format and frame ordering buffer\n+     * <p>\n+     * Frame ordering buffer allows {@link FrameProducer} to produce frame without strict ordering\n+     * (which is required by NUT format).\n+     * <p>\n+     * Note: too long frame ordering buffer may cause {@link OutOfMemoryError} or performance degradation.\n      *\n-     * @param producer frame producer\n+     * @param producer                  frame producer\n+     * @param imageFormat               video frame image format\n+     * @param frameOrderingBufferMillis frame ordering buffer milliseconds\n      * @return FrameInput\n+     * @see ImageFormats\n      */\n-    public static FrameInput withProducer(final FrameProducer producer) {\n-        return new FrameInput(producer);\n+    public static FrameInput withProducer(final FrameProducer producer, final ImageFormat imageFormat,\n+                                          long frameOrderingBufferMillis) {\n+        return new FrameInput(\n+                new NutFrameWriter(producer, imageFormat, frameOrderingBufferMillis),\n+                "nut"\n+        );\n+    }\n+\n+    protected interface FrameWriter {\n+        void write(OutputStream outputStream) throws IOException;\n     }\n \n     @ThreadSafe\n     protected static class FrameInputNegotiator implements TcpNegotiator {\n-        private final FrameProducer producer;\n+        private final FrameWriter frameWriter;\n \n-        @GuardedBy("this")\n-        private boolean alpha;\n         @GuardedBy("this")\n         private boolean frameRateSet;\n-        @GuardedBy("this")\n-        private Long frameOrderingBufferMillis;\n-\n-        public FrameInputNegotiator(FrameProducer producer) {\n-            this.producer = producer;\n-        }\n \n-        public synchronized void setAlpha(final boolean alpha) {\n-            this.alpha = alpha;\n+        public FrameInputNegotiator(FrameWriter frameWriter) {\n+            this.frameWriter = frameWriter;\n         }\n \n         public synchronized void setFrameRateSet(final boolean frameRateSet) {\n             this.frameRateSet = frameRateSet;\n         }\n \n-        public synchronized void setFrameOrderingBufferMillis(\n-                final Long frameOrderingBufferMillis) {\n-            this.frameOrderingBufferMillis = frameOrderingBufferMillis;\n-        }\n-\n         @Override\n         public synchronized void negotiate(final Socket socket) throws IOException {\n             if (!frameRateSet) {\n@@ -182,9 +175,8 @@ public class FrameInput extends TcpInput<FrameInput> implements Input {\n                         + "and may produce corrupted video");\n             }\n \n-            NutFrameWriter supplier = new NutFrameWriter(producer, alpha, frameOrderingBufferMillis);\n             try (OutputStream output = socket.getOutputStream()) {\n-                supplier.write(output);\n+                frameWriter.write(output);\n             }\n         }\n     }\ndiff --git a/src/main/java/com/github/kokorin/jaffree/ffmpeg/FrameOutput.java b/src/main/java/com/github/kokorin/jaffree/ffmpeg/FrameOutput.java\nindex 8390d6c..8d6a382 100644\n--- a/src/main/java/com/github/kokorin/jaffree/ffmpeg/FrameOutput.java\n+++ b/src/main/java/com/github/kokorin/jaffree/ffmpeg/FrameOutput.java\n@@ -20,7 +20,6 @@ package com.github.kokorin.jaffree.ffmpeg;\n import com.github.kokorin.jaffree.JaffreeException;\n import com.github.kokorin.jaffree.StreamType;\n import com.github.kokorin.jaffree.net.TcpNegotiator;\n-import net.jcip.annotations.GuardedBy;\n import net.jcip.annotations.ThreadSafe;\n \n import java.io.IOException;\n@@ -31,38 +30,24 @@ import java.net.Socket;\n  * Allows to consume in Java audio &amp; video frames produced by ffmpeg.\n  */\n public class FrameOutput extends TcpOutput<FrameOutput> implements Output {\n-    private final FrameOutputNegotiator negotiator;\n-\n-    private static final String PIXEL_FORMAT_ALPHA = "abgr";\n-    private static final String PIXEL_FORMAT_RGB = "bgr24";\n \n     /**\n-     * Creates {@link FrameOutput}.\n+     * Create {@link FrameOutput} for {@link FFmpeg}\n      *\n-     * @param consumer frame consumer\n-     */\n-    public FrameOutput(final FrameConsumer consumer) {\n-        this(new FrameOutputNegotiator(consumer));\n-    }\n-\n-    protected FrameOutput(FrameOutputNegotiator negotiator) {\n-        super(negotiator);\n-        this.negotiator = negotiator;\n-        super.setFormat("nut");\n-        super.setCodec(StreamType.VIDEO.code(), "rawvideo");\n-        super.setCodec(StreamType.AUDIO.code(), "pcm_s32be");\n-        super.setPixelFormat(null, PIXEL_FORMAT_RGB);\n-    }\n-\n-    /**\n-     * @param alpha true if video contains alpha channel and it should be extracted\n-     * @return this\n+     * @param frameReader frame reader\n+     * @param format      media format\n+     * @param videoCodec  video codec\n+     * @param pixelFormat pixel format\n+     * @param audioCodec  audio codec\n+     * @see NutFrameReader\n      */\n-    public FrameOutput setAlpha(boolean alpha) {\n-        negotiator.setAlpha(alpha);\n-        String pixelFormat = alpha ? PIXEL_FORMAT_ALPHA : PIXEL_FORMAT_RGB;\n+    protected FrameOutput(FrameReader frameReader, String format,\n+                          String videoCodec, String pixelFormat, String audioCodec) {\n+        super(new FrameOutputNegotiator(frameReader));\n+        super.setFormat(format);\n+        super.setCodec(StreamType.VIDEO.code(), videoCodec);\n         super.setPixelFormat(null, pixelFormat);\n-        return this;\n+        super.setCodec(StreamType.AUDIO.code(), audioCodec);\n     }\n \n     @Override\n@@ -107,7 +92,7 @@ public class FrameOutput extends TcpOutput<FrameOutput> implements Output {\n      * @return FrameOutput\n      */\n     public static FrameOutput withConsumer(final FrameConsumer consumer) {\n-        return new FrameOutput(consumer);\n+        return withConsumer(consumer, ImageFormats.BGR24);\n     }\n \n     /**\n@@ -117,28 +102,31 @@ public class FrameOutput extends TcpOutput<FrameOutput> implements Output {\n      * @return FrameOutput\n      */\n     public static FrameOutput withConsumerAlpha(final FrameConsumer consumer) {\n-        return withConsumer(consumer)\n-                .setAlpha(true);\n+        return withConsumer(consumer, ImageFormats.ABGR);\n+    }\n+\n+    protected static FrameOutput withConsumer(final FrameConsumer consumer, final ImageFormat imageFormat) {\n+        return new FrameOutput(\n+                new NutFrameReader(consumer, imageFormat),\n+                "nut", "rawvideo", imageFormat.getPixelFormat(), "pcm_s32be"\n+        );\n+    }\n+\n+    protected interface FrameReader {\n+        void read(InputStream inputStream) throws IOException;\n     }\n \n     @ThreadSafe\n     protected static class FrameOutputNegotiator implements TcpNegotiator {\n-        private final FrameConsumer consumer;\n-        @GuardedBy("this")\n-        private boolean alpha = false;\n-\n-        public FrameOutputNegotiator(FrameConsumer consumer) {\n-            this.consumer = consumer;\n-        }\n+        private final FrameReader frameReader;\n \n-        public synchronized void setAlpha(boolean alpha) {\n-            this.alpha = alpha;\n+        public FrameOutputNegotiator(FrameReader frameReader) {\n+            this.frameReader = frameReader;\n         }\n \n         @Override\n         public synchronized void negotiate(Socket socket) throws IOException {\n             try (InputStream inputStream = socket.getInputStream()) {\n-                NutFrameReader frameReader = new NutFrameReader(consumer, alpha);\n                 frameReader.read(inputStream);\n             }\n         }\ndiff --git a/src/main/java/com/github/kokorin/jaffree/ffmpeg/ImageFormat.java b/src/main/java/com/github/kokorin/jaffree/ffmpeg/ImageFormat.java\nnew file mode 100644\nindex 0000000..9b7affd\n--- /dev/null\n+++ b/src/main/java/com/github/kokorin/jaffree/ffmpeg/ImageFormat.java\n@@ -0,0 +1,66 @@\n+/*\n+ *    Copyright 2021 Denis Kokorin\n+ *\n+ *    Licensed under the Apache License, Version 2.0 (the "License");\n+ *    you may not use this file except in compliance with the License.\n+ *    You may obtain a copy of the License at\n+ *\n+ *        http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ *    Unless required by applicable law or agreed to in writing, software\n+ *    distributed under the License is distributed on an "AS IS" BASIS,\n+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ *    See the License for the specific language governing permissions and\n+ *    limitations under the License.\n+ *\n+ */\n+\n+package com.github.kokorin.jaffree.ffmpeg;\n+\n+import java.awt.image.BufferedImage;\n+\n+/**\n+ * Allows conversion of {@link BufferedImage} to byte array and vice versa.\n+ * <p>\n+ * Intended for adding custom image formats for programmatic video production and consumption.\n+ *\n+ * @see FrameInput\n+ * @see FrameOutput\n+ * @see ImageFormats\n+ */\n+public interface ImageFormat {\n+    /**\n+     * @return pixel format\n+     */\n+    String getPixelFormat();\n+\n+    /**\n+     * @return bytes per pixel\n+     */\n+    int getBytesPerPixel();\n+\n+    /**\n+     * Returns FOURCC (four character code)\n+     * @return FOURCC\n+     * @see <a href="https://www.fourcc.org/fourcc.php">What is a FOURCC?</a>\n+     */\n+    byte[] getFourCC();\n+\n+    /**\n+     * Converts byte array to {@link BufferedImage} of specified size\n+     *\n+     * @param data   raw image data\n+     * @param width  image width\n+     * @param height image height\n+     * @return BufferedImage\n+     */\n+    BufferedImage toImage(byte[] data, int width, int height);\n+\n+    /**\n+     * Converts BufferedImage to raw image data\n+     *\n+     * @param image image\n+     * @return raw data\n+     */\n+    byte[] toBytes(BufferedImage image);\n+}\ndiff --git a/src/main/java/com/github/kokorin/jaffree/ffmpeg/ImageFormats.java b/src/main/java/com/github/kokorin/jaffree/ffmpeg/ImageFormats.java\nnew file mode 100644\nindex 0000000..f63d9ba\n--- /dev/null\n+++ b/src/main/java/com/github/kokorin/jaffree/ffmpeg/ImageFormats.java\n@@ -0,0 +1,115 @@\n+/*\n+ *    Copyright 2021 Denis Kokorin\n+ *\n+ *    Licensed under the Apache License, Version 2.0 (the "License");\n+ *    you may not use this file except in compliance with the License.\n+ *    You may obtain a copy of the License at\n+ *\n+ *        http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ *    Unless required by applicable law or agreed to in writing, software\n+ *    distributed under the License is distributed on an "AS IS" BASIS,\n+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ *    See the License for the specific language governing permissions and\n+ *    limitations under the License.\n+ *\n+ */\n+\n+package com.github.kokorin.jaffree.ffmpeg;\n+\n+import com.github.kokorin.jaffree.JaffreeException;\n+\n+import java.awt.*;\n+import java.awt.color.ColorSpace;\n+import java.awt.image.BufferedImage;\n+import java.awt.image.ComponentColorModel;\n+import java.awt.image.DataBuffer;\n+import java.awt.image.DataBufferByte;\n+import java.awt.image.Raster;\n+import java.awt.image.WritableRaster;\n+\n+public enum ImageFormats implements ImageFormat {\n+    BGR24(\n+            "bgr24",\n+            3,\n+            new byte[]{\'B\', \'G\', \'R\', 24},\n+            BufferedImage.TYPE_3BYTE_BGR,\n+            new ComponentColorModel(\n+                    ColorSpace.getInstance(ColorSpace.CS_sRGB),\n+                    new int[]{8, 8, 8}, false, false,\n+                    Transparency.OPAQUE,\n+                    DataBuffer.TYPE_BYTE\n+            ),\n+            new int[]{2, 1, 0}\n+    ),\n+    ABGR(\n+            "abgr",\n+            4,\n+            new byte[]{\'A\', \'B\', \'G\', \'R\'},\n+            BufferedImage.TYPE_4BYTE_ABGR,\n+            new ComponentColorModel(\n+                    ColorSpace.getInstance(ColorSpace.CS_sRGB),\n+                    new int[]{8, 8, 8, 8}, true, false,\n+                    Transparency.TRANSLUCENT, DataBuffer.TYPE_BYTE\n+            ),\n+            new int[]{3, 2, 1, 0}\n+    );\n+\n+    private final String pixelFormat;\n+    private final int bytesPerPixel;\n+    private final byte[] fourcc;\n+    private final int imageType;\n+    private final ComponentColorModel componentColorModel;\n+    private final int[] bOffs;\n+\n+    ImageFormats(String pixelFormat, int bytesPerPixel, byte[] fourcc, int imageType,\n+                 ComponentColorModel componentColorModel, int[] bOffs) {\n+        this.pixelFormat = pixelFormat;\n+        this.bytesPerPixel = bytesPerPixel;\n+        this.fourcc = fourcc;\n+        this.imageType = imageType;\n+        this.componentColorModel = componentColorModel;\n+        this.bOffs = bOffs;\n+    }\n+\n+    @Override\n+    public String getPixelFormat() {\n+        return pixelFormat;\n+    }\n+\n+    @Override\n+    public int getBytesPerPixel() {\n+        return bytesPerPixel;\n+    }\n+\n+    @Override\n+    public byte[] getFourCC() {\n+        return fourcc.clone();\n+    }\n+\n+    @Override\n+    public BufferedImage toImage(byte[] data, int width, int height) {\n+        int expectedLength = width * height * bytesPerPixel;\n+        if (data.length != expectedLength) {\n+            throw new JaffreeException("Not enough bytes: " + data.length + ", expected " + expectedLength);\n+        }\n+\n+        DataBuffer buffer = new DataBufferByte(data, data.length);\n+        WritableRaster raster = Raster.createInterleavedRaster(\n+                buffer, width, height,\n+                width * bytesPerPixel, bytesPerPixel,\n+                bOffs, null\n+        );\n+\n+        return new BufferedImage(componentColorModel, raster, false, null);\n+    }\n+\n+    @Override\n+    public byte[] toBytes(BufferedImage image) {\n+        if (image.getType() != imageType) {\n+            throw new JaffreeException("Wrong image type: " + image.getType() + ", expected: " + imageType);\n+        }\n+\n+        return ((DataBufferByte) image.getRaster().getDataBuffer()).getData();\n+    }\n+}\ndiff --git a/src/main/java/com/github/kokorin/jaffree/ffmpeg/NutFrameReader.java b/src/main/java/com/github/kokorin/jaffree/ffmpeg/NutFrameReader.java\nindex 6b0c977..0e5e60c 100644\n--- a/src/main/java/com/github/kokorin/jaffree/ffmpeg/NutFrameReader.java\n+++ b/src/main/java/com/github/kokorin/jaffree/ffmpeg/NutFrameReader.java\n@@ -26,15 +26,7 @@ import com.github.kokorin.jaffree.nut.StreamHeader;\n import org.slf4j.Logger;\n import org.slf4j.LoggerFactory;\n \n-import java.awt.*;\n-import java.awt.color.ColorSpace;\n import java.awt.image.BufferedImage;\n-import java.awt.image.ColorModel;\n-import java.awt.image.ComponentColorModel;\n-import java.awt.image.DataBuffer;\n-import java.awt.image.DataBufferByte;\n-import java.awt.image.Raster;\n-import java.awt.image.WritableRaster;\n import java.io.IOException;\n import java.io.InputStream;\n import java.nio.ByteBuffer;\n@@ -46,23 +38,21 @@ import java.util.List;\n  * {@link NutFrameReader} reads InputStream in Nut format and passes parsed frames\n  * to {@link FrameConsumer}.\n  */\n-public class NutFrameReader {\n+public class NutFrameReader implements FrameOutput.FrameReader {\n     private final FrameConsumer frameConsumer;\n-    private final boolean alphaChannel;\n+    private final ImageFormat imageFormat;\n \n-    private static final int RGB_BYTES_PER_PIXEL = 3;\n-    private static final int ALPHA_BYTES_PER_PIXEL = 3;\n     private static final Logger LOGGER = LoggerFactory.getLogger(NutFrameReader.class);\n \n     /**\n      * Creates {@link NutFrameReader}.\n      *\n      * @param frameConsumer frame consumer\n-     * @param alphaChannel  video stream alpha channel\n+     * @param imageFormat image format\n      */\n-    public NutFrameReader(final FrameConsumer frameConsumer, final boolean alphaChannel) {\n+    public NutFrameReader(FrameConsumer frameConsumer, ImageFormat imageFormat) {\n         this.frameConsumer = frameConsumer;\n-        this.alphaChannel = alphaChannel;\n+        this.imageFormat = imageFormat;\n     }\n \n     /**\n@@ -70,6 +60,7 @@ public class NutFrameReader {\n      *\n      * @param input input to read\n      */\n+    @Override\n     public void read(final InputStream input) throws IOException {\n         NutInputStream stream = new NutInputStream(input);\n         NutReader nutReader = new NutReader(stream);\n@@ -133,7 +124,6 @@ public class NutFrameReader {\n         return result;\n     }\n \n-    @SuppressWarnings("checkstyle:magicnumber")\n     private Frame parseFrame(final StreamHeader track, final NutFrame frame) {\n         if (frame == null || frame.data == null || frame.data.length == 0 || frame.eor) {\n             return null;\n@@ -145,45 +135,11 @@ public class NutFrameReader {\n         if (track.streamType == StreamHeader.Type.VIDEO) {\n             int width = track.video.width;\n             int height = track.video.height;\n-\n-            // Sometimes if duration limit is specified, ffmpeg creates\n-            // NutFrame with insufficient data\n-            int pixelCount = width * height;\n-            if (!alphaChannel && pixelCount * RGB_BYTES_PER_PIXEL != frame.data.length\n-                    || alphaChannel && pixelCount * ALPHA_BYTES_PER_PIXEL != frame.data.length) {\n-                return null;\n-            }\n-\n-            DataBuffer buffer = new DataBufferByte(frame.data, frame.data.length);\n-            ColorSpace cs = ColorSpace.getInstance(ColorSpace.CS_sRGB);\n-\n-            final ColorModel colorModel;\n-            final WritableRaster raster;\n-\n-            // TODO extract to BufferedImageUtil\n-            if (!alphaChannel) {\n-                int[] nBits = {8, 8, 8};\n-                int[] bOffs = {2, 1, 0};\n-                colorModel = new ComponentColorModel(cs, nBits, false, false,\n-                        Transparency.OPAQUE,\n-                        DataBuffer.TYPE_BYTE);\n-                raster = Raster.createInterleavedRaster(buffer,\n-                        width, height,\n-                        width * 3, 3,\n-                        bOffs, null);\n-            } else {\n-                int[] nBits = {8, 8, 8, 8};\n-                int[] bOffs = {3, 2, 1, 0};\n-                colorModel = new ComponentColorModel(cs, nBits, true, false,\n-                        Transparency.TRANSLUCENT,\n-                        DataBuffer.TYPE_BYTE);\n-                raster = Raster.createInterleavedRaster(buffer,\n-                        width, height,\n-                        width * 4, 4,\n-                        bOffs, null);\n+            // sometimes ffmpeg can send too short byte array as frame raw data for the last frame\n+            // ignoring such frame, anyway there will be no more frames after it\n+            if (frame.data.length == width * height * imageFormat.getBytesPerPixel()) {\n+                image = imageFormat.toImage(frame.data, width, height);\n             }\n-\n-            image = new BufferedImage(colorModel, raster, false, null);\n         } else if (track.streamType == StreamHeader.Type.AUDIO) {\n             ByteBuffer data = ByteBuffer.wrap(frame.data);\n \ndiff --git a/src/main/java/com/github/kokorin/jaffree/ffmpeg/NutFrameWriter.java b/src/main/java/com/github/kokorin/jaffree/ffmpeg/NutFrameWriter.java\nindex ceb6664..2c1a6c0 100644\n--- a/src/main/java/com/github/kokorin/jaffree/ffmpeg/NutFrameWriter.java\n+++ b/src/main/java/com/github/kokorin/jaffree/ffmpeg/NutFrameWriter.java\n@@ -30,7 +30,6 @@ import org.slf4j.Logger;\n import org.slf4j.LoggerFactory;\n \n import java.awt.image.BufferedImage;\n-import java.awt.image.DataBufferByte;\n import java.io.IOException;\n import java.io.OutputStream;\n import java.nio.ByteBuffer;\n@@ -41,14 +40,11 @@ import java.util.Objects;\n /**\n  * {@link NutFrameWriter} allows writing uncompressed (raw) Frames in Nut format.\n  */\n-// TODO combine with NutWriter\n-public class NutFrameWriter {\n+public class NutFrameWriter implements FrameInput.FrameWriter {\n     private final FrameProducer producer;\n-    private final boolean alpha;\n+    private final ImageFormat imageFormat;\n     private final Long frameOrderingBufferMillis;\n \n-    private static final byte[] FOURCC_ABGR = {\'A\', \'B\', \'G\', \'R\'};\n-    private static final byte[] FOURCC_BGR24 = {\'B\', \'G\', \'R\', 24};\n     //PCM Signed Differential?\n     private static final byte[] FOURCC_PCM_S32BE = {32, \'D\', \'S\', \'P\'};\n \n@@ -58,24 +54,24 @@ public class NutFrameWriter {\n     /**\n      * Creates {@link NutFrameWriter} with default frame reordering buffer length.\n      *\n-     * @param producer frame producer\n-     * @param alpha    video stream alpha channel\n+     * @param producer    frame producer\n+     * @param imageFormat video frames image format\n      */\n-    public NutFrameWriter(final FrameProducer producer, final boolean alpha) {\n-        this(producer, alpha, null);\n+    public NutFrameWriter(final FrameProducer producer, final ImageFormat imageFormat) {\n+        this(producer, imageFormat, null);\n     }\n \n     /**\n      * Creates {@link NutFrameWriter}.\n      *\n      * @param producer                  frame producer\n-     * @param alpha                     video stream alpha channel\n+     * @param imageFormat               video frames image format\n      * @param frameOrderingBufferMillis frame reordering buffer length\n      */\n-    public NutFrameWriter(final FrameProducer producer, final boolean alpha,\n+    public NutFrameWriter(final FrameProducer producer, final ImageFormat imageFormat,\n                           final Long frameOrderingBufferMillis) {\n         this.producer = producer;\n-        this.alpha = alpha;\n+        this.imageFormat = imageFormat;\n         this.frameOrderingBufferMillis = frameOrderingBufferMillis;\n     }\n \n@@ -120,7 +116,7 @@ public class NutFrameWriter {\n                     streamHeader = new StreamHeader(\n                             stream.getId(),\n                             StreamHeader.Type.VIDEO,\n-                            alpha ? FOURCC_ABGR : FOURCC_BGR24,\n+                            imageFormat.getFourCC(),\n                             i,\n                             0,\n                             60_000,\n@@ -197,15 +193,7 @@ public class NutFrameWriter {\n             switch (streamHeader.streamType) {\n                 case VIDEO:\n                     BufferedImage image = frame.getImage();\n-\n-                    if (alpha && BufferedImage.TYPE_4BYTE_ABGR != image.getType()) {\n-                        throw new JaffreeException("Type of BufferedImage must be TYPE_4BYTE_ABGR");\n-                    }\n-                    if (!alpha && BufferedImage.TYPE_3BYTE_BGR != image.getType()) {\n-                        throw new JaffreeException("Type of BufferedImage must be TYPE_3BYTE_BGR");\n-                    }\n-\n-                    data = ((DataBufferByte) image.getRaster().getDataBuffer()).getData();\n+                    data = imageFormat.toBytes(image);\n                     break;\n \n                 case AUDIO:\ndiff --git a/src/main/java/com/github/kokorin/jaffree/ffmpeg/Stream.java b/src/main/java/com/github/kokorin/jaffree/ffmpeg/Stream.java\nindex f8c7630..15221c8 100644\n--- a/src/main/java/com/github/kokorin/jaffree/ffmpeg/Stream.java\n+++ b/src/main/java/com/github/kokorin/jaffree/ffmpeg/Stream.java\n@@ -23,6 +23,7 @@ public class Stream {\n     private Long timebase;\n     private Integer width;\n     private Integer height;\n+    //TODO check if sampleRate can have Integer type\n     private Long sampleRate;\n     private Integer channels;\n \ndiff --git a/src/main/java/com/github/kokorin/jaffree/nut/NutWriter.java b/src/main/java/com/github/kokorin/jaffree/nut/NutWriter.java\nindex 5af4aab..c4d8c0b 100644\n--- a/src/main/java/com/github/kokorin/jaffree/nut/NutWriter.java\n+++ b/src/main/java/com/github/kokorin/jaffree/nut/NutWriter.java\n@@ -281,7 +281,7 @@ public class NutWriter {\n      */\n     public void writeFrame(NutFrame frame) throws IOException {\n         if (closed) {\n-            throw new JaffreeException("NutWriter is ");\n+            throw new JaffreeException("NutWriter is closed");\n         }\n \n         StreamHeader stream = streamHeaders[frame.streamId];\ndiff --git a/src/test/java/com/github/kokorin/jaffree/ffmpeg/FrameIOTest.java b/src/test/java/com/github/kokorin/jaffree/ffmpeg/FrameIOTest.java\nindex 70796db..a266728 100644\n--- a/src/test/java/com/github/kokorin/jaffree/ffmpeg/FrameIOTest.java\n+++ b/src/test/java/com/github/kokorin/jaffree/ffmpeg/FrameIOTest.java\n@@ -14,30 +14,22 @@ import org.junit.rules.ExpectedException;\n import org.slf4j.Logger;\n import org.slf4j.LoggerFactory;\n \n-import javax.imageio.ImageIO;\n-import javax.sound.sampled.AudioFormat;\n-import javax.sound.sampled.AudioSystem;\n-import javax.sound.sampled.LineEvent;\n-import javax.sound.sampled.LineListener;\n-import javax.sound.sampled.LineUnavailableException;\n-import javax.sound.sampled.SourceDataLine;\n import java.awt.*;\n import java.awt.image.BufferedImage;\n-import java.io.ByteArrayInputStream;\n-import java.io.ByteArrayOutputStream;\n-import java.io.IOException;\n import java.nio.file.Files;\n import java.nio.file.Path;\n import java.nio.file.Paths;\n import java.util.Arrays;\n import java.util.Collections;\n-import java.util.Iterator;\n import java.util.List;\n-import java.util.concurrent.CopyOnWriteArrayList;\n-import java.util.concurrent.TimeUnit;\n+import java.util.concurrent.atomic.AtomicInteger;\n import java.util.concurrent.atomic.AtomicLong;\n import java.util.concurrent.atomic.AtomicReference;\n \n+import static org.junit.Assert.assertEquals;\n+import static org.junit.Assert.assertNotNull;\n+import static org.junit.Assert.assertTrue;\n+\n public class FrameIOTest {\n     @Rule\n     public ExpectedException expectedException = ExpectedException.none();\n@@ -56,54 +48,6 @@ public class FrameIOTest {\n         BIN = Paths.get(ffmpegHome);\n     }\n \n-    @Test\n-    public void dumpFrames() throws Exception {\n-        final Path tempDir = Files.createTempDirectory("jaffree");\n-        LOGGER.debug("Will write to " + tempDir);\n-\n-        final AtomicLong trackCounter = new AtomicLong();\n-        final AtomicLong frameCounter = new AtomicLong();\n-        FrameConsumer consumer = new FrameConsumer() {\n-\n-            @Override\n-            public void consumeStreams(List<Stream> tracks) {\n-                trackCounter.set(tracks.size());\n-            }\n-\n-            @Override\n-            public void consume(Frame frame) {\n-                if (frame == null) {\n-                    return;\n-                }\n-\n-                long n = frameCounter.incrementAndGet();\n-                String filename = String.format("frame%05d.png", n);\n-                try {\n-                    boolean written = ImageIO.write(frame.getImage(), "png", tempDir.resolve(filename).toFile());\n-                    Assert.assertTrue(written);\n-                    LOGGER.debug(filename);\n-                } catch (IOException e) {\n-                    throw new RuntimeException(e);\n-                }\n-            }\n-        };\n-\n-        FFmpegResult result = FFmpeg.atPath(BIN)\n-                .addInput(\n-                        UrlInput.fromPath(Artifacts.VIDEO_MP4)\n-                                .setDuration(1, TimeUnit.SECONDS)\n-                )\n-                .addOutput(\n-                        FrameOutput.withConsumer(consumer)\n-                                .disableStream(StreamType.AUDIO)\n-                )\n-                .execute();\n-\n-        Assert.assertNotNull(result);\n-        Assert.assertEquals(1, trackCounter.get());\n-        Assert.assertTrue(frameCounter.get() > 10);\n-    }\n-\n     @Test\n     public void countFrames() throws Exception {\n         final AtomicLong trackCounter = new AtomicLong();\n@@ -140,8 +84,8 @@ public class FrameIOTest {\n                 .execute();\n \n         Assert.assertNotNull(result);\n-        Assert.assertEquals(1, trackCounter.get());\n-        Assert.assertEquals(42L, frameCounter.get());\n+        assertEquals(1, trackCounter.get());\n+        assertEquals(42L, frameCounter.get());\n     }\n \n     @Test\n@@ -270,256 +214,290 @@ public class FrameIOTest {\n                 .setShowStreams(true)\n                 .execute();\n \n-        Assert.assertEquals(4, probe.getStreams().size());\n+        assertEquals(4, probe.getStreams().size());\n     }\n \n     @Test\n-    public void createGif() throws Exception {\n+    public void produceAndConsume() throws Exception {\n         final Path tempDir = Files.createTempDirectory("jaffree");\n-        Path output = tempDir.resolve("test.gif");\n-        LOGGER.debug("Will write to " + tempDir);\n+        Path output = tempDir.resolve("output.mkv");\n \n-        FrameProducer producer = new FrameProducer() {\n-            private final int frameCount = 30;\n-            private long frameCounter = 0;\n+        int width = 720;\n+        int height = 480;\n+        int duration = 20;\n+        int sampleRate = 44100;\n+        int fps = 10;\n+\n+        assertEquals(0, sampleRate % fps);\n+\n+        BufferedImage flag = new BufferedImage(width, height, BufferedImage.TYPE_3BYTE_BGR);\n+        Graphics2D graphics = flag.createGraphics();\n+        graphics.setPaint(Color.WHITE);\n+        graphics.fillRect(0, 0, width, height / 3);\n+        graphics.setPaint(Color.BLUE);\n+        graphics.fillRect(0, height / 3, width, height / 3);\n+        graphics.setPaint(Color.RED);\n+        graphics.fillRect(0, 2 * height / 3, width, height / 3);\n+\n+        int[] samples = new int[sampleRate / fps];\n+\n+        FrameProducer frameProducer = new FrameProducer() {\n+            private int videoPts = 0;\n+            private int audioPts = 0;\n \n             @Override\n             public List<Stream> produceStreams() {\n-                return Collections.singletonList(new Stream()\n-                        .setId(0)\n-                        .setType(Stream.Type.VIDEO)\n-                        .setTimebase(1_000L)\n-                        .setWidth(320)\n-                        .setHeight(240)\n+                return Arrays.asList(\n+                        new Stream()\n+                                .setId(0)\n+                                .setType(Stream.Type.VIDEO)\n+                                .setWidth(width)\n+                                .setHeight(height)\n+                                .setTimebase((long) fps),\n+                        new Stream()\n+                                .setId(1)\n+                                .setType(Stream.Type.AUDIO)\n+                                .setSampleRate(sampleRate)\n+                                .setChannels(1)\n+                                .setTimebase((long) fps)\n                 );\n             }\n \n             @Override\n             public Frame produce() {\n-                if (frameCounter > frameCount) {\n+                if (videoPts > fps * duration) {\n                     return null;\n                 }\n-                LOGGER.debug("Creating frame " + frameCounter);\n \n-                BufferedImage image = new BufferedImage(320, 240, BufferedImage.TYPE_3BYTE_BGR);\n-                Graphics2D graphics = image.createGraphics();\n-                graphics.setPaint(new Color(frameCounter * 1.0f / frameCount, 0, 0));\n-                graphics.fillRect(0, 0, 320, 240);\n-                frameCounter++;\n+                if (videoPts <= audioPts) {\n+                    return new Frame(0, videoPts++, flag);\n+                }\n \n-                return new Frame(0, frameCounter * 1000 / 10, image);\n+                return new Frame(1, audioPts++, samples);\n             }\n         };\n \n-        FFmpegResult result = FFmpeg.atPath(BIN)\n-                .addInput(\n-                        FrameInput.withProducer(producer)\n+        FFmpeg.atPath()\n+                .addInput(FrameInput\n+                        .withProducer(frameProducer)\n+                        .setFrameRate(fps)\n                 )\n-                .addOutput(\n-                        UrlOutput.toPath(output)\n+                .addOutput(UrlOutput\n+                        .toPath(output)\n+                        .setCodec(StreamType.VIDEO, "ffv1")\n                 )\n                 .execute();\n \n-        Assert.assertNotNull(result);\n-    }\n-\n-    @Test\n-    @Ignore("This test plays sound via javax.sound")\n-    public void readAndPlayAudio() throws Exception {\n-        final AtomicLong trackCounter = new AtomicLong();\n-        final AtomicLong frameCounter = new AtomicLong();\n-\n-        FrameConsumer consumer = new FrameConsumer() {\n-            SourceDataLine line = null;\n-            List<Stream> tracks;\n+        AtomicInteger videoCounter = new AtomicInteger();\n+        AtomicInteger sampleCounter = new AtomicInteger();\n \n-            @Override\n-            public void consumeStreams(List<Stream> tracks) {\n-                trackCounter.set(tracks.size());\n-                this.tracks = tracks;\n-\n-                for (Stream track : tracks) {\n-                    if (line == null && track.getType() == Stream.Type.AUDIO) {\n-                        AudioFormat audioFormat = new AudioFormat(track.getSampleRate(), 8, track.getChannels(), true, false);\n-                        try {\n-                            line = AudioSystem.getSourceDataLine(audioFormat);\n-                            line.addLineListener(new LineListener() {\n-                                @Override\n-                                public void update(LineEvent event) {\n-                                    LOGGER.debug(event.toString());\n-                                }\n-                            });\n-                            line.open(audioFormat);\n-                            line.start();\n-                        } catch (LineUnavailableException e) {\n-                            e.printStackTrace();\n-                            throw new RuntimeException(e);\n-                        }\n+        FFmpeg.atPath()\n+                .addInput(UrlInput.fromPath(output))\n+                .addOutput(FrameOutput.withConsumer(new FrameConsumer() {\n+                    @Override\n+                    public void consumeStreams(List<Stream> streams) {\n+                        assertEquals(2, streams.size());\n+\n+                        Stream video = streams.get(0);\n+                        assertEquals(0, video.getId());\n+                        assertEquals(Stream.Type.VIDEO, video.getType());\n+                        assertEquals((Integer) width, video.getWidth());\n+                        assertEquals((Integer) height, video.getHeight());\n+\n+                        Stream audio = streams.get(1);\n+                        assertEquals(1, audio.getId());\n+                        assertEquals(Stream.Type.AUDIO, audio.getType());\n+                        assertEquals(Long.valueOf(sampleRate), audio.getSampleRate());\n+                        assertEquals((Integer) 1, audio.getChannels());\n                     }\n-                }\n-            }\n \n-            @Override\n-            public void consume(Frame frame) {\n-                if (tracks.get(frame.getStreamId()).getType() == Stream.Type.AUDIO) {\n-                    frameCounter.incrementAndGet();\n+                    @Override\n+                    public void consume(Frame frame) {\n+                        if (frame == null) {\n+                            return;\n+                        }\n \n-                    int[] samples = frame.getSamples();\n-                    byte[] bytes = new byte[samples.length];\n+                        if (frame.getStreamId() == 0) {\n+                            BufferedImage image = frame.getImage();\n+                            assertEquals(BufferedImage.TYPE_3BYTE_BGR, image.getType());\n+\n+                            assertEquals(Color.WHITE, new Color(image.getRGB(1, 1)));\n+                            assertEquals(Color.BLUE, new Color(image.getRGB(width / 2, height / 2)));\n+                            assertEquals(Color.RED, new Color(image.getRGB(width - 1, height - 1)));\n+\n+                            // fully compare every 10 frames to speed up test\n+                            if (videoCounter.get() % 10 == 0) {\n+                                for (int x = 0; x < width; x++) {\n+                                    for (int y = 0; y < height; y++) {\n+                                        assertEquals("x: " + x + ", y: " + y,\n+                                                flag.getRGB(x, y), image.getRGB(x, y));\n+                                    }\n+                                }\n+                            }\n \n-                    int coeff = Integer.MAX_VALUE / Byte.MAX_VALUE;\n+                            videoCounter.incrementAndGet();\n+                            return;\n+                        }\n \n-                    for (int i = 0; i < samples.length; i++) {\n-                        bytes[i] = (byte) (samples[i] / coeff);\n+                        int[] samples = frame.getSamples();\n+                        assertNotNull(samples);\n+                        sampleCounter.addAndGet(samples.length);\n                     }\n-                    line.write(bytes, 0, bytes.length);\n-                }\n-            }\n-        };\n-\n-        FFmpegResult result = FFmpeg.atPath(BIN)\n-                .addInput(\n-                        UrlInput.fromPath(Artifacts.VIDEO_MP4)\n-                                .setDuration(3, TimeUnit.SECONDS)\n-                )\n-                .addOutput(\n-                        FrameOutput.withConsumer(consumer)\n-                                .disableStream(StreamType.VIDEO)\n-                )\n+                }))\n                 .execute();\n \n-        Assert.assertNotNull(result);\n-        Assert.assertEquals(2, trackCounter.get());\n-        Assert.assertTrue(frameCounter.get() > 10);\n+        // total number of frames can differ from expected\n+        int expectedFrames = fps * duration;\n+        int actualFrames = videoCounter.get();\n+        assertTrue(expectedFrames == actualFrames || expectedFrames == actualFrames + 1);\n+        // total number of samples can differ in output file, assert the difference is less then 1%\n+        assertEquals(1., 1. * (sampleRate * duration) / sampleCounter.get(), 0.01);\n     }\n \n+\n     @Test\n-    public void createMp3() throws Exception {\n+    public void produceAndConsumeWithAlphaChannel() throws Exception {\n         final Path tempDir = Files.createTempDirectory("jaffree");\n-        Path output = tempDir.resolve("test.mp3");\n-        LOGGER.debug("Will write to " + tempDir);\n+        Path output = tempDir.resolve("output.mkv");\n \n-        FrameProducer producer = new FrameProducer() {\n-            private long frameCounter = 0;\n+        int width = 480;\n+        int height = 480;\n+        int duration = 20;\n+        int sampleRate = 44100;\n+        int fps = 10;\n+\n+        assertEquals(0, sampleRate % fps);\n+\n+        BufferedImage redCross = new BufferedImage(width, height, BufferedImage.TYPE_4BYTE_ABGR);\n+        Graphics2D graphics = redCross.createGraphics();\n+        graphics.setPaint(Color.RED);\n+        graphics.fillRect(width / 3, height / 4, width / 3, height / 2);\n+        graphics.fillRect(width / 4, height / 3, width / 2, height / 3);\n+\n+        int[] samples = new int[sampleRate / fps];\n+\n+        FrameProducer frameProducer = new FrameProducer() {\n+            private int videoPts = 0;\n+            private int audioPts = 0;\n \n             @Override\n             public List<Stream> produceStreams() {\n-                return Collections.singletonList(new Stream()\n-                        .setType(Stream.Type.AUDIO)\n-                        .setTimebase(44100L)\n-                        .setSampleRate(44100)\n-                        .setChannels(1)\n+                return Arrays.asList(\n+                        new Stream()\n+                                .setId(0)\n+                                .setType(Stream.Type.VIDEO)\n+                                .setWidth(width)\n+                                .setHeight(height)\n+                                .setTimebase((long) fps),\n+                        new Stream()\n+                                .setId(1)\n+                                .setType(Stream.Type.AUDIO)\n+                                .setSampleRate(sampleRate)\n+                                .setChannels(1)\n+                                .setTimebase((long) fps)\n                 );\n             }\n \n             @Override\n             public Frame produce() {\n-                if (frameCounter > 30) {\n+                if (videoPts > fps * duration) {\n                     return null;\n                 }\n \n-\n-                long timestamp = frameCounter * 1000 / 10;\n-                int[] samples = new int[4410];\n-                for (int i = 0; i < samples.length; i++) {\n-                    samples[i] = (int) (Integer.MAX_VALUE * Math.sin(300. * (timestamp + i * 100. / samples.length)));\n+                if (videoPts <= audioPts) {\n+                    return new Frame(0, videoPts++, redCross);\n                 }\n-                frameCounter++;\n \n-                return new Frame(0, frameCounter * 4410, samples);\n+                return new Frame(1, audioPts++, samples);\n             }\n         };\n \n-        FFmpegResult result = FFmpeg.atPath(BIN)\n-                .addInput(\n-                        FrameInput.withProducer(producer)\n+        FFmpeg.atPath()\n+                .addInput(FrameInput\n+                        .withProducerAlpha(frameProducer)\n+                        .setFrameRate(fps)\n                 )\n-                .addOutput(\n-                        UrlOutput.toPath(output)\n+                .addOutput(UrlOutput\n+                        .toPath(output)\n+                        .setCodec(StreamType.VIDEO, "ffv1")\n                 )\n                 .execute();\n \n-        Assert.assertNotNull(result);\n-    }\n+        AtomicInteger videoCounter = new AtomicInteger();\n+        AtomicInteger sampleCounter = new AtomicInteger();\n \n-    @Test\n-    public void writeAndRead() throws Exception {\n-        int sampleRate = 44100;\n-        int samplesPerFrame = 4410;\n-        final Stream track = new Stream()\n-                .setId(0)\n-                .setType(Stream.Type.AUDIO)\n-                .setSampleRate(sampleRate)\n-                .setTimebase((long) sampleRate)\n-                .setChannels(1);\n-        final List<Frame> frames = new CopyOnWriteArrayList<>();\n-        for (int i = 0; i < 10; i++) {\n-            Frame frame = new Frame(track.getId(), i * samplesPerFrame, new int[samplesPerFrame]);\n-            frames.add(frame);\n-        }\n+        FFmpeg.atPath()\n+                .addInput(UrlInput.fromPath(output))\n+                .addOutput(FrameOutput.withConsumerAlpha(new FrameConsumer() {\n+                    @Override\n+                    public void consumeStreams(List<Stream> streams) {\n+                        assertEquals(2, streams.size());\n+\n+                        Stream video = streams.get(0);\n+                        assertEquals(0, video.getId());\n+                        assertEquals(Stream.Type.VIDEO, video.getType());\n+                        assertEquals((Integer) width, video.getWidth());\n+                        assertEquals((Integer) height, video.getHeight());\n+\n+                        Stream audio = streams.get(1);\n+                        assertEquals(1, audio.getId());\n+                        assertEquals(Stream.Type.AUDIO, audio.getType());\n+                        assertEquals(Long.valueOf(sampleRate), audio.getSampleRate());\n+                        assertEquals((Integer) 1, audio.getChannels());\n+                    }\n \n-        FrameProducer producer = new FrameProducer() {\n-            Iterator<Frame> frameIterator = frames.iterator();\n+                    @Override\n+                    public void consume(Frame frame) {\n+                        if (frame == null) {\n+                            return;\n+                        }\n \n-            @Override\n-            public List<Stream> produceStreams() {\n-                return Collections.singletonList(track);\n-            }\n+                        if (frame.getStreamId() == 0) {\n+                            BufferedImage image = frame.getImage();\n+                            assertEquals(BufferedImage.TYPE_4BYTE_ABGR, image.getType());\n \n-            @Override\n-            public Frame produce() {\n-                if (frameIterator.hasNext()) {\n-                    return frameIterator.next();\n-                }\n-                return null;\n-            }\n-        };\n+                            assertEquals(0, image.getRGB(0, 0));\n+                            assertEquals(Color.RED.getRGB(), image.getRGB(width / 2, height / 2));\n \n-        ByteArrayOutputStream buffer = new ByteArrayOutputStream();\n+                            // fully compare every 10 frames to speed up test\n+                            if (videoCounter.get() % 10 == 0) {\n+                                for (int x = 0; x < width; x++) {\n+                                    for (int y = 0; y < height; y++) {\n+                                        assertEquals("x: " + x + ", y: " + y,\n+                                                redCross.getRGB(x, y), image.getRGB(x, y));\n+                                    }\n+                                }\n+                            }\n \n-        NutFrameWriter writer = new NutFrameWriter(producer, false, null);\n-        writer.write(buffer);\n+                            videoCounter.incrementAndGet();\n+                            return;\n+                        }\n \n-        final List<Stream> actualTracks = new CopyOnWriteArrayList<>();\n-        final List<Frame> actualFrames = new CopyOnWriteArrayList<>();\n-        FrameConsumer consumer = new FrameConsumer() {\n-            @Override\n-            public void consumeStreams(List<Stream> tracks) {\n-                actualTracks.addAll(tracks);\n-            }\n+                        int[] samples = frame.getSamples();\n+                        assertNotNull(samples);\n+                        sampleCounter.addAndGet(samples.length);\n+                    }\n+                }))\n+                .execute();\n \n-            @Override\n-            public void consume(Frame frame) {\n-                if (frame != null) {\n-                    actualFrames.add(frame);\n-                }\n-            }\n-        };\n \n-        ByteArrayInputStream input = new ByteArrayInputStream(buffer.toByteArray());\n-        NutFrameReader reader = new NutFrameReader(consumer, false);\n-        reader.read(input);\n-\n-        Assert.assertEquals(1, actualTracks.size());\n-        Assert.assertEquals(track.getId(), actualTracks.get(0).getId());\n-        Assert.assertEquals(track.getType(), actualTracks.get(0).getType());\n-        Assert.assertEquals(track.getTimebase(), actualTracks.get(0).getTimebase());\n-        Assert.assertEquals(track.getSampleRate(), actualTracks.get(0).getSampleRate());\n-        Assert.assertEquals(track.getChannels(), actualTracks.get(0).getChannels());\n-\n-        Assert.assertEquals(frames.size(), actualFrames.size());\n-        for (int i = 0; i < frames.size(); i++) {\n-            Frame frame = frames.get(i);\n-            Frame actualFrame = actualFrames.get(i);\n-\n-            Assert.assertEquals(frame.getStreamId(), actualFrame.getStreamId());\n-            Assert.assertArrayEquals(frame.getSamples(), actualFrame.getSamples());\n-            // TODO: compare images\n-        }\n+        // total number of frames can differ from expected\n+        int expectedFrames = fps * duration;\n+        int actualFrames = videoCounter.get();\n+        assertTrue(expectedFrames == actualFrames || expectedFrames == actualFrames + 1);\n+        // total number of samples can differ in output file, assert the difference is less then 1%\n+        assertEquals(1., 1. * (sampleRate * duration) / sampleCounter.get(), 0.01);\n     }\n \n-    public void testNutGenerationAndConsumption(final int duration, final int fps, final long timebase, final int width, final int height) throws Exception {\n-        Assert.assertEquals(0, timebase % fps);\n+    @Test\n+    public void videoFramerateHighResolution() throws Exception {\n+        testNutGenerationAndConsumption(10, 25, 1000, 480, 240);\n+        testNutGenerationAndConsumption(10, 25, 1000, 1280, 720);\n+        testNutGenerationAndConsumption(10, 25, 1000, 2560, 1440);\n+    }\n+\n+    private void testNutGenerationAndConsumption(final int duration, final int fps, final long timebase, final int width, final int height) throws Exception {\n+        assertEquals(0, timebase % fps);\n         Path mp4Path = Files.createTempFile("highResolution", ".mp4");\n \n         final AtomicReference<FFmpegProgress> progressRef = new AtomicReference<>();\n@@ -591,12 +569,4 @@ public class FrameIOTest {\n                         + ", width=" + width + ", height=" + height + ", frames=" + progressRef.get().getFrame(),\n                 expectedFrames == progressRef.get().getFrame() || expectedFrames + 1 == progressRef.get().getFrame());\n     }\n-\n-    @Test\n-    public void videoFramerateHighResolution() throws Exception {\n-        testNutGenerationAndConsumption(10, 25, 1000, 480, 240);\n-        testNutGenerationAndConsumption(10, 25, 1000, 1280, 720);\n-        testNutGenerationAndConsumption(10, 25, 1000, 2560, 1440);\n-    }\n-\n }\ndiff --git a/src/test/java/examples/MosaicExample.java b/src/test/java/examples/MosaicExample.java\nindex f76173f..551de0b 100644\n--- a/src/test/java/examples/MosaicExample.java\n+++ b/src/test/java/examples/MosaicExample.java\n@@ -9,6 +9,7 @@ import com.github.kokorin.jaffree.ffmpeg.FrameConsumer;\n import com.github.kokorin.jaffree.ffmpeg.FrameInput;\n import com.github.kokorin.jaffree.ffmpeg.FrameOutput;\n import com.github.kokorin.jaffree.ffmpeg.FrameProducer;\n+import com.github.kokorin.jaffree.ffmpeg.ImageFormats;\n import com.github.kokorin.jaffree.ffmpeg.Stream;\n import com.github.kokorin.jaffree.ffmpeg.UrlInput;\n import com.github.kokorin.jaffree.ffmpeg.UrlOutput;\n@@ -113,9 +114,8 @@ public class MosaicExample {\n         FFmpegResult mosaicResult = FFmpeg.atPath(ffmpegBin)\n                 .addInput(\n                         FrameInput\n-                                .withProducer(frameProducer)\n+                                .withProducer(frameProducer, ImageFormats.BGR24, 5_000L)\n                                 .setFrameRate(frameRate)\n-                                .setFrameOrderingBuffer(5_000L)\n                 )\n                 .setOverwriteOutput(true)\n                 .setLogLevel(LogLevel.TRACE)\n', 'commit_subject': 'Refactor FrameInput & FrameOutput, improve tests', 'repo_name': 'kokorin/Jaffree', 'committer': {'name': 'kokorin', 'email': '9crqUVAXd6Q17EY354lkbYeB', 'time_sec': '1617864739', 'tz_offset': '180', 'date': {'seconds': '1617864739'}}, 'commit_tree': 'aef33419d111e847735c2657c4586d4b3e1b260b', 'commit_message': 'Refactor FrameInput & FrameOutput, improve tests\n', 'language_name': 'Java', 'license': 'apache-2.0', 'hunks': '@@ -9,6 +9,7 @@ import com.github.kokorin.jaffree.ffmpeg.FrameConsumer;\n import com.github.kokorin.jaffree.ffmpeg.FrameInput;\n import com.github.kokorin.jaffree.ffmpeg.FrameOutput;\n import com.github.kokorin.jaffree.ffmpeg.FrameProducer;\n+import com.github.kokorin.jaffree.ffmpeg.ImageFormats;\n import com.github.kokorin.jaffree.ffmpeg.Stream;\n import com.github.kokorin.jaffree.ffmpeg.UrlInput;\n import com.github.kokorin.jaffree.ffmpeg.UrlOutput;\n@@ -113,9 +114,8 @@ public class MosaicExample {\n         FFmpegResult mosaicResult = FFmpeg.atPath(ffmpegBin)\n                 .addInput(\n                         FrameInput\n-                                .withProducer(frameProducer)\n+                                .withProducer(frameProducer, ImageFormats.BGR24, 5_000L)\n                                 .setFrameRate(frameRate)\n-                                .setFrameOrderingBuffer(5_000L)\n                 )\n                 .setOverwriteOutput(true)\n                 .setLogLevel(LogLevel.TRACE)\n', 'addition_count': 2, 'deletion_count': 2, 'src_file': 'a/src/test/java/examples/MosaicExample.java', 'tgt_file': 'b/src/test/java/examples/MosaicExample.java', 'file_extension': '.java', 'before_file': 'package examples;\n\nimport com.github.kokorin.jaffree.LogLevel;\nimport com.github.kokorin.jaffree.StreamType;\nimport com.github.kokorin.jaffree.ffmpeg.FFmpeg;\nimport com.github.kokorin.jaffree.ffmpeg.FFmpegResult;\nimport com.github.kokorin.jaffree.ffmpeg.Frame;\nimport com.github.kokorin.jaffree.ffmpeg.FrameConsumer;\nimport com.github.kokorin.jaffree.ffmpeg.FrameInput;\nimport com.github.kokorin.jaffree.ffmpeg.FrameOutput;\nimport com.github.kokorin.jaffree.ffmpeg.FrameProducer;\nimport com.github.kokorin.jaffree.ffmpeg.Stream;\nimport com.github.kokorin.jaffree.ffmpeg.UrlInput;\nimport com.github.kokorin.jaffree.ffmpeg.UrlOutput;\nimport com.github.kokorin.jaffree.ffprobe.FFprobe;\nimport com.github.kokorin.jaffree.ffprobe.FFprobeResult;\nimport org.slf4j.Logger;\nimport org.slf4j.LoggerFactory;\n\nimport java.awt.Color;\nimport java.awt.Graphics;\nimport java.awt.Image;\nimport java.awt.image.BufferedImage;\nimport java.awt.image.ImageObserver;\nimport java.nio.file.Path;\nimport java.nio.file.Paths;\nimport java.util.ArrayList;\nimport java.util.Arrays;\nimport java.util.Deque;\nimport java.util.HashMap;\nimport java.util.Iterator;\nimport java.util.LinkedList;\nimport java.util.List;\nimport java.util.Map;\nimport java.util.concurrent.CopyOnWriteArrayList;\n\n/**\n * Note: for some reason ffmpeg 3.3.1 fails when using amix or amerge filters (in this example)\n * Use ffmpeg 4.0 at least, amix works (while amerge still fails)\n */\npublic class MosaicExample {\n    private final Path ffmpegBin;\n    private final List<String> inputs;\n    private final int sampleRate = 44100;\n    private final int frameRate = 25;\n\n    public static final Logger LOGGER = LoggerFactory.getLogger(MosaicExample.class);\n\n    public MosaicExample(String ffmpegBin, List<String> inputs) {\n        this.ffmpegBin = Paths.get(ffmpegBin);\n        this.inputs = inputs;\n    }\n\n    public void execute() {\n        final List<FFmpegResult> results = new CopyOnWriteArrayList<>();\n        List<FrameIterator> frameIterators = new ArrayList<>();\n\n        for (int i = 0; i < inputs.size(); i++) {\n            String input = inputs.get(i);\n\n            boolean hasAudioStream = false;\n            // We must test input for audio stream\n            // Other corner-cases (no video stream, multiple audio stream are not considered)\n            FFprobeResult probeResult = FFprobe.atPath(ffmpegBin)\n                    .setShowStreams(true)\n                    .setInput(Paths.get(input))\n                    .execute();\n            for (com.github.kokorin.jaffree.ffprobe.Stream stream : probeResult.getStreams()) {\n                if (StreamType.AUDIO == stream.getCodecType()) {\n                    hasAudioStream = true;\n                    break;\n                }\n            }\n\n            FrameIterator frameIterator = new FrameIterator();\n            frameIterators.add(frameIterator);\n\n            final FFmpeg ffmpeg = FFmpeg.atPath(ffmpegBin)\n                    .addInput(UrlInput\n                            .fromUrl(input)\n                            .setDuration(15_000))\n                    .addOutput(FrameOutput\n                            .withConsumer(frameIterator.getConsumer())\n                            .setFrameRate(frameRate)\n                            .addArguments("-ac", "1")\n                            .addArguments("-ar", Integer.toString(sampleRate))\n                    )\n                    .setContextName("input" + i);\n\n            if (!hasAudioStream) {\n                ffmpeg\n                        .addInput(\n                                UrlInput.fromUrl("anullsrc")\n                                        .setFormat("lavfi")\n                        )\n                        .addArgument("-shortest");\n            }\n\n            Thread ffmpegThread = new Thread(new Runnable() {\n                @Override\n                public void run() {\n                    FFmpegResult result = ffmpeg.execute();\n                    results.add(result);\n\n                }\n            }, "Reader-" + i + "-main");\n            ffmpegThread.setDaemon(true);\n            ffmpegThread.start();\n        }\n\n        FrameProducer frameProducer = produceMosaic(frameIterators);\n\n        FFmpegResult mosaicResult = FFmpeg.atPath(ffmpegBin)\n                .addInput(\n                        FrameInput\n                                .withProducer(frameProducer)\n                                .setFrameRate(frameRate)\n                                .setFrameOrderingBuffer(5_000L)\n                )\n                .setOverwriteOutput(true)\n                .setLogLevel(LogLevel.TRACE)\n                .addOutput(UrlOutput\n                                .toUrl("mosaic.mp4")\n                        //.addMap(0)\n                )\n                .addArguments("-filter_complex", "amix=inputs=" + inputs.size())\n                .setContextName("result")\n                .execute();\n    }\n\n    private FrameProducer produceMosaic(final List<FrameIterator> frameIterators) {\n        final int rows = (int) Math.round(Math.sqrt(frameIterators.size()));\n        final int columns = (int) Math.ceil(1.0 * frameIterators.size() / rows);\n\n        return new FrameProducer() {\n            private final int elementWidth = 320;\n            private final int elementHeight = 240;\n            private final int mosaicWidth = columns * elementWidth;\n            private final int mosaicHeight = rows * elementHeight;\n            private final long videoFrameDuration = 1000 / frameRate;\n\n            private final Map<Integer, Deque<Frame>> audioQueues = new HashMap<>();\n            private long timecode = 0;\n            private final Frame[] nextVideoFrames = new Frame[frameIterators.size()];\n            private long nextVideoFrameTimecode = 0;\n            private long nextAudioFrameTimecode = 0;\n\n            @Override\n            public List<Stream> produceStreams() {\n                List<Stream> streams = new ArrayList<>();\n                streams.add(\n                        new Stream()\n                                .setId(0)\n                                .setType(Stream.Type.VIDEO)\n                                .setTimebase(1000L)\n                                .setWidth(mosaicWidth)\n                                .setHeight(mosaicHeight)\n                );\n\n                // We copy all audio streams to output and merge tham with amerge filter\n                for (int i = 0; i < frameIterators.size(); i++) {\n                    streams.add(\n                            new Stream()\n                                    .setId(1 + i)\n                                    .setType(Stream.Type.AUDIO)\n                                    .setTimebase((long) sampleRate)\n                                    .setChannels(1)\n                                    .setSampleRate(sampleRate)\n                    );\n                }\n\n                return streams;\n            }\n\n            @Override\n            public Frame produce() {\n                Frame result = null;\n                if (nextVideoFrameTimecode <= timecode) {\n                    if (nextVideoFrameTimecode == 0) {\n                        readNextVideoFrames(nextVideoFrameTimecode);\n                    }\n\n                    result = produceVideoFrame(nextVideoFrames);\n                    // We have to read video frames ahead, otherwise we will have no audio frames read\n                    readNextVideoFrames(nextVideoFrameTimecode);\n                } else if (nextAudioFrameTimecode <= timecode) {\n                    result = produceAudioFrame();\n                }\n\n                timecode = Math.min(nextVideoFrameTimecode, nextAudioFrameTimecode);\n\n                return result;\n            }\n\n            public Frame produceVideoFrame(Frame[] videoFrames) {\n                BufferedImage mosaic = new BufferedImage(mosaicWidth, mosaicHeight, BufferedImage.TYPE_3BYTE_BGR);\n                Graphics mosaicGraphics = mosaic.getGraphics();\n                mosaicGraphics.setColor(new Color(0));\n                mosaicGraphics.fillRect(0, 0, mosaicWidth, mosaicHeight);\n\n                boolean atLeastHasOneElement = false;\n\n                for (int i = 0; i < videoFrames.length; i++) {\n                    Frame videoFrame = videoFrames[i];\n                    if (videoFrame == null) {\n                        continue;\n                    }\n\n                    atLeastHasOneElement = true;\n\n                    BufferedImage element = videoFrame.getImage();\n                    int row = i / columns;\n                    int column = i % columns;\n\n                    ImageObserver observer = new ImageObserver() {\n                        @Override\n                        public boolean imageUpdate(Image img, int infoflags, int x, int y, int width, int height) {\n                            return false;\n                        }\n                    };\n\n                    int dx1 = column * elementWidth;\n                    int dy1 = row * elementHeight;\n                    int dx2 = dx1 + elementWidth;\n                    int dy2 = dy1 + elementHeight;\n\n                    mosaicGraphics.drawImage(element,\n                            dx1, dy1, dx2, dy2,\n                            0, 0, element.getWidth(), element.getHeight(),\n                            observer\n                    );\n                }\n\n                if (!atLeastHasOneElement) {\n                    return null;\n                }\n\n                Frame result = Frame.createVideoFrame(0, nextVideoFrameTimecode, mosaic);\n\n                nextVideoFrameTimecode += videoFrameDuration;\n\n                return result;\n            }\n\n            private Frame produceAudioFrame() {\n                long minPts = Long.MAX_VALUE;\n                long nextPts = Long.MAX_VALUE;\n                int minI = -1;\n                for (int i = 0; i < frameIterators.size(); i++) {\n                    Deque<Frame> aQueue = audioQueues.get(i);\n                    Frame frame = aQueue.peekFirst();\n                    if (frame == null) {\n                        continue;\n                    }\n\n                    long queuePts = frame.getPts();\n                    if (queuePts < minPts) {\n                        nextPts = minPts;\n                        minPts = queuePts;\n                        minI = i;\n                        continue;\n                    }\n                    if (queuePts < nextPts) {\n                        nextPts = queuePts;\n                    }\n                }\n\n                if (minI == -1) {\n                    return null;\n                }\n\n                Frame aFrame = audioQueues.get(minI).pollFirst();\n                if (aFrame == null) {\n                    return null;\n                }\n                aFrame = Frame.createAudioFrame(1 + minI, aFrame.getPts(), aFrame.getSamples());\n\n                if (nextPts != Long.MAX_VALUE) {\n                    nextAudioFrameTimecode = 1000L * nextPts / sampleRate;\n                } else {\n                    nextAudioFrameTimecode = nextVideoFrameTimecode;\n                }\n\n                return aFrame;\n            }\n\n            // All video input streams have forced frameRate, so we know that in every stream\n            // frames will occur with the same timestamp\n            private void readNextVideoFrames(long videoTs) {\n                for (int i = 0; i < frameIterators.size(); i++) {\n                    FrameIterator iter = frameIterators.get(i);\n\n                    if (!iter.hasNext) {\n                        nextVideoFrames[i] = null;\n                    }\n\n                    while (iter.hasNext) {\n                        Frame frame = iter.next();\n                        if (frame == null) {\n                            break;\n                        }\n\n                        Stream stream = iter.getStream(frame.getStreamId());\n                        if (stream == null) {\n                            break;\n                        }\n\n                        switch (stream.getType()) {\n                            case VIDEO:\n                                nextVideoFrames[i] = frame;\n                                break;\n                            case AUDIO:\n                                Deque<Frame> aQueue = audioQueues.get(i);\n                                if (aQueue == null) {\n                                    aQueue = new LinkedList<>();\n                                    audioQueues.put(i, aQueue);\n                                }\n                                aQueue.addLast(frame);\n                                break;\n                        }\n                        long frameTs = 1000L * frame.getPts() / stream.getTimebase();\n                        if (frameTs >= videoTs) {\n                            break;\n                        }\n                    }\n                }\n            }\n        };\n    }\n\n    public static void main(String[] args) {\n        Iterator<String> argIter = Arrays.asList(args).iterator();\n\n        String ffmpegBin = null;\n        List<String> inputs = new ArrayList<>();\n\n        while (argIter.hasNext()) {\n            String arg = argIter.next();\n\n            if ("-ffmpeg_bin".equals(arg)) {\n                ffmpegBin = argIter.next();\n            } else {\n                inputs.add(arg);\n            }\n        }\n\n        if (ffmpegBin == null || inputs.isEmpty()) {\n            System.err.println("Arguments: -ffmpeg_bin </path/to/ffmpeg/bin>");\n            return;\n        }\n\n        new MosaicExample(ffmpegBin, inputs).execute();\n    }\n\n    public static class FrameIterator implements Iterator<Frame> {\n        private volatile boolean hasNext = true;\n        private volatile Frame next = null;\n        private volatile List<Stream> tracks;\n\n        private final FrameConsumer consumer = new FrameConsumer() {\n            @Override\n            public void consumeStreams(List<Stream> tracks) {\n                FrameIterator.this.tracks = tracks;\n            }\n\n            @Override\n            public void consume(Frame frame) {\n                while (next != null) {\n                    try {\n                        Thread.sleep(10);\n                    } catch (InterruptedException e) {\n                        LOGGER.warn("Exception while supplying frame", e);\n                    }\n                }\n\n                hasNext = frame != null;\n                next = frame;\n            }\n        };\n\n        @Override\n        public boolean hasNext() {\n            waitForNextFrame();\n            return hasNext;\n        }\n\n        @Override\n        public Frame next() {\n            waitForNextFrame();\n            Frame result = next;\n            next = null;\n            return result;\n        }\n\n        @Override\n        public void remove() {\n            throw new UnsupportedOperationException("remove");\n        }\n\n        public FrameConsumer getConsumer() {\n            return consumer;\n        }\n\n        public List<Stream> getTracks() {\n            return tracks;\n        }\n\n        public Stream getStream(int id) {\n            for (Stream stream : tracks) {\n                if (stream.getId() == id) {\n                    return stream;\n                }\n            }\n\n            return null;\n        }\n\n        private void waitForNextFrame() {\n            while (hasNext && next == null) {\n                try {\n                    Thread.sleep(10);\n                } catch (InterruptedException e) {\n                    LOGGER.warn("Exception while waiting for frame", e);\n                }\n            }\n        }\n    }\n}\n', 'message': 'Refactor FrameInput & FrameOutput, improve tests\n', 'author_name': 'kokorin', 'lang': 'java'}