解决2019协议设备播放问题

18045010223

7 天以前 afe371d39a054b2f2a9e5875b945584eec8a8141

解决2019协议设备播放问题

 ffmpeg.zip

Binary files differ

 src/main/java/cn/org/hentai/jtt1078/app/Test.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,19 @@
package cn.org.hentai.jtt1078.app;

public class Test {
    public static void main(String[] args) {
        byte[] buffer = {
                48, 49, 99, 100, -127, 98, 0, 0, 0, 0, 0, 0, 1, 51, 7, 68, 67, -128, 1, 1, 0, 0, 1, -104, 53, -33, -2, 57, 0, 0, 0, 100, 3, -74, 0, 0, 0, 1, 103, 77, 0, 10, -106, 84, 2, -128, 45, -120, 0, 0, 0, 1, 104, -18, 60, -128, 0, 0, 0, 1, 101, -120, -128, 64, 5, -1, -96, -32, 64, 111, 81};

       /* byte[] buffer = {
                48, 49, 99, 100, -127,
                98, 0, 2, 2, 2,
                50,80, 2, 32, 1,
                3,0, 0, 1, -104, 55, -112, 118, 111, 0, 0, 0, 0, 3, -74, 114, 66, 70, -114, -16, -109, 45, 86, -20,};*/
        int position = 28;
        int h = buffer[position] & 0xff;
        int l = buffer[position + 1] & 0xff;
        int result = ((h << 8) | l) & 0xffff;
        System.out.println("result = " + result);
    }
}

 src/main/java/cn/org/hentai/jtt1078/server/Jtt1078Decoder.java

@@ -4,48 +4,306 @@
import cn.org.hentai.jtt1078.util.ByteUtils;
import cn.org.hentai.jtt1078.util.Packet;

/**
 * Created by matrixy on 2019/4/9.
 */
public class Jtt1078Decoder
{
    //4096
    ByteHolder buffer = new ByteHolder(4096);
import java.time.Instant;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;

    public void write(byte[] block)
    {
/**
 * JTT1078 Protocol Decoder
 */

public class Jtt1078Decoder {
    private ByteHolder buffer = new ByteHolder(4096);
    private int simLen = 0;
    private String sim = "";
    private byte channel = 0;
    private static final byte[] HEAD1078 = {0x30, 0x31, 0x63, 0x64};

    public void write(byte[] block) {
        buffer.write(block);
    }

    public void write(byte[] block, int startIndex, int length)
    {
    public void write(byte[] block, int startIndex, int length) {
        byte[] buff = new byte[length];
        System.arraycopy(block, startIndex, buff, 0, length);
        write(buff);
    }

    public Packet decode()
    {
        if (this.buffer.size() < 30) return null;
    public Packet decode() {
        if (buffer.size() < 4) return null;

        if ((buffer.getInt(0) & 0x7fffffff) != 0x30316364)
        {
            String header = ByteUtils.toString(buffer.array(30));
            throw new RuntimeException("invalid protocol header: " + header);
        // Check header
        if (!Arrays.equals(buffer.array(4), HEAD1078)) {
//            log.warn("Invalid protocol header, expected:30316364, actual:{}",
//                    ByteUtils.toHexString(buffer.array(4)));
            buffer.clear();
            return null;
        }

        int lengthOffset = 28;
        int dataType = (this.buffer.get(15) >> 4) & 0x0f;
        // é€ä¼ æ•°æ®ç±»åž‹ï¼š0100，没有后面的时间以及Last I Frame Interval和Last Frame Interval字段
        if (dataType == 0x04) lengthOffset = 28 - 8 - 2 - 2;
        else if (dataType == 0x03) lengthOffset = 28 - 4;
        int bodyLength = this.buffer.getShort(lengthOffset);
        // Try to determine protocol version if not set
        if (simLen == 0) {
            // Try version 16 (2013 protocol)
            Packet packet = decodePacket(16);
            if (packet != null) {
                Analyze analyze = parse16(packet.getBytes());
                sim = analyze.getSim();
                channel = analyze.getCh();
                simLen = 6;
                System.out.printf("%s-%d device started output%n", sim, channel);
                return packet;
            }

        int packetLength = bodyLength + lengthOffset + 2;
            // Try version 19 (2019 protocol)
            packet = decodePacket(19);
            if (packet != null) {
                Analyze analyze = parse19(packet.getBytes());
                sim = analyze.getSim();
                channel = analyze.getCh();
                simLen = 10;
                System.out.printf("%s-%d device started output%n", sim, channel);
                return packet;
            }

        if (this.buffer.size() < packetLength) return null;
        byte[] block = new byte[packetLength];
        this.buffer.sliceInto(block, packetLength);
        return Packet.create(block);
            return null;
        } else {
            return decodePacket(simLen == 6 ? 16 : 19);
        }
    }

    private Packet decodePacket(int version) {
        int dataTypeIndex = version == 16 ? 15 : 19;
        if (buffer.size() < dataTypeIndex + 1) return null;

        byte dataType = (byte) ((buffer.get(dataTypeIndex) >> 4) & 0x0F);
        int dataLenIndex;

        switch (dataType) {
            case 3:
                dataLenIndex = dataTypeIndex + 9;
                break;
            case 4:
                dataLenIndex = dataTypeIndex + 1;
                break;
            default:
                dataLenIndex = dataTypeIndex + 13;
        }
        //log.info("dataType:"+dataType);
        if (buffer.size() < dataLenIndex + 2) return null;
        int dataLen = buffer.getShort(dataLenIndex) & 0xFFFF;
        int packetLength = dataLenIndex + 2 + dataLen;

        if (buffer.size() < packetLength) return null;

        // Check next packet header
        if (buffer.size() >= packetLength + 4) {
            byte[] nextHeader = new byte[4];
            for (int i = 0; i < 4; i++) {
                nextHeader[i] = buffer.get(packetLength + i);
            }
            if (!Arrays.equals(nextHeader, HEAD1078)) {
               // log.warn("Invalid data type: {}", dataType);
                buffer.clear();
                return null;
            }
        }

        byte[] packetData = new byte[packetLength];
        buffer.sliceInto(packetData, packetLength);
        // printPacketInfo(packetData, version);
        return Packet.create(packetData);
    }

    private Analyze parse16(byte[] data) {
        Analyze analyze = new Analyze();
        analyze.setPt((byte) (data[5] & 0x7F));
        analyze.setSn((short) (((data[6] & 0xFF) << 8) | (data[7] & 0xFF)));

        StringBuilder simBuilder = new StringBuilder();
        for (int i = 0; i < 6; i++) {
            simBuilder.append(nextBcd(data, 8 + i));
        }
        analyze.setSim(simBuilder.toString());

        analyze.setCh(data[14]);
        analyze.setDt((byte) ((data[15] >> 4) & 0x0F));
        analyze.setFi((byte) (data[15] & 0x0F));

        int dataLenIndex;
        switch (analyze.getDt()) {
            case 3:
                analyze.setTimestamp(setTimestamp(data, 16));
                dataLenIndex = 24;
                break;
            case 4:
                dataLenIndex = 16;
                break;
            default:
                analyze.setTimestamp(setTimestamp(data, 16));
                analyze.setLastIInterval((short) (((data[24] & 0xFF) << 8) | (data[25] & 0xFF)));
                analyze.setLastInterval((short) (((data[26] & 0xFF) << 8) | (data[27] & 0xFF)));
                dataLenIndex = 28;
        }

        analyze.setDataLen((short) (((data[dataLenIndex] & 0xFF) << 8) | (data[dataLenIndex + 1] & 0xFF)));
        byte[] payload = new byte[analyze.getDataLen()];
        System.arraycopy(data, dataLenIndex + 2, payload, 0, payload.length);
        analyze.setData(payload);

        return analyze;
    }

    private Analyze parse19(byte[] data) {
        Analyze analyze = new Analyze();
        analyze.setPt((byte) (data[5] & 0x7F));
        analyze.setSn((short) (((data[6] & 0xFF) << 8) | (data[7] & 0xFF)));

        StringBuilder simBuilder = new StringBuilder();
        for (int i = 0; i < 10; i++) {
            simBuilder.append(nextBcd(data, 8 + i));
        }
        analyze.setSim(simBuilder.toString());

        analyze.setCh(data[18]);
        analyze.setDt((byte) ((data[19] >> 4) & 0x0F));
        analyze.setFi((byte) (data[19] & 0x0F));

        int dataLenIndex;
        switch (analyze.getDt()) {
            case 3:
                analyze.setTimestamp(setTimestamp(data, 20));
                dataLenIndex = 28;
                break;
            case 4:
                dataLenIndex = 20;
                break;
            default:
                analyze.setTimestamp(setTimestamp(data, 20));
                analyze.setLastIInterval((short) (((data[28] & 0xFF) << 8) | (data[29] & 0xFF)));
                analyze.setLastInterval((short) (((data[30] & 0xFF) << 8) | (data[31] & 0xFF)));
                dataLenIndex = 32;
        }

        analyze.setDataLen((short) (((data[dataLenIndex] & 0xFF) << 8) | (data[dataLenIndex + 1] & 0xFF)));
        byte[] payload = new byte[analyze.getDataLen()];
        System.arraycopy(data, dataLenIndex + 2, payload, 0, payload.length);
        analyze.setData(payload);

        return analyze;
    }

    private long setTimestamp(byte[] data, int offset) {
        long timestamp = 0;
        for (int i = 0; i < 8; i++) {
            timestamp = (timestamp << 8) | (data[offset + i] & 0xFF);
        }
        return timestamp;
    }

    private String nextBcd(byte[] data, int offset) {
        byte val = data[offset];
        int ch1 = (val >> 4) & 0x0F;
        int ch2 = val & 0x0F;
        return String.format("%d%d", ch1, ch2);
    }

    private void printPacketInfo(byte[] packet, int version) {
        System.out.println("\n=== JTT1078 " + (version == 16 ? "2013" : "2019") + " Protocol Packet ===");
        System.out.println("Header: " + ByteUtils.toHexString(packet, 0, 4));

        Analyze analyze = version == 16 ? parse16(packet) : parse19(packet);
        System.out.println("Payload Type: " + analyze.getPt());
        System.out.println("Sequence Number: " + analyze.getSn());
        System.out.println("SIM: " + analyze.getSim());
        System.out.println("Channel: " + analyze.getCh());
        System.out.println("Data Type: " + getDataTypeDescription(analyze.getDt()));
        System.out.println("Frame Info: " + getFrameInfoDescription(analyze.getFi()));

        if (analyze.getTimestamp() != 0) {
            Instant instant = Instant.ofEpochMilli(analyze.getTimestamp());
            ZonedDateTime beijingTime = instant.atZone(ZoneOffset.ofHours(8));
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            System.out.println("Timestamp: " + formatter.format(beijingTime));
        }

        if (analyze.getLastIInterval() != 0) {
            System.out.println("Last I Frame Interval: " + analyze.getLastIInterval());
        }
        if (analyze.getLastInterval() != 0) {
            System.out.println("Last Frame Interval: " + analyze.getLastInterval());
        }

        System.out.println("Data Length: " + analyze.getDataLen());
        System.out.println("=================================\n");
    }

    private String getDataTypeDescription(int dataType) {
        switch (dataType) {
            case 0: return "Video I Frame";
            case 1: return "Video P Frame";
            case 2: return "Video B Frame";
            case 3: return "Audio Frame";
            case 4: return "Transparent Data";
            default: return "Unknown (" + dataType + ")";
        }
    }

    private String getFrameInfoDescription(int fi) {
        switch (fi) {
            case 0: return "Atomic Packet";
            case 1: return "First Packet";
            case 2: return "Last Packet";
            case 3: return "Middle Packet";
            default: return "Unknown (" + fi + ")";
        }
    }
}

class Analyze {
    private byte pt;
    private short sn;
    private String sim;
    private byte ch;
    private byte dt;
    private byte fi;
    private long timestamp;
    private short lastIInterval;
    private short lastInterval;
    private short dataLen;
    private byte[] data;

    // Getters and setters
    public byte getPt() { return pt; }
    public void setPt(byte pt) { this.pt = pt; }

    public short getSn() { return sn; }
    public void setSn(short sn) { this.sn = sn; }

    public String getSim() { return sim; }
    public void setSim(String sim) { this.sim = sim; }

    public byte getCh() { return ch; }
    public void setCh(byte ch) { this.ch = ch; }

    public byte getDt() { return dt; }
    public void setDt(byte dt) { this.dt = dt; }

    public byte getFi() { return fi; }
    public void setFi(byte fi) { this.fi = fi; }

    public long getTimestamp() { return timestamp; }
    public void setTimestamp(long timestamp) { this.timestamp = timestamp; }

    public short getLastIInterval() { return lastIInterval; }
    public void setLastIInterval(short lastIInterval) { this.lastIInterval = lastIInterval; }

    public short getLastInterval() { return lastInterval; }
    public void setLastInterval(short lastInterval) { this.lastInterval = lastInterval; }

    public short getDataLen() { return dataLen; }
    public void setDataLen(short dataLen) { this.dataLen = dataLen; }

    public byte[] getData() { return data; }
    public void setData(byte[] data) { this.data = data; }
}

 src/main/java/cn/org/hentai/jtt1078/server/Jtt1078Handler.java

@@ -24,18 +24,28 @@
    static Logger logger = LoggerFactory.getLogger(Jtt1078Handler.class);
    private static final AttributeKey<Session> SESSION_KEY = AttributeKey.valueOf("session-key");

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Packet packet) throws Exception
    {
    // @Override
    protected void channelRead011(ChannelHandlerContext ctx, Packet packet) throws Exception {
        io.netty.channel.Channel nettyChannel = ctx.channel();

        // æ£€æŸ¥åè®®ç‰ˆæœ¬
        packet.seek(14);
        byte b14 = packet.nextByte();
        byte b15 = packet.nextByte();
        boolean is2019Protocol = (b14 & 0xFF) == 0x00 && (b15 & 0xFF) == 0x00;
        int simLength = is2019Protocol ? 10 : 6;

        // è¯»å–SIM卡和通道号
        packet.seek(8);
        String sim = packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD() + packet.nextBCD();
        StringBuilder simBuilder = new StringBuilder();
        for(int i=0; i<simLength; i++) {
            simBuilder.append(packet.nextBCD());
        }
        String sim = simBuilder.toString();
        int channel = packet.nextByte() & 0xff;
        String tag = sim + "-" + channel;

        if (SessionManager.contains(nettyChannel, "tag") == false)
        {
        if (!SessionManager.contains(nettyChannel, "tag")) {
            Channel chl = PublishManager.getInstance().open(tag);
            SessionManager.set(nettyChannel, "tag", tag);
            logger.info("start publishing: {} -> {}-{}", Long.toHexString(chl.hashCode() & 0xffffffffL), sim, channel);
@@ -43,36 +53,131 @@

        Integer sequence = SessionManager.get(nettyChannel, "video-sequence");
        if (sequence == null) sequence = 0;
        // 1. åšå¥½åºå·
        // 2. éŸ³é¢‘需要转码后提供订阅
        int lengthOffset = 28;
        int dataType = (packet.seek(15).nextByte() >> 4) & 0x0f;
        int pkType = packet.seek(15).nextByte() & 0x0f;
        // é€ä¼ æ•°æ®ç±»åž‹ï¼š0100，没有后面的时间以及Last I Frame Interval和Last Frame Interval字段
        if (dataType == 0x04) lengthOffset = 28 - 8 - 2 - 2;
        else if (dataType == 0x03) lengthOffset = 28 - 4;

        // åŠ¨æ€èŽ·å–数据类型和包类型
        int dataTypePos = is2019Protocol ? 19 : 15;
        packet.seek(dataTypePos);
        int dataType = (packet.nextByte() >> 4) & 0x0f;
        int pkType = packet.nextByte() & 0x0f;

        // è®¡ç®—数据偏移量
        int baseOffset = is2019Protocol ? 32 : 28;
        int lengthOffset = baseOffset;
        if (dataType == 0x04) {
            lengthOffset = baseOffset - 8 - 2 - 2;
        } else if (dataType == 0x03) {
            lengthOffset = baseOffset - 4;
        }

        int pt = packet.seek(5).nextByte() & 0x7f;
        int timestampOffset = is2019Protocol ? 20 : 16;

        if (dataType == 0x00 || dataType == 0x01 || dataType == 0x02)
        {
            // ç¢°åˆ°ç»“束标记时，序号+1
            if (pkType == 0 || pkType == 2)
            {
        if (dataType == 0x00 || dataType == 0x01 || dataType == 0x02) {
            if (pkType == 0 || pkType == 2) {
                sequence += 1;
                SessionManager.set(nettyChannel, "video-sequence", sequence);
            }
            long timestamp = packet.seek(16).nextLong();
            PublishManager.getInstance().publishVideo(tag, sequence, timestamp, pt, packet.seek(lengthOffset + 2).nextBytes());
            long timestamp = packet.seek(timestampOffset).nextLong();
            byte[] videoData = packet.seek(lengthOffset + 2).nextBytes();
            logger.debug("Publishing video data - size: {}, seq: {}, ts: {}", videoData.length, sequence, timestamp);
            PublishManager.getInstance().publishVideo(tag, sequence, timestamp, pt, videoData);
        }
        else if (dataType == 0x03)
        {
            long timestamp = packet.seek(16).nextLong();
            byte[] data = packet.seek(lengthOffset + 2).nextBytes();
            PublishManager.getInstance().publishAudio(tag, sequence, timestamp, pt, data);
        else if (dataType == 0x03) {
            long timestamp = packet.seek(timestampOffset).nextLong();
            byte[] audioData = packet.seek(lengthOffset + 2).nextBytes();
            logger.debug("Publishing audio data - size: {}, seq: {}, ts: {}", audioData.length, sequence, timestamp);
            PublishManager.getInstance().publishAudio(tag, sequence, timestamp, pt, audioData);
        }
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Packet packet) throws Exception {
        io.netty.channel.Channel nettyChannel = ctx.channel();

        // 1. åè®®ç‰ˆæœ¬æ£€æµ‹
        boolean is2019Protocol = detectProtocolVersion(packet);
        int simLength = is2019Protocol ? 10 : 6;

        // 2. è¯»å–SIM卡号和通道号
        packet.seek(8);
        StringBuilder simBuilder = new StringBuilder();
        for (int i = 0; i < simLength; i++) {
            simBuilder.append(packet.nextBCD());
        }
        String sim = simBuilder.toString();

        int channelPos = is2019Protocol ? 18 : 14;
        packet.seek(channelPos);
        int channel = packet.nextByte() & 0xff;
        String tag = sim + "-" + channel;

        // 3. ä¼šè¯ç®¡ç†
        if (!SessionManager.contains(nettyChannel, "tag")) {
            Channel chl = PublishManager.getInstance().open(tag);
            SessionManager.set(nettyChannel, "tag", tag);
            logger.info("start publishing: {} -> {}-{} (Protocol: {})",
                    Long.toHexString(chl.hashCode() & 0xffffffffL),
                    sim, channel,
                    is2019Protocol ? "2019" : "2016");
        }

        // 4. æ•°æ®å¤„理
        Integer sequence = SessionManager.get(nettyChannel, "video-sequence");
        if (sequence == null) sequence = 0;

        int dataTypePos = is2019Protocol ? 19 : 15;
        packet.seek(dataTypePos);
        int dataType = (packet.nextByte() >> 4) & 0x0f;
        int pkType = packet.nextByte() & 0x0f;

        int baseOffset = is2019Protocol ? 32 : 28;
        int lengthOffset = baseOffset;
        if (dataType == 0x04) lengthOffset = baseOffset - 8 - 2 - 2;
        else if (dataType == 0x03) lengthOffset = baseOffset - 4;

        int pt = packet.seek(5).nextByte() & 0x7f;
        int timestampOffset = is2019Protocol ? 20 : 16;

        if (dataType == 0x00 || dataType == 0x01 || dataType == 0x02) {
            if (pkType == 0 || pkType == 2) {
                sequence += 1;
                SessionManager.set(nettyChannel, "video-sequence", sequence);
            }
            long timestamp = packet.seek(timestampOffset).nextLong();
            byte[] videoData = packet.seek(lengthOffset + 2).nextBytes();
            PublishManager.getInstance().publishVideo(tag, sequence, timestamp, pt, videoData);
        }
        else if (dataType == 0x03) {
            long timestamp = packet.seek(timestampOffset).nextLong();
            byte[] audioData = packet.seek(lengthOffset + 2).nextBytes();
            PublishManager.getInstance().publishAudio(tag, sequence, timestamp, pt, audioData);
        }
    }

    private boolean detectProtocolVersion(Packet packet) {
        // æ–¹æ³•1：检查6字节SIM卡号后的填充
        packet.seek(8 + 6); // 2016协议SIM卡结束位置
        byte b14 = packet.nextByte();
        byte b15 = packet.nextByte();

        // å¦‚果是2019协议，这里应该是0x00填充
        if ((b14 & 0xFF) == 0x00 && (b15 & 0xFF) == 0x00) {
            return true;
        }

        // æ–¹æ³•2：检查数据类型位置的有效性
        packet.seek(15);
        byte dtByte = packet.nextByte();
        int dataType = (dtByte >> 4) & 0x0F;
        int frameType = dtByte & 0x0F;

        // å¦‚果数据类型或帧类型无效，可能是2019协议
        if (dataType > 4 || frameType > 3) {
            return true;
        }

        // é»˜è®¤è¿”回2016协议
        return false;
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception
    {

 src/main/java/cn/org/hentai/jtt1078/server/Jtt1078MessageDecoder.java

@@ -28,7 +28,7 @@
        {
            int l = i < k - 1 ? 512 : length - (i * 512);
            in.readBytes(block, 0, l);

            //System.out.println("block:"+block.length);
            decoder.write(block, 0, l);

            while (true)

 src/main/java/cn/org/hentai/jtt1078/util/ByteHolder.java

@@ -13,6 +13,8 @@

    public ByteHolder(int bufferSize)
    {
        System.out.print("ByteHolder(bufferSize:"+bufferSize+")");
    	
        this.buffer = new byte[bufferSize];
    }

@@ -56,6 +58,7 @@

    public void sliceInto(byte[] dest, int length)
    {
        //System.out.println();
        System.arraycopy(this.buffer, 0, dest, 0, length);
        // å¾€å‰æŒªlength个位
        System.arraycopy(this.buffer, length, this.buffer, 0, this.size - length);

 src/main/java/cn/org/hentai/jtt1078/util/ByteUtils.java

@@ -5,6 +5,14 @@
 */
public final class ByteUtils
{
    public static String toHexString(byte[] bytes, int offset, int length) {
        StringBuilder sb = new StringBuilder(length * 3);
        for (int i = offset; i < offset + length && i < bytes.length; i++) {
            sb.append(String.format("%02X ", bytes[i]));
        }
        return sb.toString().trim();
    }

    public static byte[] parse(String hexString)
    {
        String[] hexes = hexString.split(" ");

 src/main/resources/app.properties

@@ -7,6 +7,7 @@

# ffmpeg可执行文件路径，可以留空
ffmpeg.path = E:/jtt1078/ffmpeg/bin/ffmpeg.exe
//ffmpeg.path = D:/vehicle/ffmpeg/bin/ffmpeg.exe

# é…ç½®rtmp地址将在终端发送RTP消息包时，额外的向RTMP服务器推流
# TAG的形式就是SIM-CHANNEL，如13800138999-2
@@ -15,4 +16,4 @@
#rtmp.url = rtmp://47.104.204.210/live/{TAG}

# è®¾ç½®ä¸ºon时，控制台将输出ffmpeg的输出
debug.mode = off
debug.mode = on