import org.freedesktop.gstreamer.Gst;
import org.freedesktop.gstreamer.GstObject;
import org.freedesktop.gstreamer.Element;
import org.freedesktop.gstreamer.Bus;
import org.freedesktop.gstreamer.ClockTime;
import org.freedesktop.gstreamer.Element;
import org.freedesktop.gstreamer.ElementFactory;
import org.freedesktop.gstreamer.Pipeline;
import org.freedesktop.gstreamer.message.Message;
import org.freedesktop.gstreamer.message.MessageType;
import org.freedesktop.gstreamer.event.EOSEvent;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class FakeVm {
	private static final int NUM_PIPELINES = 12;
	private static final List<Pipeline> pipelines = new ArrayList<>();
	private static final List<CountDownLatch> latches = new ArrayList<>();
	private static final String PIPELINE_STR = "rtspsrc name=videosrc latency=200 ! queue ! rtph264depay ! h264parse ! queue ! mpegtsmux ! filesink name=filesink";

	public static void main(String[] args) {
		System.out.println("Test - real pipeline single thread.");
		Gst.init();

		for (int i = 0; i < NUM_PIPELINES; i++) {
			startPipeline(i);
		}

		sleep(3000);

		for (Pipeline pipeline : pipelines) {
			Gst.invokeLater(() -> pipeline.getSources().get(0).sendEvent(new EOSEvent()));
		}

		for (int i = 0; i < NUM_PIPELINES; i++) {
			stopPipeline(i);
		}
	}

	private static void startPipeline(final int i) {
		Pipeline pipeline = (Pipeline) Gst.parseLaunch(PIPELINE_STR);
		pipelines.add(pipeline);
		pipeline.getElementByName("videosrc").set("location", "rtsp://<REMOVED-ANY-H264-SOURCE-WILL-DO>");
		pipeline.getElementByName("filesink").set("location", "/tmp/output" + i + ".mp4");

		CountDownLatch done = new CountDownLatch(1);
		latches.add(done);

		Bus bus = pipeline.getBus();
		bus.connect((Bus.EOS) (source) -> {
			Gst.invokeLater(pipeline::stop);
			done.countDown();
		});

		bus.connect((Bus.ERROR) (source, code, message) -> {
			System.out.println(i + " Error: code: " + code + ", message: " + message);
			Gst.invokeLater(pipeline::stop);
			done.countDown();
		});

		pipeline.play();
	}

	private static void stopPipeline(final int i) {
		try {
			CountDownLatch done = latches.get(i);
			if (done.await(10, TimeUnit.SECONDS)) {
				Gst.invokeLater(() -> pipelines.get(i).close());
			} else {
				System.out.println(i + " Timed out");
				Gst.invokeLater(() -> pipelines.get(i).stop());
				Gst.invokeLater(() -> pipelines.get(i).close());
			}
		} catch (InterruptedException e) {
			System.out.println(i + " Exception");
			Gst.invokeLater(() -> pipelines.get(i).stop());
			Gst.invokeLater(() -> pipelines.get(i).close());
		}
	}

	private static void sleep(long millis) {
		try {
			Thread.sleep(millis);
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
	}
}