forked from j62/ctbrec
Remove mpegts library
This commit is contained in:
0xboobface
parent
e879ab25cf
commit
b8cdb2200e
|
|
@ -1,10 +0,0 @@
|
|||
Changes made to mpegts-streamer for ctbrec
|
||||
=================
|
||||
* Remove dependency on jcodec
|
||||
* Add sleepingEnabled flag in Streamer to disable sleeping completely and make processing of stream much faster
|
||||
* Add sanity check in ContinuityFixer to fix avoid an IndexOutOfBoundsException
|
||||
* Wait for bufferingThread and streamingThread in Streamer.stream() to make it a blocking method
|
||||
* Add BlockingMultiMTSSource, which can be used to add sources, after the streaming has been started
|
||||
* Don't close the stream, if a packet can't be read in one go InputStreamMTSSource. Instead read from
|
||||
the stream until the packet is complete
|
||||
* Remove finalize method. It is deprecated in Java 9.
|
||||
|
|
@ -1,29 +0,0 @@
|
|||
package org.taktik.ioutils;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.ReadableByteChannel;
|
||||
|
||||
public class NIOUtils {
|
||||
|
||||
public static int read(ReadableByteChannel channel, ByteBuffer buffer) throws IOException {
|
||||
int rem = buffer.position();
|
||||
while (channel.read(buffer) != -1 && buffer.hasRemaining()) {
|
||||
}
|
||||
return buffer.position() - rem;
|
||||
}
|
||||
|
||||
public static int skip(ByteBuffer buffer, int count) {
|
||||
int toSkip = Math.min(buffer.remaining(), count);
|
||||
buffer.position(buffer.position() + toSkip);
|
||||
return toSkip;
|
||||
}
|
||||
|
||||
public static final ByteBuffer read(ByteBuffer buffer, int count) {
|
||||
ByteBuffer slice = buffer.duplicate();
|
||||
int limit = buffer.position() + count;
|
||||
slice.limit(limit);
|
||||
buffer.position(limit);
|
||||
return slice;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,6 +0,0 @@
|
|||
package org.taktik.mpegts;
|
||||
|
||||
public class Constants {
|
||||
public static final int MPEGTS_PACKET_SIZE = 188;
|
||||
public static final byte TS_MARKER = 0x47;
|
||||
}
|
||||
|
|
@ -1,608 +0,0 @@
|
|||
package org.taktik.mpegts;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import org.taktik.ioutils.NIOUtils;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
|
||||
public class MTSPacket extends PacketSupport {
|
||||
private boolean transportErrorIndicator; // Transport Error Indicator (TEI)
|
||||
private boolean payloadUnitStartIndicator; // Payload Unit Start Indicator
|
||||
private boolean transportPriority; // Transport Priority
|
||||
private int pid; // Packet Identifier (PID)
|
||||
private int scramblingControl; // Scrambling control
|
||||
private boolean adaptationFieldExist; // Adaptation field exist
|
||||
private boolean containsPayload; // Contains payload
|
||||
private int continuityCounter; // Continuity counter
|
||||
|
||||
private AdaptationField adaptationField;
|
||||
private ByteBuffer payload;
|
||||
|
||||
public static class AdaptationField {
|
||||
private MTSPacket packet;
|
||||
private boolean discontinuityIndicator; // Discontinuity indicator
|
||||
private boolean randomAccessIndicator; // Random Access indicator
|
||||
private boolean elementaryStreamPriorityIndicator; // Elementary stream priority indicator
|
||||
private boolean pcrFlag; // PCR flag
|
||||
private boolean opcrFlag; // OPCR flag
|
||||
private boolean splicingPointFlag; // Splicing point flag
|
||||
private boolean transportPrivateDataFlag; // Transport private data flag
|
||||
private boolean adaptationFieldExtensionFlag; // Adaptation field extension flag
|
||||
private PCR pcr; // PCR
|
||||
private PCR opcr; // OPCR
|
||||
private byte spliceCountdown; // Splice countdown
|
||||
private byte[] privateData; // Private data
|
||||
private byte[] extension; // Extension
|
||||
|
||||
public static class PCR {
|
||||
private AdaptationField field;
|
||||
public long base; // 33 bits
|
||||
public int extension; // 9 bits
|
||||
public byte reserved; // 6 bits
|
||||
|
||||
public PCR(AdaptationField field, long base, int extension, byte reserved) {
|
||||
this.base = base;
|
||||
this.extension = extension;
|
||||
this.reserved = reserved;
|
||||
this.field = field;
|
||||
}
|
||||
|
||||
public long getValue() {
|
||||
return base * 300 + extension;
|
||||
}
|
||||
|
||||
public void setValue(long value) {
|
||||
base = value / 300;
|
||||
extension = (int)value % 300;
|
||||
field.markDirty();
|
||||
}
|
||||
|
||||
public void write(ByteBuffer buffer) {
|
||||
buffer.putInt((int) ((base & 0x1FFFFFFFFL) >> 1));
|
||||
int middleByte = 0;
|
||||
middleByte |= ((base & 0x1) << 7);
|
||||
middleByte |= ((reserved & 0x3F) << 1);
|
||||
middleByte |= ((extension & 0x1FF) >> 8);
|
||||
buffer.put((byte) middleByte);
|
||||
buffer.put((byte) (extension & 0xff));
|
||||
}
|
||||
}
|
||||
|
||||
public AdaptationField(MTSPacket packet, boolean discontinuityIndicator, boolean randomAccessIndicator, boolean elementaryStreamPriorityIndicator, boolean pcrFlag, boolean opcrFlag, boolean splicingPointFlag, boolean transportPrivateDataFlag, boolean adaptationFieldExtensionFlag, PCR pcr, PCR opcr, byte spliceCountdown, byte[] privateData, byte[] extension) {
|
||||
this.packet = packet;
|
||||
this.discontinuityIndicator = discontinuityIndicator;
|
||||
this.randomAccessIndicator = randomAccessIndicator;
|
||||
this.elementaryStreamPriorityIndicator = elementaryStreamPriorityIndicator;
|
||||
this.pcrFlag = pcrFlag;
|
||||
this.opcrFlag = opcrFlag;
|
||||
this.splicingPointFlag = splicingPointFlag;
|
||||
this.transportPrivateDataFlag = transportPrivateDataFlag;
|
||||
this.adaptationFieldExtensionFlag = adaptationFieldExtensionFlag;
|
||||
this.pcr = pcr;
|
||||
if (this.pcr != null) {
|
||||
this.pcr.field = this;
|
||||
}
|
||||
this.opcr = opcr;
|
||||
if (this.opcr != null) {
|
||||
this.opcr.field = this;
|
||||
}
|
||||
this.spliceCountdown = spliceCountdown;
|
||||
this.privateData = privateData;
|
||||
this.extension = extension;
|
||||
}
|
||||
|
||||
public boolean isDiscontinuityIndicator() {
|
||||
return discontinuityIndicator;
|
||||
}
|
||||
|
||||
public void setDiscontinuityIndicator(boolean discontinuityIndicator) {
|
||||
this.discontinuityIndicator = discontinuityIndicator;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
private void markDirty() {
|
||||
packet.markDirty();
|
||||
}
|
||||
|
||||
public boolean isRandomAccessIndicator() {
|
||||
return randomAccessIndicator;
|
||||
}
|
||||
|
||||
public void setRandomAccessIndicator(boolean randomAccessIndicator) {
|
||||
this.randomAccessIndicator = randomAccessIndicator;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public boolean isElementaryStreamPriorityIndicator() {
|
||||
return elementaryStreamPriorityIndicator;
|
||||
}
|
||||
|
||||
public void setElementaryStreamPriorityIndicator(boolean elementaryStreamPriorityIndicator) {
|
||||
this.elementaryStreamPriorityIndicator = elementaryStreamPriorityIndicator;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public boolean isPcrFlag() {
|
||||
return pcrFlag;
|
||||
}
|
||||
|
||||
public void setPcrFlag(boolean pcrFlag) {
|
||||
this.pcrFlag = pcrFlag;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public boolean isOpcrFlag() {
|
||||
return opcrFlag;
|
||||
}
|
||||
|
||||
public void setOpcrFlag(boolean opcrFlag) {
|
||||
this.opcrFlag = opcrFlag;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public boolean isSplicingPointFlag() {
|
||||
return splicingPointFlag;
|
||||
}
|
||||
|
||||
public void setSplicingPointFlag(boolean splicingPointFlag) {
|
||||
this.splicingPointFlag = splicingPointFlag;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public boolean isTransportPrivateDataFlag() {
|
||||
return transportPrivateDataFlag;
|
||||
}
|
||||
|
||||
public void setTransportPrivateDataFlag(boolean transportPrivateDataFlag) {
|
||||
this.transportPrivateDataFlag = transportPrivateDataFlag;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public boolean isAdaptationFieldExtensionFlag() {
|
||||
return adaptationFieldExtensionFlag;
|
||||
}
|
||||
|
||||
public void setAdaptationFieldExtensionFlag(boolean adaptationFieldExtensionFlag) {
|
||||
this.adaptationFieldExtensionFlag = adaptationFieldExtensionFlag;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public PCR getPcr() {
|
||||
return pcr;
|
||||
}
|
||||
|
||||
public void setPcr(PCR pcr) {
|
||||
this.pcr = pcr;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public PCR getOpcr() {
|
||||
return opcr;
|
||||
}
|
||||
|
||||
public void setOpcr(PCR opcr) {
|
||||
this.opcr = opcr;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public byte getSpliceCountdown() {
|
||||
return spliceCountdown;
|
||||
}
|
||||
|
||||
public void setSpliceCountdown(byte spliceCountdown) {
|
||||
this.spliceCountdown = spliceCountdown;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public byte[] getPrivateData() {
|
||||
return privateData;
|
||||
}
|
||||
|
||||
public void setPrivateData(byte[] privateData) {
|
||||
this.privateData = privateData;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public byte[] getExtension() {
|
||||
return extension;
|
||||
}
|
||||
|
||||
public void setExtension(byte[] extension) {
|
||||
this.extension = extension;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public void write(ByteBuffer buffer, int payloadLength) {
|
||||
int length = 183 - payloadLength;
|
||||
int remaining = length;
|
||||
buffer.put((byte) (length & 0xff));
|
||||
int firstByte = 0;
|
||||
|
||||
// Discontinuity indicator
|
||||
if (discontinuityIndicator) {
|
||||
firstByte |= 0x80;
|
||||
}
|
||||
|
||||
// Random Access indicator
|
||||
if (randomAccessIndicator) {
|
||||
firstByte |= 0x40;
|
||||
}
|
||||
|
||||
// Elementary stream priority indicator
|
||||
if (elementaryStreamPriorityIndicator) {
|
||||
firstByte |= 0x20;
|
||||
}
|
||||
|
||||
// PCR flag
|
||||
if (pcrFlag) {
|
||||
firstByte |= 0x10;
|
||||
}
|
||||
|
||||
// OPCR flag
|
||||
if (opcrFlag) {
|
||||
firstByte |= 0x08;
|
||||
}
|
||||
|
||||
// Splicing point flag
|
||||
if (splicingPointFlag) {
|
||||
firstByte |= 0x04;
|
||||
}
|
||||
|
||||
// Transport private data flag
|
||||
if (transportPrivateDataFlag) {
|
||||
firstByte |= 0x2;
|
||||
}
|
||||
|
||||
// Adaptation field extension flag
|
||||
if (adaptationFieldExtensionFlag) {
|
||||
firstByte |= 0x01;
|
||||
}
|
||||
|
||||
buffer.put((byte) firstByte);
|
||||
remaining --;
|
||||
|
||||
if (pcrFlag && pcr != null) {
|
||||
pcr.write(buffer);
|
||||
remaining -= 6;
|
||||
}
|
||||
|
||||
if (opcrFlag && opcr != null) {
|
||||
opcr.write(buffer);
|
||||
remaining -= 6;
|
||||
}
|
||||
|
||||
if (splicingPointFlag) {
|
||||
buffer.put(spliceCountdown);
|
||||
remaining--;
|
||||
}
|
||||
|
||||
if (transportPrivateDataFlag && privateData != null) {
|
||||
buffer.put(privateData);
|
||||
remaining -= privateData.length;
|
||||
}
|
||||
|
||||
if (adaptationFieldExtensionFlag && extension != null) {
|
||||
buffer.put(extension);
|
||||
remaining -= extension.length;
|
||||
}
|
||||
|
||||
if (remaining < 0) {
|
||||
throw new IllegalStateException("Adaptation field too big!");
|
||||
}
|
||||
while (remaining-- > 0) {
|
||||
buffer.put((byte)0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public MTSPacket(boolean transportErrorIndicator, boolean payloadUnitStartIndicator, boolean transportPriority, int pid, int scramblingControl, int continuityCounter) {
|
||||
super();
|
||||
this.buffer = ByteBuffer.allocate(Constants.MPEGTS_PACKET_SIZE);
|
||||
this.transportErrorIndicator = transportErrorIndicator;
|
||||
this.payloadUnitStartIndicator = payloadUnitStartIndicator;
|
||||
this.transportPriority = transportPriority;
|
||||
this.pid = pid;
|
||||
this.scramblingControl = scramblingControl;
|
||||
this.continuityCounter = continuityCounter;
|
||||
this.adaptationFieldExist = false;
|
||||
this.containsPayload = false;
|
||||
}
|
||||
|
||||
public MTSPacket(ByteBuffer buffer) {
|
||||
super(buffer);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void write() {
|
||||
// First write payload
|
||||
int payloadLength = 0;
|
||||
if (containsPayload && payload != null) {
|
||||
payload.clear();
|
||||
payloadLength = payload.capacity();
|
||||
buffer.position(Constants.MPEGTS_PACKET_SIZE - payloadLength);
|
||||
buffer.put(payload);
|
||||
}
|
||||
buffer.rewind();
|
||||
// First byte
|
||||
buffer.put((byte) 0x47);
|
||||
|
||||
// Bytes 2->3
|
||||
int secondAndThirdBytes = 0;
|
||||
if (transportErrorIndicator) {
|
||||
secondAndThirdBytes |= 0x8000;
|
||||
}
|
||||
if (payloadUnitStartIndicator) {
|
||||
secondAndThirdBytes |= 0x4000;
|
||||
}
|
||||
if (transportPriority) {
|
||||
secondAndThirdBytes |= 0x2000;
|
||||
}
|
||||
secondAndThirdBytes |= (pid & 0x1fff);
|
||||
|
||||
buffer.putShort((short) secondAndThirdBytes);
|
||||
|
||||
int fourthByte = 0;
|
||||
|
||||
// Byte 4
|
||||
fourthByte |= (scramblingControl & 0xc0);
|
||||
if (adaptationFieldExist) {
|
||||
fourthByte |= 0x20;
|
||||
}
|
||||
if (containsPayload) {
|
||||
fourthByte |= 0x10;
|
||||
}
|
||||
fourthByte |= (continuityCounter & 0x0f);
|
||||
buffer.put((byte) fourthByte);
|
||||
|
||||
// Adaptation field
|
||||
if (adaptationFieldExist) {
|
||||
if (adaptationField == null) {
|
||||
buffer.put((byte) 0);
|
||||
} else {
|
||||
adaptationField.write(buffer, payloadLength);
|
||||
}
|
||||
}
|
||||
|
||||
// Payload
|
||||
if (containsPayload && payload != null) {
|
||||
payload.rewind();
|
||||
buffer.put(payload);
|
||||
}
|
||||
if (buffer.remaining() != 0) {
|
||||
throw new IllegalStateException("buffer.remaining() = " + buffer.remaining() + ", should be zero!");
|
||||
}
|
||||
}
|
||||
|
||||
protected void parse() {
|
||||
// Sync byte
|
||||
int marker = buffer.get() & 0xff;
|
||||
Preconditions.checkArgument(Constants.TS_MARKER == marker);
|
||||
|
||||
// Second/Third byte
|
||||
int secondAndThirdBytes = buffer.getShort();
|
||||
|
||||
// Transport Error Indicator (TEI)
|
||||
boolean transportErrorIndicator = (secondAndThirdBytes & 0x8000) != 0;
|
||||
|
||||
// Payload Unit Start Indicator
|
||||
boolean payloadUnitStartIndicator = (secondAndThirdBytes & 0x4000) != 0;
|
||||
|
||||
// Transport Priority
|
||||
boolean transportPriority = (secondAndThirdBytes & 0x2000) != 0;
|
||||
|
||||
// Packet Identifier (PID)
|
||||
int pid = secondAndThirdBytes & 0x1fff;
|
||||
|
||||
// Fourth byte
|
||||
int fourthByte = buffer.get() & 0xff;
|
||||
|
||||
// Scrambling control
|
||||
int scramblingControl = fourthByte & 0xc0;
|
||||
|
||||
// Adaptation field exist
|
||||
boolean adaptationFieldExist = (fourthByte & 0x20) != 0;
|
||||
|
||||
// Contains payload
|
||||
boolean containsPayload = (fourthByte & 0x10) != 0;
|
||||
|
||||
// Continuity counter
|
||||
int continuityCounter = fourthByte & 0x0f;
|
||||
|
||||
MTSPacket.AdaptationField adaptationField = null;
|
||||
if (adaptationFieldExist) {
|
||||
// Adaptation Field Length
|
||||
int adaptationFieldLength = buffer.get() & 0xff;
|
||||
if (adaptationFieldLength != 0) {
|
||||
|
||||
int remainingBytes = adaptationFieldLength;
|
||||
|
||||
// Get next byte
|
||||
int nextByte = buffer.get() & 0xff;
|
||||
remainingBytes--;
|
||||
|
||||
// Discontinuity indicator
|
||||
boolean discontinuityIndicator = (nextByte & 0x80) != 0;
|
||||
|
||||
// Random Access indicator
|
||||
boolean randomAccessIndicator = (nextByte & 0x40) != 0;
|
||||
|
||||
// Elementary stream priority indicator
|
||||
boolean elementaryStreamPriorityIndicator = (nextByte & 0x20) != 0;
|
||||
|
||||
// PCR flag
|
||||
boolean pcrFlag = (nextByte & 0x10) != 0;
|
||||
|
||||
// OPCR flag
|
||||
boolean opcrFlag = (nextByte & 0x08) != 0;
|
||||
|
||||
// Splicing point flag
|
||||
boolean splicingPointFlag = (nextByte & 0x04) != 0;
|
||||
|
||||
// Transport private data flag
|
||||
boolean transportPrivateDataFlag = (nextByte & 0x2) != 0;
|
||||
|
||||
// Adaptation field extension flag
|
||||
boolean adaptationFieldExtensionFlag = (nextByte & 0x01) != 0;
|
||||
|
||||
// PCR
|
||||
MTSPacket.AdaptationField.PCR pcr = null;
|
||||
if (pcrFlag) {
|
||||
pcr = parsePCR();
|
||||
remainingBytes -= 6;
|
||||
}
|
||||
|
||||
// OPCR
|
||||
MTSPacket.AdaptationField.PCR opcr = null;
|
||||
if (opcrFlag) {
|
||||
opcr = parsePCR();
|
||||
remainingBytes -= 6;
|
||||
}
|
||||
|
||||
// Splice countdown
|
||||
byte spliceCountdown = 0;
|
||||
if (splicingPointFlag) {
|
||||
spliceCountdown = buffer.get();
|
||||
remainingBytes--;
|
||||
}
|
||||
|
||||
byte[] privateData = null;
|
||||
if (transportPrivateDataFlag) {
|
||||
int transportPrivateDataLength = buffer.get() & 0xff;
|
||||
privateData = new byte[transportPrivateDataLength];
|
||||
buffer.get(privateData);
|
||||
remainingBytes -= transportPrivateDataLength;
|
||||
}
|
||||
|
||||
byte[] extension = null;
|
||||
if (adaptationFieldExtensionFlag) {
|
||||
int extensionLength = buffer.get() & 0xff;
|
||||
extension = new byte[extensionLength];
|
||||
buffer.get(extension);
|
||||
remainingBytes -= extensionLength;
|
||||
}
|
||||
|
||||
// Skip remaining bytes
|
||||
NIOUtils.skip(buffer, remainingBytes);
|
||||
|
||||
adaptationField = new MTSPacket.AdaptationField(this, discontinuityIndicator, randomAccessIndicator, elementaryStreamPriorityIndicator, pcrFlag, opcrFlag, splicingPointFlag, transportPrivateDataFlag, adaptationFieldExtensionFlag, pcr, opcr, spliceCountdown, privateData, extension);
|
||||
}
|
||||
}
|
||||
|
||||
this.transportErrorIndicator = transportErrorIndicator;
|
||||
this.payloadUnitStartIndicator = payloadUnitStartIndicator;
|
||||
this.transportPriority = transportPriority;
|
||||
this.pid = pid;
|
||||
this.scramblingControl = scramblingControl;
|
||||
this.adaptationFieldExist = adaptationFieldExist;
|
||||
this.containsPayload = containsPayload;
|
||||
this.continuityCounter = continuityCounter;
|
||||
this.adaptationField = adaptationField;
|
||||
|
||||
// Payload
|
||||
this.payload = containsPayload ? buffer.slice() : null;
|
||||
}
|
||||
|
||||
private AdaptationField.PCR parsePCR() {
|
||||
AdaptationField.PCR pcr;
|
||||
byte[] pcrBytes = new byte[6];
|
||||
buffer.get(pcrBytes);
|
||||
|
||||
long pcrBits = ((pcrBytes[0] & 0xffL) << 40) | ((pcrBytes[1] & 0xffL) << 32) | ((pcrBytes[2] & 0xffL) << 24) | ((pcrBytes[3] & 0xffL) << 16) | ((pcrBytes[4] & 0xffL) << 8) | (pcrBytes[5] & 0xffL);
|
||||
long base = (pcrBits & 0xFFFFFFFF8000L) >> 15;
|
||||
byte reserved = (byte) ((pcrBits & 0x7E00) >> 9);
|
||||
int extension = (int) (pcrBits & 0x1FFL);
|
||||
pcr = new AdaptationField.PCR(null, base, extension, reserved);
|
||||
return pcr;
|
||||
}
|
||||
|
||||
public boolean isTransportErrorIndicator() {
|
||||
return transportErrorIndicator;
|
||||
}
|
||||
|
||||
public void setTransportErrorIndicator(boolean transportErrorIndicator) {
|
||||
this.transportErrorIndicator = transportErrorIndicator;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public boolean isPayloadUnitStartIndicator() {
|
||||
return payloadUnitStartIndicator;
|
||||
}
|
||||
|
||||
public void setPayloadUnitStartIndicator(boolean payloadUnitStartIndicator) {
|
||||
this.payloadUnitStartIndicator = payloadUnitStartIndicator;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public boolean isTransportPriority() {
|
||||
return transportPriority;
|
||||
}
|
||||
|
||||
public void setTransportPriority(boolean transportPriority) {
|
||||
this.transportPriority = transportPriority;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public int getPid() {
|
||||
return pid;
|
||||
}
|
||||
|
||||
public void setPid(int pid) {
|
||||
this.pid = pid;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public int getScramblingControl() {
|
||||
return scramblingControl;
|
||||
}
|
||||
|
||||
public void setScramblingControl(int scramblingControl) {
|
||||
this.scramblingControl = scramblingControl;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public boolean isAdaptationFieldExist() {
|
||||
return adaptationFieldExist;
|
||||
}
|
||||
|
||||
public void setAdaptationFieldExist(boolean adaptationFieldExist) {
|
||||
this.adaptationFieldExist = adaptationFieldExist;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public boolean isContainsPayload() {
|
||||
return containsPayload;
|
||||
}
|
||||
|
||||
public void setContainsPayload(boolean containsPayload) {
|
||||
this.containsPayload = containsPayload;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public int getContinuityCounter() {
|
||||
return continuityCounter;
|
||||
}
|
||||
|
||||
public void setContinuityCounter(int continuityCounter) {
|
||||
this.continuityCounter = continuityCounter;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public AdaptationField getAdaptationField() {
|
||||
return adaptationField;
|
||||
}
|
||||
|
||||
public void setAdaptationField(AdaptationField adaptationField) {
|
||||
this.adaptationField = adaptationField;
|
||||
markDirty();
|
||||
}
|
||||
|
||||
public ByteBuffer getPayload() {
|
||||
return payload;
|
||||
}
|
||||
|
||||
public void setPayload(ByteBuffer payload) {
|
||||
this.payload = payload;
|
||||
markDirty();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,49 +0,0 @@
|
|||
package org.taktik.mpegts;
|
||||
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import com.google.common.collect.Lists;
|
||||
|
||||
public class PATSection extends PSISection {
|
||||
private Integer[] networkPids;
|
||||
private Map<Integer, Integer> programs;
|
||||
|
||||
public PATSection(PSISection psi, Integer[] networkPids, Map<Integer, Integer> programs) {
|
||||
super(psi);
|
||||
this.networkPids = networkPids;
|
||||
this.programs = programs;
|
||||
}
|
||||
|
||||
public Integer[] getNetworkPids() {
|
||||
return networkPids;
|
||||
}
|
||||
|
||||
public Map<Integer, Integer> getPrograms() {
|
||||
return programs;
|
||||
}
|
||||
|
||||
public static PATSection parse(ByteBuffer data) {
|
||||
PSISection psi = PSISection.parse(data);
|
||||
if (psi == null) {
|
||||
return null;
|
||||
}
|
||||
List<Integer> networkPids = Lists.newArrayList();
|
||||
Map<Integer, Integer> programs = new HashMap<>();
|
||||
|
||||
while (data.remaining() > 4) {
|
||||
int programNum = data.getShort() & 0xffff;
|
||||
int w = data.getShort();
|
||||
int pid = w & 0x1fff;
|
||||
if (programNum == 0)
|
||||
networkPids.add(pid);
|
||||
else
|
||||
programs.put(programNum, pid);
|
||||
}
|
||||
|
||||
return new PATSection(psi, networkPids.toArray(new Integer[networkPids.size()]), programs);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,133 +0,0 @@
|
|||
package org.taktik.mpegts;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import org.taktik.ioutils.NIOUtils;
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* This class is part of JCodec ( www.jcodec.org ) This software is distributed
|
||||
* under FreeBSD License
|
||||
*
|
||||
* Represents PMT ( Program Map Table ) of the MPEG Transport stream
|
||||
*
|
||||
* This section contains information about streams of an individual program, a
|
||||
* program usually contains two or more streams, such as video, audio, text,
|
||||
* etc..
|
||||
*
|
||||
* @author The JCodec project
|
||||
*
|
||||
*/
|
||||
public class PMTSection extends PSISection {
|
||||
|
||||
private int pcrPid;
|
||||
// private Tag[] tags;
|
||||
// private PMTStream[] streams;
|
||||
|
||||
public PMTSection(PSISection psi, int pcrPid) {//, Tag[] tags, PMTStream[] streams) {
|
||||
super(psi);
|
||||
this.pcrPid = pcrPid;
|
||||
// this.tags = tags;
|
||||
// this.streams = streams;
|
||||
}
|
||||
|
||||
public int getPcrPid() {
|
||||
return pcrPid;
|
||||
}
|
||||
|
||||
// public Tag[] getTags() {
|
||||
// return tags;
|
||||
// }
|
||||
//
|
||||
// public PMTStream[] getStreams() {
|
||||
// return streams;
|
||||
// }
|
||||
|
||||
public static PMTSection parse(ByteBuffer data) {
|
||||
PSISection psi = PSISection.parse(data);
|
||||
|
||||
int w1 = data.getShort() & 0xffff;
|
||||
int pcrPid = w1 & 0x1fff;
|
||||
|
||||
//int w2 = data.getShort() & 0xffff;
|
||||
//int programInfoLength = w2 & 0xfff;
|
||||
|
||||
// List<Tag> tags = parseTags(NIOUtils.read(data, programInfoLength));
|
||||
// List<PMTStream> streams = new ArrayList<PMTStream>();
|
||||
// while (data.remaining() > 4) {
|
||||
// int streamType = data.get() & 0xff;
|
||||
// int wn = data.getShort() & 0xffff;
|
||||
// int elementaryPid = wn & 0x1fff;
|
||||
//
|
||||
//
|
||||
// int wn1 = data.getShort() & 0xffff;
|
||||
// int esInfoLength = wn1 & 0xfff;
|
||||
// ByteBuffer read = NIOUtils.read(data, esInfoLength);
|
||||
// streams.add(new PMTStream(streamType, elementaryPid, MPSUtils.parseDescriptors(read)));
|
||||
// }
|
||||
|
||||
return new PMTSection(psi, pcrPid);
|
||||
}
|
||||
|
||||
static List<Tag> parseTags(ByteBuffer bb) {
|
||||
List<Tag> tags = new ArrayList<Tag>();
|
||||
while (bb.hasRemaining()) {
|
||||
int tag = bb.get();
|
||||
int tagLen = bb.get();
|
||||
tags.add(new Tag(tag, NIOUtils.read(bb, tagLen)));
|
||||
}
|
||||
return tags;
|
||||
}
|
||||
|
||||
|
||||
public static class Tag {
|
||||
private int tag;
|
||||
private ByteBuffer content;
|
||||
|
||||
public Tag(int tag, ByteBuffer content) {
|
||||
this.tag = tag;
|
||||
this.content = content;
|
||||
}
|
||||
|
||||
public int getTag() {
|
||||
return tag;
|
||||
}
|
||||
|
||||
public ByteBuffer getContent() {
|
||||
return content;
|
||||
}
|
||||
}
|
||||
|
||||
// public static class PMTStream {
|
||||
// private int streamTypeTag;
|
||||
// private int pid;
|
||||
// private List<MPEGMediaDescriptor> descriptors;
|
||||
// private StreamType streamType;
|
||||
//
|
||||
// public PMTStream(int streamTypeTag, int pid, List<MPEGMediaDescriptor> descriptors) {
|
||||
// this.streamTypeTag = streamTypeTag;
|
||||
// this.pid = pid;
|
||||
// this.descriptors = descriptors;
|
||||
// this.streamType = StreamType.fromTag(streamTypeTag);
|
||||
// }
|
||||
//
|
||||
// public int getStreamTypeTag() {
|
||||
// return streamTypeTag;
|
||||
// }
|
||||
//
|
||||
// public StreamType getStreamType() {
|
||||
// return streamType;
|
||||
// }
|
||||
//
|
||||
// public int getPid() {
|
||||
// return pid;
|
||||
// }
|
||||
//
|
||||
// public List<MPEGMediaDescriptor> getDesctiptors() {
|
||||
// return descriptors;
|
||||
// }
|
||||
// }
|
||||
}
|
||||
|
|
@ -1,90 +0,0 @@
|
|||
package org.taktik.mpegts;
|
||||
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
/**
|
||||
* This class is part of JCodec ( www.jcodec.org ) This software is distributed
|
||||
* under FreeBSD License
|
||||
*
|
||||
* Represents a section of PSI payload ( Program Stream Information ) MPEG
|
||||
* Transport stream
|
||||
*
|
||||
* @author The JCodec project
|
||||
*
|
||||
*/
|
||||
public class PSISection {
|
||||
private int tableId;
|
||||
private int specificId;
|
||||
private int versionNumber;
|
||||
private int currentNextIndicator;
|
||||
private int sectionNumber;
|
||||
private int lastSectionNumber;
|
||||
|
||||
/**
|
||||
* A copy constructor
|
||||
*
|
||||
* @param other
|
||||
*/
|
||||
public PSISection(PSISection other) {
|
||||
this(other.tableId, other.specificId, other.versionNumber, other.currentNextIndicator, other.sectionNumber,
|
||||
other.lastSectionNumber);
|
||||
}
|
||||
|
||||
public PSISection(int tableId, int specificId, int versionNumber, int currentNextIndicator, int sectionNumber,
|
||||
int lastSectionNumber) {
|
||||
this.tableId = tableId;
|
||||
this.specificId = specificId;
|
||||
this.versionNumber = versionNumber;
|
||||
this.currentNextIndicator = currentNextIndicator;
|
||||
this.sectionNumber = sectionNumber;
|
||||
this.lastSectionNumber = lastSectionNumber;
|
||||
}
|
||||
|
||||
public static PSISection parse(ByteBuffer data) {
|
||||
int tableId = data.get() & 0xff;
|
||||
int w0 = data.getShort() & 0xffff;
|
||||
if ((w0 & 0xC000) != 0x8000) {
|
||||
return null;
|
||||
}
|
||||
|
||||
int sectionLength = w0 & 0xfff;
|
||||
|
||||
data.limit(data.position() + sectionLength);
|
||||
|
||||
int specificId = data.getShort() & 0xffff;
|
||||
int b0 = data.get() & 0xff;
|
||||
int versionNumber = (b0 >> 1) & 0x1f;
|
||||
int currentNextIndicator = b0 & 1;
|
||||
|
||||
int sectionNumber = data.get() & 0xff;
|
||||
int lastSectionNumber = data.get() & 0xff;
|
||||
|
||||
return new PSISection(tableId, specificId, versionNumber, currentNextIndicator, sectionNumber,
|
||||
lastSectionNumber);
|
||||
}
|
||||
|
||||
public int getTableId() {
|
||||
return tableId;
|
||||
}
|
||||
|
||||
public int getSpecificId() {
|
||||
return specificId;
|
||||
}
|
||||
|
||||
public int getVersionNumber() {
|
||||
return versionNumber;
|
||||
}
|
||||
|
||||
public int getCurrentNextIndicator() {
|
||||
return currentNextIndicator;
|
||||
}
|
||||
|
||||
public int getSectionNumber() {
|
||||
return sectionNumber;
|
||||
}
|
||||
|
||||
public int getLastSectionNumber() {
|
||||
return lastSectionNumber;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,37 +0,0 @@
|
|||
package org.taktik.mpegts;
|
||||
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
public abstract class PacketSupport {
|
||||
protected ByteBuffer buffer;
|
||||
protected boolean dirty;
|
||||
|
||||
public PacketSupport() {
|
||||
dirty = false;
|
||||
}
|
||||
|
||||
public PacketSupport(ByteBuffer buffer) {
|
||||
this.buffer = buffer;
|
||||
buffer.rewind();
|
||||
parse();
|
||||
buffer.rewind();
|
||||
dirty = false;
|
||||
}
|
||||
|
||||
public ByteBuffer getBuffer() {
|
||||
if (dirty) {
|
||||
write();
|
||||
buffer.rewind();
|
||||
dirty = false;
|
||||
}
|
||||
return buffer;
|
||||
}
|
||||
|
||||
protected abstract void parse();
|
||||
protected abstract void write();
|
||||
|
||||
protected void markDirty() {
|
||||
dirty = true;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,347 +0,0 @@
|
|||
package org.taktik.mpegts;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.TreeMap;
|
||||
import java.util.concurrent.ArrayBlockingQueue;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.taktik.mpegts.sinks.MTSSink;
|
||||
import org.taktik.mpegts.sources.MTSSource;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
import com.google.common.collect.Maps;
|
||||
|
||||
public class Streamer {
|
||||
static final Logger log = LoggerFactory.getLogger("streamer");
|
||||
|
||||
private MTSSource source;
|
||||
private MTSSink sink;
|
||||
|
||||
private ArrayBlockingQueue<MTSPacket> buffer;
|
||||
private int bufferSize;
|
||||
private boolean endOfSourceReached;
|
||||
private boolean streamingShouldStop;
|
||||
|
||||
private PATSection patSection;
|
||||
private TreeMap<Integer,PMTSection> pmtSection;
|
||||
|
||||
private Thread bufferingThread;
|
||||
private Thread streamingThread;
|
||||
|
||||
private boolean sleepingEnabled;
|
||||
private String name;
|
||||
|
||||
private Streamer(MTSSource source, MTSSink sink, int bufferSize, boolean sleepingEnabled, String name) {
|
||||
this.source = source;
|
||||
this.sink = sink;
|
||||
this.bufferSize = bufferSize;
|
||||
this.sleepingEnabled = sleepingEnabled;
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
public void stream() throws InterruptedException {
|
||||
buffer = new ArrayBlockingQueue<>(bufferSize);
|
||||
patSection = null;
|
||||
pmtSection = Maps.newTreeMap();
|
||||
endOfSourceReached = false;
|
||||
streamingShouldStop = false;
|
||||
log.info("PreBuffering {} packets", bufferSize);
|
||||
try {
|
||||
preBuffer();
|
||||
} catch (Exception e) {
|
||||
throw new IllegalStateException("Error while buffering", e);
|
||||
}
|
||||
log.info("Done PreBuffering");
|
||||
|
||||
bufferingThread = new Thread(this::fillBuffer, "Buffering ["+name+"]");
|
||||
bufferingThread.setDaemon(true);
|
||||
bufferingThread.start();
|
||||
|
||||
streamingThread = new Thread(this::internalStream, "Streaming ["+name+"]");
|
||||
streamingThread.setDaemon(true);
|
||||
streamingThread.start();
|
||||
|
||||
bufferingThread.join();
|
||||
streamingThread.join();
|
||||
|
||||
try {
|
||||
sink.close();
|
||||
} catch(Exception e) {
|
||||
log.error("Couldn't close sink", e);
|
||||
}
|
||||
}
|
||||
|
||||
public void stop() {
|
||||
streamingShouldStop = true;
|
||||
try {
|
||||
source.close();
|
||||
} catch (Exception e) {
|
||||
log.error("Couldn't close source", e);
|
||||
}
|
||||
try {
|
||||
sink.close();
|
||||
} catch (Exception e) {
|
||||
log.error("Couldn't close sink", e);
|
||||
}
|
||||
buffer.clear();
|
||||
try {
|
||||
bufferingThread.interrupt();
|
||||
streamingThread.interrupt();
|
||||
} catch (Exception e) {
|
||||
log.error("Couldn't interrupt streamer threads");
|
||||
}
|
||||
}
|
||||
|
||||
public void switchSink(MTSSink sink) {
|
||||
MTSSink old = this.sink;
|
||||
this.sink = sink;
|
||||
try {
|
||||
old.close();
|
||||
} catch (Exception e) {
|
||||
log.error("Couldn't close old sink while switching sinks", e);
|
||||
}
|
||||
}
|
||||
|
||||
private void internalStream() {
|
||||
boolean resetState = false;
|
||||
MTSPacket packet = null;
|
||||
long packetCount = 0;
|
||||
Long firstPcrValue = null;
|
||||
Long firstPcrTime = null;
|
||||
Long lastPcrValue = null;
|
||||
Long lastPcrTime = null;
|
||||
Long averageSleep = null;
|
||||
while (!streamingShouldStop) {
|
||||
if (resetState) {
|
||||
firstPcrValue = null;
|
||||
firstPcrTime = null;
|
||||
lastPcrValue = null;
|
||||
lastPcrTime = null;
|
||||
averageSleep = null;
|
||||
resetState = false;
|
||||
}
|
||||
|
||||
// Initialize time to sleep
|
||||
long sleepNanos = 0;
|
||||
|
||||
try {
|
||||
packet = buffer.take();
|
||||
if (packet == null) {
|
||||
if (endOfSourceReached) {
|
||||
packet = buffer.take();
|
||||
if (packet == null) {
|
||||
break;
|
||||
}
|
||||
} else {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
} catch (InterruptedException e1) {
|
||||
if(!endOfSourceReached && !streamingShouldStop) {
|
||||
log.error("Interrupted while waiting for packet");
|
||||
continue;
|
||||
} else {
|
||||
Thread.currentThread().interrupt();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
int pid = packet.getPid();
|
||||
|
||||
if (pid == 0 && packet.isPayloadUnitStartIndicator()) {
|
||||
ByteBuffer payload = packet.getPayload();
|
||||
payload.rewind();
|
||||
int pointer = payload.get() & 0xff;
|
||||
payload.position(payload.position() + pointer);
|
||||
patSection = PATSection.parse(payload);
|
||||
for (Integer pmtPid : pmtSection.keySet()) {
|
||||
if (!patSection.getPrograms().values().contains(pmtPid)) {
|
||||
pmtSection.remove(pmtPid);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (pid != 0 && patSection!=null && patSection.getPrograms().values().contains(pid) && packet.isPayloadUnitStartIndicator()) {
|
||||
ByteBuffer payload = packet.getPayload();
|
||||
payload.rewind();
|
||||
int pointer = payload.get() & 0xff;
|
||||
payload.position(payload.position() + pointer);
|
||||
pmtSection.put(pid, PMTSection.parse(payload));
|
||||
}
|
||||
|
||||
// Check PID matches PCR PID
|
||||
if (averageSleep != null) {
|
||||
sleepNanos = averageSleep;
|
||||
}
|
||||
|
||||
// Check for PCR
|
||||
if (packet.getAdaptationField() != null && packet.getAdaptationField().getPcr() != null) {
|
||||
if (packet.getPid() == getPCRPid()) {
|
||||
if (!packet.getAdaptationField().isDiscontinuityIndicator()) {
|
||||
// Get PCR and current nano time
|
||||
long pcrValue = packet.getAdaptationField().getPcr().getValue();
|
||||
long pcrTime = System.nanoTime();
|
||||
|
||||
// Compute sleepNanosOrig
|
||||
if (firstPcrValue == null || firstPcrTime == null) {
|
||||
firstPcrValue = pcrValue;
|
||||
firstPcrTime = pcrTime;
|
||||
}
|
||||
|
||||
// Compute sleepNanosPrevious
|
||||
Long sleepNanosPrevious = null;
|
||||
if (lastPcrValue != null && lastPcrTime != null) {
|
||||
if (pcrValue <= lastPcrValue) {
|
||||
log.trace("PCR discontinuity ! {}", packet.getPid());
|
||||
resetState = true;
|
||||
} else {
|
||||
sleepNanosPrevious = ((pcrValue - lastPcrValue) / 27 * 1000) - (pcrTime - lastPcrTime);
|
||||
}
|
||||
}
|
||||
|
||||
// Set sleep time based on PCR if possible
|
||||
if (sleepNanosPrevious != null) {
|
||||
// Safety : We should never have to wait more than 100ms
|
||||
if (sleepNanosPrevious > 100000000) {
|
||||
log.warn("PCR sleep ignored, too high !");
|
||||
resetState = true;
|
||||
} else {
|
||||
sleepNanos = sleepNanosPrevious;
|
||||
}
|
||||
}
|
||||
|
||||
// Set lastPcrValue/lastPcrTime
|
||||
lastPcrValue = pcrValue;
|
||||
lastPcrTime = pcrTime + sleepNanos;
|
||||
} else {
|
||||
log.warn("Skipped PCR - Discontinuity indicator");
|
||||
}
|
||||
} else {
|
||||
log.debug("Skipped PCR - PID does not match");
|
||||
}
|
||||
}
|
||||
|
||||
// Sleep if needed
|
||||
if (sleepNanos > 0 && sleepingEnabled) {
|
||||
log.trace("Sleeping {} millis, {} nanos", sleepNanos / 1000000, sleepNanos % 1000000);
|
||||
try {
|
||||
Thread.sleep(sleepNanos / 1000000, (int) (sleepNanos % 1000000));
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.warn("Streaming sleep interrupted!");
|
||||
}
|
||||
}
|
||||
|
||||
// Stream packet
|
||||
if(!streamingShouldStop && !Thread.interrupted()) {
|
||||
try {
|
||||
sink.send(packet);
|
||||
} catch (Exception e) {
|
||||
log.error("Error sending packet to sink", e);
|
||||
}
|
||||
}
|
||||
|
||||
packetCount++;
|
||||
}
|
||||
log.info("Sent {} MPEG-TS packets", packetCount);
|
||||
synchronized (this) {
|
||||
notifyAll();
|
||||
}
|
||||
}
|
||||
|
||||
private void preBuffer() throws Exception {
|
||||
MTSPacket packet;
|
||||
int packetNumber = 0;
|
||||
while ((packetNumber < bufferSize) && (packet = source.nextPacket()) != null) {
|
||||
buffer.add(packet);
|
||||
packetNumber++;
|
||||
}
|
||||
}
|
||||
|
||||
private void fillBuffer() {
|
||||
try {
|
||||
MTSPacket packet;
|
||||
while (!streamingShouldStop && (packet = source.nextPacket()) != null) {
|
||||
boolean put = false;
|
||||
while (!put) {
|
||||
put = putPacketToBuffer(packet);
|
||||
}
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.error("Error reading from source", e);
|
||||
} catch (Exception e) {
|
||||
log.error("Error reading from source", e);
|
||||
} finally {
|
||||
endOfSourceReached = true;
|
||||
try {
|
||||
streamingThread.interrupt();
|
||||
} catch(Exception e) {
|
||||
log.error("Couldn't interrupt streaming thread", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private boolean putPacketToBuffer(MTSPacket packet) {
|
||||
try {
|
||||
buffer.put(packet);
|
||||
return true;
|
||||
} catch (InterruptedException ignored) {
|
||||
Thread.currentThread().interrupt();
|
||||
log.error("Error adding packet to buffer", ignored);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private int getPCRPid() {
|
||||
if ((!pmtSection.isEmpty())) {
|
||||
return pmtSection.values().iterator().next().getPcrPid();
|
||||
}
|
||||
return -1;
|
||||
}
|
||||
|
||||
public static StreamerBuilder builder() {
|
||||
return new StreamerBuilder();
|
||||
}
|
||||
|
||||
public static class StreamerBuilder {
|
||||
private MTSSink sink;
|
||||
private MTSSource source;
|
||||
private int bufferSize = 1000;
|
||||
private boolean sleepingEnabled = false;
|
||||
private String name;
|
||||
|
||||
public StreamerBuilder setSink(MTSSink sink) {
|
||||
this.sink = sink;
|
||||
return this;
|
||||
}
|
||||
|
||||
public StreamerBuilder setSource(MTSSource source) {
|
||||
this.source = source;
|
||||
return this;
|
||||
}
|
||||
|
||||
public StreamerBuilder setBufferSize(int bufferSize) {
|
||||
this.bufferSize = bufferSize;
|
||||
return this;
|
||||
}
|
||||
|
||||
public StreamerBuilder setSleepingEnabled(boolean sleepingEnabled) {
|
||||
this.sleepingEnabled = sleepingEnabled;
|
||||
return this;
|
||||
}
|
||||
|
||||
public StreamerBuilder setName(String name) {
|
||||
this.name = name;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Streamer build() {
|
||||
Preconditions.checkNotNull(sink);
|
||||
Preconditions.checkNotNull(source);
|
||||
return new Streamer(source, sink, bufferSize, sleepingEnabled, name);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -1,45 +0,0 @@
|
|||
package org.taktik.mpegts;
|
||||
|
||||
import java.io.File;
|
||||
|
||||
import org.taktik.mpegts.sinks.MTSSink;
|
||||
import org.taktik.mpegts.sinks.UDPTransport;
|
||||
import org.taktik.mpegts.sources.MTSSource;
|
||||
import org.taktik.mpegts.sources.MTSSources;
|
||||
import org.taktik.mpegts.sources.ResettableMTSSource;
|
||||
|
||||
public class StreamerTest {
|
||||
public static void main(String[] args) throws Exception {
|
||||
|
||||
// Set up mts sink
|
||||
MTSSink transport = UDPTransport.builder()
|
||||
//.setAddress("239.222.1.1")
|
||||
.setAddress("127.0.0.1")
|
||||
.setPort(1234)
|
||||
.setSoTimeout(5000)
|
||||
.setTtl(1)
|
||||
.build();
|
||||
|
||||
|
||||
ResettableMTSSource ts1 = MTSSources.from(new File("/Users/abaudoux/Downloads/EBSrecording.mpg"));
|
||||
|
||||
// media132, media133 --> ok
|
||||
// media133, media132 --> ok
|
||||
// media123, media132 --> ko
|
||||
|
||||
|
||||
// Build source
|
||||
MTSSource source = MTSSources.loop(ts1);
|
||||
|
||||
// build streamer
|
||||
Streamer streamer = Streamer.builder()
|
||||
.setSource(source)
|
||||
//.setSink(ByteChannelSink.builder().setByteChannel(fc).build())
|
||||
.setSink(transport)
|
||||
.build();
|
||||
|
||||
// Start streaming
|
||||
streamer.stream();
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -1,43 +0,0 @@
|
|||
package org.taktik.mpegts.sinks;
|
||||
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
import java.nio.channels.ByteChannel;
|
||||
|
||||
public class ByteChannelSink implements MTSSink {
|
||||
|
||||
private ByteChannel byteChannel;
|
||||
|
||||
private ByteChannelSink(ByteChannel byteChannel) {
|
||||
this.byteChannel = byteChannel;
|
||||
}
|
||||
|
||||
public static ByteChannelSinkBuilder builder() {
|
||||
return new ByteChannelSinkBuilder();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void send(MTSPacket packet) throws Exception {
|
||||
byteChannel.write(packet.getBuffer());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws Exception {
|
||||
byteChannel.close();
|
||||
}
|
||||
|
||||
public static class ByteChannelSinkBuilder {
|
||||
private ByteChannel byteChannel;
|
||||
|
||||
private ByteChannelSinkBuilder(){}
|
||||
|
||||
public ByteChannelSink build() {
|
||||
return new ByteChannelSink(byteChannel);
|
||||
}
|
||||
|
||||
public ByteChannelSinkBuilder setByteChannel(ByteChannel byteChannel) {
|
||||
this.byteChannel = byteChannel;
|
||||
return this;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,7 +0,0 @@
|
|||
package org.taktik.mpegts.sinks;
|
||||
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
public interface MTSSink extends AutoCloseable {
|
||||
void send(MTSPacket packet) throws Exception;
|
||||
}
|
||||
|
|
@ -1,77 +0,0 @@
|
|||
package org.taktik.mpegts.sinks;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.DatagramPacket;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.MulticastSocket;
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
public class UDPTransport implements MTSSink {
|
||||
|
||||
private final InetSocketAddress inetSocketAddress;
|
||||
private final MulticastSocket multicastSocket;
|
||||
|
||||
|
||||
private UDPTransport(String address, int port, int ttl, int soTimeout) throws IOException {
|
||||
// InetSocketAddress
|
||||
inetSocketAddress = new InetSocketAddress(address, port);
|
||||
|
||||
// Create the socket but we don't bind it as we are only going to send data
|
||||
// Note that we don't have to join the multicast group if we are only sending data and not receiving
|
||||
multicastSocket = new MulticastSocket();
|
||||
multicastSocket.setReuseAddress(true);
|
||||
multicastSocket.setSoTimeout(soTimeout);
|
||||
multicastSocket.setTimeToLive(ttl);
|
||||
}
|
||||
|
||||
public static UDPTransport.UDPTransportBuilder builder() {
|
||||
return new UDPTransportBuilder();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void send(MTSPacket packet) throws IOException {
|
||||
ByteBuffer buffer = packet.getBuffer();
|
||||
Preconditions.checkArgument(buffer.hasArray());
|
||||
DatagramPacket datagramPacket = new DatagramPacket(buffer.array(), buffer.arrayOffset(), buffer.limit(), inetSocketAddress);
|
||||
multicastSocket.send(datagramPacket);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
multicastSocket.close();
|
||||
}
|
||||
|
||||
public static class UDPTransportBuilder {
|
||||
private String address;
|
||||
private int port;
|
||||
private int ttl;
|
||||
private int soTimeout;
|
||||
|
||||
public UDPTransportBuilder setAddress(String address) {
|
||||
this.address = address;
|
||||
return this;
|
||||
}
|
||||
|
||||
public UDPTransportBuilder setPort(int port) {
|
||||
this.port = port;
|
||||
return this;
|
||||
}
|
||||
|
||||
public UDPTransportBuilder setTtl(int ttl) {
|
||||
this.ttl = ttl;
|
||||
return this;
|
||||
}
|
||||
|
||||
public UDPTransportBuilder setSoTimeout(int timeout) {
|
||||
this.soTimeout = timeout;
|
||||
return this;
|
||||
}
|
||||
|
||||
public UDPTransport build() throws IOException {
|
||||
return new UDPTransport(address, port, ttl, soTimeout);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,117 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.ByteChannel;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.taktik.ioutils.NIOUtils;
|
||||
import org.taktik.mpegts.Constants;
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
public abstract class AbstractByteChannelMTSSource<T extends ByteChannel> extends AbstractMTSSource {
|
||||
static final Logger log = LoggerFactory.getLogger("source");
|
||||
|
||||
private static final int BUFFER_SIZE = Constants.MPEGTS_PACKET_SIZE * 1000;
|
||||
|
||||
protected ByteBuffer buffer;
|
||||
protected T byteChannel;
|
||||
|
||||
|
||||
protected AbstractByteChannelMTSSource(T byteChannel) throws IOException {
|
||||
this.byteChannel = byteChannel;
|
||||
fillBuffer();
|
||||
}
|
||||
|
||||
protected void fillBuffer() throws IOException {
|
||||
buffer = ByteBuffer.allocate(BUFFER_SIZE);
|
||||
NIOUtils.read(byteChannel, buffer);
|
||||
buffer.flip();
|
||||
}
|
||||
|
||||
protected boolean lastBuffer() {
|
||||
return buffer.capacity() > buffer.limit();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MTSPacket nextPacketInternal() throws IOException {
|
||||
ByteBuffer packetBuffer = null;
|
||||
while (true) {
|
||||
boolean foundFirstMarker = false;
|
||||
int skipped = 0;
|
||||
while (!foundFirstMarker) {
|
||||
if (!buffer.hasRemaining()) {
|
||||
if (lastBuffer()) {
|
||||
return null;
|
||||
}
|
||||
buffer = ByteBuffer.allocate(BUFFER_SIZE);
|
||||
if (NIOUtils.read(byteChannel, buffer) <= 0) {
|
||||
return null;
|
||||
}
|
||||
buffer.flip();
|
||||
}
|
||||
if ((buffer.get(buffer.position()) & 0xff) == Constants.TS_MARKER) {
|
||||
foundFirstMarker = true;
|
||||
} else {
|
||||
buffer.position(buffer.position() + 1);
|
||||
skipped++;
|
||||
}
|
||||
}
|
||||
if (skipped > 0) {
|
||||
log.info("Skipped {} bytes looking for TS marker", skipped);
|
||||
}
|
||||
if (buffer.remaining() >= Constants.MPEGTS_PACKET_SIZE) {
|
||||
if ((buffer.remaining() == Constants.MPEGTS_PACKET_SIZE) ||
|
||||
(buffer.get(buffer.position() + Constants.MPEGTS_PACKET_SIZE) & 0xff) == Constants.TS_MARKER) {
|
||||
packetBuffer = buffer.slice();
|
||||
packetBuffer.limit(Constants.MPEGTS_PACKET_SIZE);
|
||||
buffer.position(buffer.position() + Constants.MPEGTS_PACKET_SIZE);
|
||||
} else {
|
||||
log.info("no second marker found");
|
||||
buffer.position(buffer.position() + 1);
|
||||
}
|
||||
} else if (!lastBuffer()) {
|
||||
log.info("NEW BUFFER");
|
||||
|
||||
ByteBuffer newBuffer = ByteBuffer.allocate(BUFFER_SIZE);
|
||||
newBuffer.put(buffer);
|
||||
buffer = newBuffer;
|
||||
if (NIOUtils.read(byteChannel, buffer) <= 0) {
|
||||
return null;
|
||||
}
|
||||
buffer.flip();
|
||||
if (buffer.remaining() >= Constants.MPEGTS_PACKET_SIZE) {
|
||||
if ((buffer.remaining() == Constants.MPEGTS_PACKET_SIZE) ||
|
||||
(buffer.get(buffer.position() + Constants.MPEGTS_PACKET_SIZE) & 0xff) == Constants.TS_MARKER) {
|
||||
packetBuffer = buffer.slice();
|
||||
packetBuffer.limit(Constants.MPEGTS_PACKET_SIZE);
|
||||
buffer.position(buffer.position() + Constants.MPEGTS_PACKET_SIZE);
|
||||
} else {
|
||||
log.info("no second marker found");
|
||||
buffer.position(buffer.position() + 1);
|
||||
}
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (packetBuffer != null) {
|
||||
// Parse the packet
|
||||
try {
|
||||
return new MTSPacket(packetBuffer);
|
||||
} catch (Exception e) {
|
||||
packetBuffer = null;
|
||||
log.warn("Error parsing packet", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void closeInternal() throws Exception {
|
||||
byteChannel.close();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,31 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
public abstract class AbstractMTSSource implements MTSSource {
|
||||
private boolean closed;
|
||||
|
||||
@Override
|
||||
public final MTSPacket nextPacket() throws Exception {
|
||||
if (isClosed()) {
|
||||
throw new IllegalStateException("Source is closed");
|
||||
}
|
||||
return nextPacketInternal();
|
||||
}
|
||||
|
||||
@Override
|
||||
public final void close() throws Exception {
|
||||
try {
|
||||
closeInternal();
|
||||
} finally {
|
||||
closed = true;
|
||||
}
|
||||
}
|
||||
|
||||
protected boolean isClosed() {
|
||||
return closed;
|
||||
}
|
||||
|
||||
protected abstract MTSPacket nextPacketInternal() throws Exception;
|
||||
protected abstract void closeInternal() throws Exception;
|
||||
}
|
||||
|
|
@ -1,135 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
import ctbrec.recorder.ProgressListener;
|
||||
|
||||
public class BlockingMultiMTSSource extends AbstractMTSSource implements AutoCloseable {
|
||||
|
||||
private static final transient Logger LOG = LoggerFactory.getLogger(BlockingMultiMTSSource.class);
|
||||
|
||||
private final boolean fixContinuity;
|
||||
private ContinuityFixer continuityFixer;
|
||||
|
||||
private final BlockingQueue<MTSSource> sources;
|
||||
private MTSSource currentSource;
|
||||
private int downloadedSegments = 0;
|
||||
private int totalSegments = -1;
|
||||
private ProgressListener listener;
|
||||
private int lastProgress = 0;
|
||||
|
||||
private BlockingMultiMTSSource(boolean fixContinuity) {
|
||||
this.fixContinuity = fixContinuity;
|
||||
if (fixContinuity) {
|
||||
continuityFixer = new ContinuityFixer();
|
||||
}
|
||||
this.sources = new LinkedBlockingQueue<>(10);
|
||||
}
|
||||
|
||||
public void addSource(MTSSource source) throws InterruptedException {
|
||||
this.sources.put(source);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MTSPacket nextPacketInternal() throws Exception {
|
||||
if(currentSource == null) {
|
||||
currentSource = sources.take();
|
||||
}
|
||||
|
||||
MTSPacket packet = currentSource.nextPacket();
|
||||
packet = switchSourceIfNeeded(packet);
|
||||
|
||||
if (fixContinuity) {
|
||||
try {
|
||||
continuityFixer.fixContinuity(packet);
|
||||
} catch(Exception e) {
|
||||
LOG.warn("Failed to fix continuity. MTSPacket probably invalid");
|
||||
return nextPacketInternal();
|
||||
}
|
||||
}
|
||||
return packet;
|
||||
}
|
||||
|
||||
private MTSPacket switchSourceIfNeeded(MTSPacket packet) throws Exception {
|
||||
if(packet == null) {
|
||||
// end of source has been reached, switch to the next source
|
||||
closeCurrentSource();
|
||||
|
||||
downloadedSegments++;
|
||||
if(listener != null && totalSegments > 0) {
|
||||
int progress = (int)(downloadedSegments * 100.0 / totalSegments);
|
||||
if(progress > lastProgress) {
|
||||
listener.update(progress);
|
||||
lastProgress = progress;
|
||||
}
|
||||
}
|
||||
if(downloadedSegments == totalSegments) {
|
||||
LOG.debug("All segments written. Queue size {}", sources.size());
|
||||
return null;
|
||||
}
|
||||
|
||||
return firstPacketFromNextSource();
|
||||
}
|
||||
return packet;
|
||||
}
|
||||
|
||||
private MTSPacket firstPacketFromNextSource() throws Exception {
|
||||
switchSource();
|
||||
return currentSource.nextPacket();
|
||||
}
|
||||
|
||||
private void switchSource() throws InterruptedException {
|
||||
currentSource = sources.take();
|
||||
}
|
||||
|
||||
private void closeCurrentSource() throws Exception {
|
||||
currentSource.close();
|
||||
}
|
||||
|
||||
private void setProgressListener(ProgressListener listener) {
|
||||
this.listener = listener;
|
||||
}
|
||||
|
||||
public void setTotalSegments(int total) {
|
||||
this.totalSegments = total;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void closeInternal() throws Exception {
|
||||
for (MTSSource source : sources) {
|
||||
source.close();
|
||||
}
|
||||
}
|
||||
|
||||
public static BlockingMultiMTSSourceBuilder builder() {
|
||||
return new BlockingMultiMTSSourceBuilder();
|
||||
}
|
||||
|
||||
public static class BlockingMultiMTSSourceBuilder {
|
||||
boolean fixContinuity = false;
|
||||
ProgressListener listener;
|
||||
|
||||
public BlockingMultiMTSSourceBuilder setFixContinuity(boolean fixContinuity) {
|
||||
this.fixContinuity = fixContinuity;
|
||||
return this;
|
||||
}
|
||||
public BlockingMultiMTSSourceBuilder setProgressListener(ProgressListener listener) {
|
||||
this.listener = listener;
|
||||
return this;
|
||||
}
|
||||
|
||||
public BlockingMultiMTSSource build() {
|
||||
BlockingMultiMTSSource source = new BlockingMultiMTSSource(fixContinuity);
|
||||
if(listener != null) {
|
||||
source.setProgressListener(listener);
|
||||
}
|
||||
return source;
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -1,33 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.channels.ByteChannel;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
|
||||
public class ByteChannelMTSSource extends AbstractByteChannelMTSSource<ByteChannel> {
|
||||
|
||||
private ByteChannelMTSSource(ByteChannel byteChannel) throws IOException {
|
||||
super(byteChannel);
|
||||
}
|
||||
|
||||
public static ByteChannelMTSSourceBuilder builder() {
|
||||
return new ByteChannelMTSSourceBuilder();
|
||||
}
|
||||
|
||||
public static class ByteChannelMTSSourceBuilder {
|
||||
private ByteChannel byteChannel;
|
||||
|
||||
private ByteChannelMTSSourceBuilder(){}
|
||||
|
||||
public ByteChannelMTSSourceBuilder setByteChannel(ByteChannel byteChannel) {
|
||||
this.byteChannel = byteChannel;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ByteChannelMTSSource build() throws IOException {
|
||||
Preconditions.checkNotNull(byteChannel, "byteChannel cannot be null");
|
||||
return new ByteChannelMTSSource(byteChannel);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,75 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
import com.google.common.io.ByteSource;
|
||||
import org.taktik.mpegts.Constants;
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
import java.io.InputStream;
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
public class ByteSourceMTSSource extends AbstractMTSSource implements ResettableMTSSource {
|
||||
|
||||
private ByteSource byteSource;
|
||||
|
||||
private InputStream stream;
|
||||
|
||||
|
||||
private ByteSourceMTSSource(ByteSource byteSource) {
|
||||
this.byteSource = byteSource;
|
||||
}
|
||||
|
||||
public static ByteSourceMTSSourceBuilder builder() {
|
||||
return new ByteSourceMTSSourceBuilder();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void reset() throws Exception {
|
||||
if (stream != null) {
|
||||
stream.close();
|
||||
}
|
||||
stream = byteSource.openBufferedStream();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MTSPacket nextPacketInternal() throws Exception {
|
||||
if (stream == null) {
|
||||
stream = byteSource.openBufferedStream();
|
||||
}
|
||||
|
||||
byte[] barray = new byte[Constants.MPEGTS_PACKET_SIZE];
|
||||
if (stream.read(barray) != Constants.MPEGTS_PACKET_SIZE) {
|
||||
stream.close();
|
||||
return null;
|
||||
}
|
||||
|
||||
// Parse the packet
|
||||
return new MTSPacket(ByteBuffer.wrap(barray));
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void closeInternal() throws Exception {
|
||||
if (stream != null) {
|
||||
try (InputStream ignored = stream){
|
||||
//close
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static class ByteSourceMTSSourceBuilder {
|
||||
private ByteSource byteSource;
|
||||
|
||||
private ByteSourceMTSSourceBuilder() {
|
||||
}
|
||||
|
||||
public ByteSourceMTSSource build() {
|
||||
Preconditions.checkNotNull(byteSource);
|
||||
return new ByteSourceMTSSource(byteSource);
|
||||
}
|
||||
|
||||
public ByteSourceMTSSourceBuilder setByteSource(ByteSource byteSource) {
|
||||
this.byteSource = byteSource;
|
||||
return this;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,194 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
import com.google.common.collect.Lists;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
|
||||
public class ConcatenatingMTSSource extends AbstractMTSSource {
|
||||
static final Logger log = LoggerFactory.getLogger("multisource");
|
||||
private List<MTSSource> sources;
|
||||
private MTSSource currentSource;
|
||||
private int idx;
|
||||
private boolean fixContinuity;
|
||||
|
||||
private ContinuityFixer continuityFixer;
|
||||
private int maxLoops;
|
||||
private int currentLoop;
|
||||
private boolean closeCurrentSource;
|
||||
|
||||
protected ConcatenatingMTSSource(boolean fixContinuity, int maxloops, Collection<MTSSource> sources) {
|
||||
Preconditions.checkArgument(sources.size() > 0, "Must provide at least contain one source");
|
||||
Preconditions.checkArgument(maxloops != 0, "Cannot loop zero times");
|
||||
this.sources = Lists.newArrayList(sources);
|
||||
this.fixContinuity = fixContinuity;
|
||||
idx = 0;
|
||||
currentSource = this.sources.get(0);
|
||||
if (fixContinuity) {
|
||||
continuityFixer = new ContinuityFixer();
|
||||
}
|
||||
this.maxLoops = maxloops;
|
||||
if (maxloops != 1) {
|
||||
checkLoopingPossible(sources);
|
||||
}
|
||||
this.currentLoop = 1;
|
||||
this.closeCurrentSource = false;
|
||||
}
|
||||
|
||||
public static MultiMTSSourceBuilder builder() {
|
||||
return new MultiMTSSourceBuilder();
|
||||
}
|
||||
|
||||
private void checkLoopingPossible(Collection<MTSSource> sources) {
|
||||
for (MTSSource source : sources) {
|
||||
checkLoopingPossible(source);
|
||||
}
|
||||
}
|
||||
|
||||
private void checkLoopingPossible(MTSSource source) {
|
||||
if (!(source instanceof ResettableMTSSource)) {
|
||||
throw new IllegalStateException("Sources must be resettable for looping");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MTSPacket nextPacketInternal() throws Exception {
|
||||
if (currentSource == null) {
|
||||
return null;
|
||||
}
|
||||
MTSPacket tsPacket = currentSource.nextPacket();
|
||||
if (tsPacket != null) {
|
||||
if (fixContinuity) {
|
||||
continuityFixer.fixContinuity(tsPacket);
|
||||
}
|
||||
return tsPacket;
|
||||
} else {
|
||||
// FIXME: infinite loop
|
||||
nextSource();
|
||||
return nextPacket();
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized void updateSources(List<MTSSource> newSources) {
|
||||
checkLoopingPossible(newSources);
|
||||
List<MTSSource> oldSources = this.sources;
|
||||
this.sources = newSources;
|
||||
for (MTSSource oldSource : oldSources) {
|
||||
if (!newSources.contains(oldSource) && oldSource != currentSource) {
|
||||
try {
|
||||
oldSource.close();
|
||||
} catch (Exception e) {
|
||||
log.error("Error closing source", e);
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
closeCurrentSource = !newSources.contains(currentSource);
|
||||
// Force next call to nextSource() to pick source 0 (first source)
|
||||
idx = -1;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected synchronized void closeInternal() throws Exception {
|
||||
for (MTSSource source : sources) {
|
||||
source.close();
|
||||
}
|
||||
if (closeCurrentSource && currentSource != null && !sources.contains(currentSource)) {
|
||||
currentSource.close();
|
||||
}
|
||||
}
|
||||
|
||||
private synchronized void nextSource() {
|
||||
if (closeCurrentSource) {
|
||||
try {
|
||||
currentSource.close();
|
||||
} catch (Exception e) {
|
||||
log.error("Error closing source", e);
|
||||
} finally {
|
||||
closeCurrentSource = false;
|
||||
}
|
||||
}
|
||||
if (fixContinuity) {
|
||||
continuityFixer.nextSource();
|
||||
}
|
||||
idx++;
|
||||
if (idx >= sources.size()) {
|
||||
currentLoop++;
|
||||
if (maxLoops > 0 && currentLoop > maxLoops) {
|
||||
currentSource = null;
|
||||
} else {
|
||||
idx = 0;
|
||||
for (MTSSource source : sources) {
|
||||
try {
|
||||
((ResettableMTSSource) source).reset();
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
currentSource = sources.get(idx);
|
||||
}
|
||||
} else {
|
||||
currentSource = sources.get(idx);
|
||||
}
|
||||
if (idx < sources.size()) {
|
||||
log.info("Switched to source #{}", idx);
|
||||
}
|
||||
}
|
||||
|
||||
public static class MultiMTSSourceBuilder {
|
||||
boolean fixContinuity = false;
|
||||
private List<MTSSource> sources = Lists.newArrayList();
|
||||
private int maxLoops = 1;
|
||||
|
||||
private MultiMTSSourceBuilder() {
|
||||
}
|
||||
|
||||
public MultiMTSSourceBuilder addSource(MTSSource source) {
|
||||
this.sources.add(source);
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiMTSSourceBuilder addSources(Collection<MTSSource> sources) {
|
||||
this.sources.addAll(sources);
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiMTSSourceBuilder setSources(MTSSource... sources) {
|
||||
this.sources = Lists.newArrayList(sources);
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiMTSSourceBuilder setSources(Collection<MTSSource> sources) {
|
||||
this.sources = Lists.newArrayList(sources);
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiMTSSourceBuilder setFixContinuity(boolean fixContinuity) {
|
||||
this.fixContinuity = fixContinuity;
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiMTSSourceBuilder loop() {
|
||||
this.maxLoops = -1;
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiMTSSourceBuilder loops(int count) {
|
||||
this.maxLoops = count;
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiMTSSourceBuilder noLoop() {
|
||||
this.maxLoops = 1;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ConcatenatingMTSSource build() {
|
||||
return new ConcatenatingMTSSource(fixContinuity, maxLoops, sources);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,214 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.Map;
|
||||
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
import com.google.common.collect.Maps;
|
||||
|
||||
|
||||
/**
|
||||
* This class will attempt to fix timestamp discontinuities
|
||||
* when switching from one source to another.
|
||||
* This should allow for smoother transitions between videos.<br>
|
||||
* This class does 3 things:
|
||||
* <ol>
|
||||
* <li> Rewrite the PCR to be continuous with the previous source</li>
|
||||
* <li> Rewrite the PTS of the PES to be continuous with the previous source</li>
|
||||
* <li> Rewrite the continuity counter to be continuous with the previous source</li>
|
||||
* </ol>
|
||||
*
|
||||
* Code using this class should call {@link #fixContinuity(org.taktik.mpegts.MTSPacket)} for each source packet,
|
||||
* then {@link #nextSource()} after the last packet of the current source and before the first packet of the next source.
|
||||
*/
|
||||
public class ContinuityFixer {
|
||||
private Map<Integer, MTSPacket> pcrPackets;
|
||||
private Map<Integer, MTSPacket> allPackets;
|
||||
private Map<Integer, Long> ptss;
|
||||
private Map<Integer, Long> lastPTSsOfPreviousSource;
|
||||
private Map<Integer, Long> lastPCRsOfPreviousSource;
|
||||
private Map<Integer, Long> firstPCRsOfCurrentSource;
|
||||
private Map<Integer, Long> firstPTSsOfCurrentSource;
|
||||
|
||||
private Map<Integer, MTSPacket> lastPacketsOfPreviousSource = Maps.newHashMap();
|
||||
private Map<Integer, MTSPacket> firstPacketsOfCurrentSource = Maps.newHashMap();
|
||||
private Map<Integer, Integer> continuityFixes = Maps.newHashMap();
|
||||
|
||||
private boolean firstSource;
|
||||
|
||||
|
||||
public ContinuityFixer() {
|
||||
pcrPackets = Maps.newHashMap();
|
||||
allPackets = Maps.newHashMap();
|
||||
ptss = Maps.newHashMap();
|
||||
lastPTSsOfPreviousSource = Maps.newHashMap();
|
||||
lastPCRsOfPreviousSource = Maps.newHashMap();
|
||||
firstPCRsOfCurrentSource = Maps.newHashMap();
|
||||
firstPTSsOfCurrentSource = Maps.newHashMap();
|
||||
|
||||
lastPacketsOfPreviousSource = Maps.newHashMap();
|
||||
firstPacketsOfCurrentSource = Maps.newHashMap();
|
||||
continuityFixes = Maps.newHashMap();
|
||||
firstSource = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Signals the {@link org.taktik.mpegts.sources.ContinuityFixer} that the following
|
||||
* packet will be from another source.
|
||||
*
|
||||
* Call this method after the last packet of the current source and before the first packet of the next source.
|
||||
*/
|
||||
public void nextSource() {
|
||||
firstPCRsOfCurrentSource.clear();
|
||||
lastPCRsOfPreviousSource.clear();
|
||||
firstPTSsOfCurrentSource.clear();
|
||||
lastPTSsOfPreviousSource.clear();
|
||||
firstPacketsOfCurrentSource.clear();
|
||||
lastPacketsOfPreviousSource.clear();
|
||||
for (MTSPacket mtsPacket : pcrPackets.values()) {
|
||||
lastPCRsOfPreviousSource.put(mtsPacket.getPid(), mtsPacket.getAdaptationField().getPcr().getValue());
|
||||
}
|
||||
lastPTSsOfPreviousSource.putAll(ptss);
|
||||
lastPacketsOfPreviousSource.putAll(allPackets);
|
||||
pcrPackets.clear();
|
||||
ptss.clear();
|
||||
allPackets.clear();
|
||||
firstSource = false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fix the continuity of the packet.
|
||||
*
|
||||
* Call this method for each source packet, in order.
|
||||
*
|
||||
* @param tsPacket The packet to fix.
|
||||
*/
|
||||
public void fixContinuity(MTSPacket tsPacket) {
|
||||
if(tsPacket == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
int pid = tsPacket.getPid();
|
||||
allPackets.put(pid, tsPacket);
|
||||
if (!firstPacketsOfCurrentSource.containsKey(pid)) {
|
||||
firstPacketsOfCurrentSource.put(pid, tsPacket);
|
||||
if (!firstSource) {
|
||||
MTSPacket lastPacketOfPreviousSource = lastPacketsOfPreviousSource.get(pid);
|
||||
int continuityFix = lastPacketOfPreviousSource == null ? 0 : lastPacketOfPreviousSource.getContinuityCounter() - tsPacket.getContinuityCounter();
|
||||
if (tsPacket.isContainsPayload()) {
|
||||
continuityFix++;
|
||||
}
|
||||
continuityFixes.put(pid, continuityFix);
|
||||
}
|
||||
}
|
||||
if (!firstSource) {
|
||||
tsPacket.setContinuityCounter((tsPacket.getContinuityCounter() + continuityFixes.get(pid)) % 16);
|
||||
}
|
||||
fixPTS(tsPacket, pid);
|
||||
fixPCR(tsPacket, pid);
|
||||
}
|
||||
|
||||
private void fixPCR(MTSPacket tsPacket, int pid) {
|
||||
if (tsPacket.isAdaptationFieldExist() && tsPacket.getAdaptationField() != null) {
|
||||
if (tsPacket.getAdaptationField().isPcrFlag()) {
|
||||
if (!firstPCRsOfCurrentSource.containsKey(pid)) {
|
||||
firstPCRsOfCurrentSource.put(pid, tsPacket.getAdaptationField().getPcr().getValue());
|
||||
}
|
||||
rewritePCR(tsPacket);
|
||||
pcrPackets.put(pid, tsPacket);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void fixPTS(MTSPacket tsPacket, int pid) {
|
||||
if (tsPacket.isContainsPayload()) {
|
||||
ByteBuffer payload = tsPacket.getPayload();
|
||||
//System.out.println("PKT RMN " + tsPacket.getPayload().remaining());
|
||||
if (/*payload.remaining() >= 3 &&*/ ((payload.get(0) & 0xff) == 0) && ((payload.get(1) & 0xff) == 0) && ((payload.get(2) & 0xff) == 1)) {
|
||||
int extension = payload.getShort(6) & 0xffff;
|
||||
if ((extension & 0x80) != 0) {
|
||||
// PTS is present
|
||||
// TODO add payload size check to avoid indexoutofboundexception
|
||||
long pts = (((payload.get(9) & 0xE)) << 29) | (((payload.getShort(10) & 0xFFFE)) << 14) | ((payload.getShort(12) & 0xFFFE) >> 1);
|
||||
if (!firstPTSsOfCurrentSource.containsKey(pid)) {
|
||||
firstPTSsOfCurrentSource.put(pid, pts);
|
||||
}
|
||||
if (!firstSource) {
|
||||
long newPts = Math.round(pts + (getTimeGap(pid) / 300.0) + 100 * ((27_000_000 / 300.0) / 1_000));
|
||||
|
||||
payload.put(9, (byte) (0x20 | ((newPts & 0x1C0000000l) >> 29) | 0x1));
|
||||
payload.putShort(10, (short) (0x1 | ((newPts & 0x3FFF8000) >> 14)));
|
||||
payload.putShort(12, (short) (0x1 | ((newPts & 0x7FFF) << 1)));
|
||||
payload.rewind();
|
||||
pts = newPts;
|
||||
}
|
||||
|
||||
ptss.put(pid, pts);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private long getTimeGap(int pid) {
|
||||
// Try with PCR of the same PID
|
||||
Long lastPCROfPreviousSource = lastPCRsOfPreviousSource.get(pid);
|
||||
if (lastPCROfPreviousSource == null) {
|
||||
lastPCROfPreviousSource = 0l;
|
||||
}
|
||||
Long firstPCROfCurrentSource = firstPCRsOfCurrentSource.get(pid);
|
||||
if (firstPCROfCurrentSource != null) {
|
||||
return lastPCROfPreviousSource - firstPCROfCurrentSource;
|
||||
}
|
||||
|
||||
// Try with any PCR
|
||||
if (!lastPCRsOfPreviousSource.isEmpty()) {
|
||||
int pcrPid = lastPCRsOfPreviousSource.keySet().iterator().next();
|
||||
lastPCROfPreviousSource = lastPCRsOfPreviousSource.get(pcrPid);
|
||||
if (lastPCROfPreviousSource == null) {
|
||||
lastPCROfPreviousSource = 0l;
|
||||
}
|
||||
firstPCROfCurrentSource = firstPCRsOfCurrentSource.get(pcrPid);
|
||||
if (firstPCROfCurrentSource != null) {
|
||||
return lastPCROfPreviousSource - firstPCROfCurrentSource;
|
||||
}
|
||||
}
|
||||
|
||||
// Try with PTS of the same PID
|
||||
Long lastPTSOfPreviousSource = lastPTSsOfPreviousSource.get(pid);
|
||||
if (lastPTSOfPreviousSource == null) {
|
||||
lastPTSOfPreviousSource = 0l;
|
||||
}
|
||||
|
||||
Long firstPTSofCurrentSource = firstPTSsOfCurrentSource.get(pid);
|
||||
if (firstPTSofCurrentSource != null) {
|
||||
return (lastPTSOfPreviousSource - firstPTSofCurrentSource) * 300;
|
||||
}
|
||||
|
||||
// Try with any PTS
|
||||
if (!lastPTSsOfPreviousSource.isEmpty()) {
|
||||
int randomPid = lastPTSsOfPreviousSource.keySet().iterator().next();
|
||||
lastPTSOfPreviousSource = lastPTSsOfPreviousSource.get(randomPid);
|
||||
if (lastPTSOfPreviousSource == null) {
|
||||
lastPTSOfPreviousSource = 0l;
|
||||
}
|
||||
|
||||
firstPTSofCurrentSource = firstPTSsOfCurrentSource.get(randomPid);
|
||||
if (firstPTSofCurrentSource != null) {
|
||||
return (lastPTSOfPreviousSource - firstPTSofCurrentSource) * 300;
|
||||
}
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
private void rewritePCR(MTSPacket tsPacket) {
|
||||
if (firstSource) {
|
||||
return;
|
||||
}
|
||||
long timeGap = getTimeGap(tsPacket.getPid());
|
||||
long pcr = tsPacket.getAdaptationField().getPcr().getValue();
|
||||
long newPcr = pcr + timeGap + 100 * ((27_000_000) / 1_000);
|
||||
tsPacket.getAdaptationField().getPcr().setValue(newPcr);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,30 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
/**
|
||||
* Decorates an MTSSource with continuity fixing. Not suitable for use with multiple different
|
||||
* MTSSources as it will not reset counters when switching sources.
|
||||
*/
|
||||
public class ContinuityFixingMTSSource extends AbstractMTSSource {
|
||||
private final ContinuityFixer continuityFixer = new ContinuityFixer();
|
||||
private final MTSSource source;
|
||||
|
||||
public ContinuityFixingMTSSource(MTSSource source) {
|
||||
this.source = source;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MTSPacket nextPacketInternal() throws Exception {
|
||||
MTSPacket packet = source.nextPacket();
|
||||
if (packet != null) {
|
||||
continuityFixer.fixContinuity(packet);
|
||||
}
|
||||
return packet;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void closeInternal() throws Exception {
|
||||
source.close();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,30 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
/**
|
||||
* Decorate a source with a declared bitrate.
|
||||
*/
|
||||
public class FixedBitrateMTSSource extends AbstractMTSSource {
|
||||
private final MTSSource source;
|
||||
private final long bitrate;
|
||||
|
||||
public FixedBitrateMTSSource(MTSSource source, long bitrate) {
|
||||
this.source = source;
|
||||
this.bitrate = bitrate;
|
||||
}
|
||||
|
||||
public long getBitrate() {
|
||||
return bitrate;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MTSPacket nextPacketInternal() throws Exception {
|
||||
return source.nextPacket();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void closeInternal() throws Exception {
|
||||
source.close();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,63 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import org.taktik.mpegts.Constants;
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
|
||||
public class InputStreamMTSSource extends AbstractMTSSource {
|
||||
|
||||
private InputStream inputStream;
|
||||
|
||||
private InputStreamMTSSource(InputStream inputStream) {
|
||||
this.inputStream = inputStream;
|
||||
}
|
||||
|
||||
public static InputStreamMTSSourceBuilder builder() {
|
||||
return new InputStreamMTSSourceBuilder();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MTSPacket nextPacketInternal() throws IOException {
|
||||
byte[] packetData = new byte[Constants.MPEGTS_PACKET_SIZE];
|
||||
int bytesRead = 0;
|
||||
while(bytesRead < Constants.MPEGTS_PACKET_SIZE) {
|
||||
int bytesLeft = Constants.MPEGTS_PACKET_SIZE - bytesRead;
|
||||
int length = inputStream.read(packetData, bytesRead, bytesLeft);
|
||||
bytesRead += length;
|
||||
if(length == -1) {
|
||||
// no more bytes available
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// Parse the packet
|
||||
return new MTSPacket(ByteBuffer.wrap(packetData));
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void closeInternal() throws Exception {
|
||||
inputStream.close();
|
||||
}
|
||||
|
||||
public static class InputStreamMTSSourceBuilder {
|
||||
private InputStream inputStream;
|
||||
|
||||
private InputStreamMTSSourceBuilder() {
|
||||
}
|
||||
|
||||
public InputStreamMTSSourceBuilder setInputStream(InputStream inputStream) {
|
||||
this.inputStream = inputStream;
|
||||
return this;
|
||||
}
|
||||
|
||||
public InputStreamMTSSource build() {
|
||||
Preconditions.checkNotNull(inputStream, "InputStream cannot be null");
|
||||
return new InputStreamMTSSource(inputStream);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,74 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
import static com.google.common.base.Preconditions.checkArgument;
|
||||
import static com.google.common.base.Preconditions.checkNotNull;
|
||||
|
||||
public class LoopingMTSSource extends AbstractMTSSource{
|
||||
private ResettableMTSSource source;
|
||||
private Integer maxLoops;
|
||||
private long currentLoop;
|
||||
|
||||
public LoopingMTSSource(ResettableMTSSource source, Integer maxLoops) {
|
||||
this.source = source;
|
||||
this.maxLoops = maxLoops;
|
||||
currentLoop = 1;
|
||||
}
|
||||
|
||||
public static LoopingMTSSourceBuilder builder() {
|
||||
return new LoopingMTSSourceBuilder();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MTSPacket nextPacketInternal() throws Exception {
|
||||
MTSPacket packet = source.nextPacket();
|
||||
if (packet == null) {
|
||||
currentLoop++;
|
||||
if (maxLoops == null || (currentLoop <= maxLoops)) {
|
||||
source.reset();
|
||||
packet = source.nextPacket();
|
||||
}
|
||||
}
|
||||
return packet;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void closeInternal() throws Exception {
|
||||
source.close();
|
||||
}
|
||||
|
||||
public static class LoopingMTSSourceBuilder {
|
||||
private ResettableMTSSource source;
|
||||
private boolean fixContinuity;
|
||||
private Integer maxLoops;
|
||||
|
||||
private LoopingMTSSourceBuilder() {
|
||||
}
|
||||
|
||||
public MTSSource build() {
|
||||
checkNotNull(source);
|
||||
checkArgument(maxLoops == null || maxLoops > 0);
|
||||
MTSSource result = new LoopingMTSSource(source, maxLoops);
|
||||
if (fixContinuity) {
|
||||
return new ContinuityFixingMTSSource(result);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public LoopingMTSSourceBuilder setSource(ResettableMTSSource source) {
|
||||
this.source = source;
|
||||
return this;
|
||||
}
|
||||
|
||||
public LoopingMTSSourceBuilder setFixContinuity(boolean fixContinuity) {
|
||||
this.fixContinuity = fixContinuity;
|
||||
return this;
|
||||
}
|
||||
|
||||
public LoopingMTSSourceBuilder setMaxLoops(Integer maxLoops) {
|
||||
this.maxLoops = maxLoops;
|
||||
return this;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,7 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
public interface MTSSource extends AutoCloseable {
|
||||
MTSPacket nextPacket() throws Exception;
|
||||
}
|
||||
|
|
@ -1,80 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
|
||||
import com.google.common.io.ByteSource;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.nio.channels.ByteChannel;
|
||||
import java.nio.channels.FileChannel;
|
||||
import java.nio.channels.SeekableByteChannel;
|
||||
|
||||
public class MTSSources {
|
||||
public static MTSSource fromSources(MTSSource... sources) {
|
||||
return fromSources(1, false, sources);
|
||||
}
|
||||
|
||||
public static MTSSource fromSources(int loops, MTSSource... sources) {
|
||||
return fromSources(loops, false, sources);
|
||||
}
|
||||
|
||||
public static MTSSource fromSources(int loops, boolean fixContinuity, MTSSource... sources) {
|
||||
return ConcatenatingMTSSource.builder()
|
||||
.setFixContinuity(fixContinuity)
|
||||
.setSources(sources)
|
||||
.loops(loops)
|
||||
.build();
|
||||
}
|
||||
|
||||
public static MTSSource from(ByteChannel channel) throws IOException {
|
||||
return ByteChannelMTSSource.builder()
|
||||
.setByteChannel(channel)
|
||||
.build();
|
||||
}
|
||||
|
||||
public static ResettableMTSSource from(SeekableByteChannel channel) throws IOException {
|
||||
return SeekableByteChannelMTSSource.builder()
|
||||
.setByteChannel(channel)
|
||||
.build();
|
||||
}
|
||||
|
||||
public static ResettableMTSSource from(File file) throws IOException {
|
||||
return SeekableByteChannelMTSSource.builder()
|
||||
.setByteChannel(FileChannel.open(file.toPath()))
|
||||
.build();
|
||||
}
|
||||
|
||||
public static ResettableMTSSource from(ByteSource byteSource) throws IOException {
|
||||
return ByteSourceMTSSource.builder()
|
||||
.setByteSource(byteSource)
|
||||
.build();
|
||||
}
|
||||
|
||||
public static MTSSource from(InputStream inputStream) throws IOException {
|
||||
return InputStreamMTSSource.builder()
|
||||
.setInputStream(inputStream)
|
||||
.build();
|
||||
}
|
||||
|
||||
public static MTSSource loop(ResettableMTSSource source) {
|
||||
return LoopingMTSSource.builder()
|
||||
.setSource(source)
|
||||
.build();
|
||||
}
|
||||
|
||||
public static MTSSource loop(ResettableMTSSource source, int maxLoops) {
|
||||
return LoopingMTSSource.builder()
|
||||
.setSource(source)
|
||||
.setMaxLoops(maxLoops)
|
||||
.build();
|
||||
}
|
||||
|
||||
public static MTSSource multiplexing(MTSSource... sources) {
|
||||
return MultiplexingMTSSource.builder()
|
||||
.setSources(sources)
|
||||
.setFixContinuity(true)
|
||||
.build();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -1,100 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
import com.google.common.collect.Lists;
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* An MTSSource using a simple round-robin packet multiplexing strategy.
|
||||
*/
|
||||
public class MultiplexingMTSSource extends AbstractMTSSource {
|
||||
private final boolean fixContinuity;
|
||||
private final List<MTSSource> sources;
|
||||
|
||||
private ContinuityFixer continuityFixer;
|
||||
private int nextSource = 0;
|
||||
|
||||
public MultiplexingMTSSource(boolean fixContinuity, Collection<MTSSource> sources) {
|
||||
Preconditions.checkArgument(sources.size() > 0, "Must provide at least contain one source");
|
||||
this.fixContinuity = fixContinuity;
|
||||
this.sources = Lists.newArrayList(sources);
|
||||
if (fixContinuity) {
|
||||
continuityFixer = new ContinuityFixer();
|
||||
}
|
||||
}
|
||||
|
||||
public static MultiplexingMTSSourceBuilder builder() {
|
||||
return new MultiplexingMTSSourceBuilder();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MTSPacket nextPacketInternal() throws Exception {
|
||||
MTSPacket packet = sources.get(nextSource).nextPacket();
|
||||
if (packet != null) {
|
||||
if (fixContinuity) {
|
||||
continuityFixer.fixContinuity(packet);
|
||||
}
|
||||
return packet;
|
||||
} else {
|
||||
// FIXME: infinite loop
|
||||
nextSource();
|
||||
return nextPacket();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected synchronized void closeInternal() throws Exception {
|
||||
for (MTSSource source : sources) {
|
||||
source.close();
|
||||
}
|
||||
}
|
||||
|
||||
private synchronized void nextSource() {
|
||||
nextSource++;
|
||||
if (nextSource == sources.size()) {
|
||||
nextSource = 0;
|
||||
}
|
||||
}
|
||||
|
||||
public static class MultiplexingMTSSourceBuilder {
|
||||
boolean fixContinuity = false;
|
||||
private List<MTSSource> sources = Lists.newArrayList();
|
||||
|
||||
public MultiplexingMTSSourceBuilder addSource(MTSSource source) {
|
||||
sources.add(source);
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiplexingMTSSourceBuilder addSources(MTSSource... sources) {
|
||||
this.sources.addAll(Lists.newArrayList(sources));
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiplexingMTSSourceBuilder addSources(Collection<MTSSource> sources) {
|
||||
this.sources.addAll(sources);
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiplexingMTSSourceBuilder setSources(MTSSource... sources) {
|
||||
this.sources = Lists.newArrayList(sources);
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiplexingMTSSourceBuilder setSources(Collection<MTSSource> sources) {
|
||||
this.sources = Lists.newArrayList(sources);
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiplexingMTSSourceBuilder setFixContinuity(boolean fixContinuity) {
|
||||
this.fixContinuity = fixContinuity;
|
||||
return this;
|
||||
}
|
||||
|
||||
public MultiplexingMTSSource build() {
|
||||
return new MultiplexingMTSSource(fixContinuity, sources);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,34 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import org.taktik.mpegts.Constants;
|
||||
import org.taktik.mpegts.MTSPacket;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.Arrays;
|
||||
|
||||
public class NullPacketSource extends AbstractMTSSource {
|
||||
|
||||
public NullPacketSource() {
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MTSPacket nextPacketInternal() throws Exception {
|
||||
byte[] buf = new byte[Constants.MPEGTS_PACKET_SIZE];
|
||||
|
||||
// payload (null bytes)
|
||||
Arrays.fill(buf, (byte) 0xff);
|
||||
|
||||
// header
|
||||
buf[0] = 0x47; // sync byte
|
||||
buf[1] = 0x1f; // PID high
|
||||
buf[2] = (byte) 0xff; // PID low
|
||||
buf[3] = 0x10; // adaptation control and continuity
|
||||
|
||||
return new MTSPacket(ByteBuffer.wrap(buf));
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void closeInternal() throws Exception {
|
||||
// does nothing
|
||||
}
|
||||
}
|
||||
|
|
@ -1,5 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
public interface ResettableMTSSource extends MTSSource {
|
||||
void reset() throws Exception;
|
||||
}
|
||||
|
|
@ -1,39 +0,0 @@
|
|||
package org.taktik.mpegts.sources;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.channels.SeekableByteChannel;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
|
||||
public class SeekableByteChannelMTSSource extends AbstractByteChannelMTSSource<SeekableByteChannel> implements ResettableMTSSource {
|
||||
|
||||
private SeekableByteChannelMTSSource(SeekableByteChannel byteChannel) throws IOException {
|
||||
super(byteChannel);
|
||||
}
|
||||
|
||||
public static SeekableByteChannelMTSSourceBuilder builder() {
|
||||
return new SeekableByteChannelMTSSourceBuilder();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void reset() throws IOException {
|
||||
byteChannel.position(0);
|
||||
fillBuffer();
|
||||
}
|
||||
|
||||
public static class SeekableByteChannelMTSSourceBuilder {
|
||||
private SeekableByteChannel byteChannel;
|
||||
|
||||
private SeekableByteChannelMTSSourceBuilder(){}
|
||||
|
||||
public SeekableByteChannelMTSSourceBuilder setByteChannel(SeekableByteChannel byteChannel) {
|
||||
this.byteChannel = byteChannel;
|
||||
return this;
|
||||
}
|
||||
|
||||
public SeekableByteChannelMTSSource build() throws IOException {
|
||||
Preconditions.checkNotNull(byteChannel, "byteChannel cannot be null");
|
||||
return new SeekableByteChannelMTSSource(byteChannel);
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue