public interface IRtmpConnection {

    //base function
    @WorkerThread
    void connect(String url);

    void publishAudioData(byte[] data, int type);

    void publishVideoData(byte[] data, int type);

    void stop();


    //setting
    void setConnectListener(RtmpConnectListener listener);

    void setSendQueue(ISendQueue sendQueue);

    void setVideoParams(int width, int height);

    void setAudioParams(int sampleRate, int sampleSize, boolean isStereo);

    enum State {
        INIT,
        HANDSHAKE,
        CONNECTING,
        CREATE_STREAM,
        PUBLISHING,
        LIVING
    }
} public class Frame {
    public static final int FRAME_TYPE_AUDIO = 1;
    public static final int FRAME_TYPE_KEY_FRAME = 2;
    public static final int FRAME_TYPE_INTER_FRAME = 3;
    public static final int FRAME_TYPE_CONFIGURATION = 4;

    public Chunk data;
    public int packetType;
    public int frameType;

    public Frame(Chunk data, int packetType, int frameType) {
        this.data = data;
        this.packetType = packetType;
        this.frameType = frameType;
    }
} public class AmfMap extends AmfObject {
    @Override
    public void writeTo(OutputStream out) throws IOException {
        // Begin the map/object/array/whatever exactly this is
        out.write(AmfType.MAP.getValue());

        // Write the "array size"
        Util.writeUnsignedInt32(out, properties.size());

        // Write key/value pairs in this object
        for (Map.Entry<String, AmfData> entry : properties.entrySet()) {
            // The key must be a STRING type, and thus the "type-definition" byte is implied (not included in message)
            AmfString.writeStringTo(out, entry.getKey(), true);
            entry.getValue().writeTo(out);
        }

        // End the object
        out.write(OBJECT_END_MARKER);
    }

    @Override
    public void readFrom(InputStream in) throws IOException {
        // Skip data type byte (we assume it's already read)
        int length = Util.readUnsignedInt32(in); // Seems this is always 0
        super.readFrom(in);
        size += 4; // Add the bytes read for parsing the array size (length)
    }

    @Override
    public int getSize() {
        if (size == -1) {
            size = super.getSize();
            size += 4; // array length bytes
        }
        return size;
    }

    @Override
    public byte[] getBytes() {
        int size = getSize();
        ByteBuffer byteBuffer = ByteBuffer.allocate(size);
        byteBuffer.put(AmfType.OBJECT.getValue());
        byteBuffer.putInt(properties.size());
        // Write key/value pairs in this object
        for (Map.Entry<String, AmfData> entry : properties.entrySet()) {
            // The key must be a STRING type, and thus the "type-definition" byte is implied (not included in message)
            AmfString keyString = new AmfString(entry.getKey(), true);
            byteBuffer.put(keyString.getBytes());
            byteBuffer.put(entry.getValue().getBytes());
        }
        byteBuffer.put(OBJECT_END_MARKER);
        return byteBuffer.array();
    }
} public class AmfNull implements AmfData {
    public static final int SIZE = 1;

    @Override
    public void writeTo(OutputStream out) throws IOException {
        out.write(AmfType.NULL.getValue());
    }

    @Override
    public void readFrom(InputStream in) throws IOException {
        // Skip data type byte (we assume it's already read)
    }

    public static void writeNullTo(OutputStream out) throws IOException {
        out.write(AmfType.NULL.getValue());
    }

    @Override
    public int getSize() {
        return SIZE;
    }

    @Override
    public byte[] getBytes() {
        byte[] data = new byte[1];
        data[0] = AmfType.NULL.getValue();
        return data;
    }
} public class AmfNumber implements AmfData{
    double value;
    /** Size of an AMF number, in bytes (including type bit) */
    public static final int SIZE = 9;

    public AmfNumber(double value) {
        this.value = value;
    }

    public AmfNumber() {
    }

    public double getValue() {
        return value;
    }

    public void setValue(double value) {
        this.value = value;
    }

    @Override
    public void writeTo(OutputStream out) throws IOException {
        out.write(AmfType.NUMBER.getValue());
        Util.writeDouble(out, value);
    }

    @Override
    public void readFrom(InputStream in) throws IOException {
        // Skip data type byte (we assume it's already read)
        value = Util.readDouble(in);
    }

    public static double readNumberFrom(InputStream in) throws IOException {
        // Skip data type byte
        in.read();
        return Util.readDouble(in);
    }

    public static void writeNumberTo(OutputStream out, double number) throws IOException {
        out.write(AmfType.NUMBER.getValue());
        Util.writeDouble(out, number);
    }

    @Override
    public int getSize() {
        return SIZE;
    }

    @Override
    public byte[] getBytes() {
        ByteBuffer dataBuffer = ByteBuffer.allocate(SIZE);
        dataBuffer.put(AmfType.NUMBER.getValue());
        dataBuffer.putDouble(value);
        return dataBuffer.array();
    }
} public class AmfObject implements AmfData {
    protected Map<String, AmfData> properties = new LinkedHashMap<>();
    protected int size = -1;
    /** Byte sequence that marks the end of an AMF object */
    protected static final byte[] OBJECT_END_MARKER = new byte[]{0x00, 0x00, 0x09};

    public AmfData getProperty(String key) {
        return properties.get(key);
    }

    public void setProperty(String key, AmfData value) {
        properties.put(key, value);
    }

    public void setProperty(String key, boolean value) {
        properties.put(key, new AmfBoolean(value));
    }

    public void setProperty(String key, String value) {
        properties.put(key, new AmfString(value, false));
    }

    public void setProperty(String key, int value) {
        properties.put(key, new AmfNumber(value));
    }

    public void setProperty(String key, double value) {
        properties.put(key, new AmfNumber(value));
    }

    @Override
    public void writeTo(OutputStream out) throws IOException {
        // Begin the object
        out.write(AmfType.OBJECT.getValue());

        // Write key/value pairs in this object
        for (Map.Entry<String, AmfData> entry : properties.entrySet()) {
            // The key must be a STRING type, and thus the "type-definition" byte is implied (not included in message)
            AmfString.writeStringTo(out, entry.getKey(), true);
            entry.getValue().writeTo(out);
        }

        // End the object
        out.write(OBJECT_END_MARKER);
    }

    @Override
    public void readFrom(InputStream in) throws IOException {
        // Skip data type byte (we assume it's already read)
        size = 1;
        InputStream markInputStream = in.markSupported() ? in : new BufferedInputStream(in);

        while (true) {
            // Look for the 3-byte object end marker [0x00 0x00 0x09]
            markInputStream.mark(3);
            byte[] endMarker = new byte[3];
            markInputStream.read(endMarker);

            if (endMarker[0] == OBJECT_END_MARKER[0] && endMarker[1] == OBJECT_END_MARKER[1] && endMarker[2] == OBJECT_END_MARKER[2]) {
                // End marker found
                size += 3;
                return;
            } else {
                // End marker not found; reset the stream to the marked position and read an AMF property
                markInputStream.reset();
                // Read the property key...
                String key = AmfString.readStringFrom(in, true);
                size += AmfString.sizeOf(key, true);
                // ...and the property value
                AmfData value = AmfDecoder.readFrom(markInputStream);
                size += value.getSize();
                properties.put(key, value);
            }
        }
    }

    @Override
    public int getSize() {
        if (size == -1) {
            size = 1; // object marker
            for (Map.Entry<String, AmfData> entry : properties.entrySet()) {
                size += AmfString.sizeOf(entry.getKey(), true);
                size += entry.getValue().getSize();
            }
            size += 3; // end of object marker

        }
        return size;
    }

    @Override
    public byte[] getBytes() {
        int size = getSize();
        ByteBuffer byteBuffer = ByteBuffer.allocate(size);
        byteBuffer.put(AmfType.OBJECT.getValue());
        // Write key/value pairs in this object
        for (Map.Entry<String, AmfData> entry : properties.entrySet()) {
            // The key must be a STRING type, and thus the "type-definition" byte is implied (not included in message)
            AmfString keyString = new AmfString(entry.getKey(), true);
            byteBuffer.put(keyString.getBytes());
            byteBuffer.put(entry.getValue().getBytes());
        }
        byteBuffer.put(OBJECT_END_MARKER);
        return byteBuffer.array();
    }
} public class AmfString implements AmfData {
    private static final String TAG = "AmfString";

    private String value;
    private boolean key;
    private int size = -1;

    public AmfString() {
    }

    public AmfString(String value, boolean isKey) {
        this.value = value;
        this.key = isKey;
    }

    public AmfString(String value) {
        this(value, false);
    }

    public AmfString(boolean isKey) {
        this.key = isKey;
    }

    public String getValue() {
        return value;
    }

    public void setValue(String value) {
        this.value = value;
    }

    public boolean isKey() {
        return key;
    }

    public void setKey(boolean key) {
        this.key = key;
    }

    @Override
    public void writeTo(OutputStream out) throws IOException {
        // Strings are ASCII encoded
        byte[] byteValue = this.value.getBytes();
        // Write the STRING data type definition (except if this String is used as a key)
        if (!key) {
            out.write(AmfType.STRING.getValue());
        }
        // Write 2 bytes indicating string length
        Util.writeUnsignedInt16(out, byteValue.length);
        // Write string
        out.write(byteValue);
    }

    @Override
    public void readFrom(InputStream in) throws IOException {
        // Skip data type byte (we assume it's already read)
        int length = Util.readUnsignedInt16(in);
        size = 3 + length; // 1 + 2 + length
        // Read string value
        byte[] byteValue = new byte[length];
        Util.readBytesUntilFull(in, byteValue);
        value = new String(byteValue);
    }

    public static String readStringFrom(InputStream in, boolean isKey) throws IOException {
        if (!isKey) {
            // Read past the data type byte
            in.read();
        }
        int length = Util.readUnsignedInt16(in);
        // Read string value
        byte[] byteValue = new byte[length];
        Util.readBytesUntilFull(in, byteValue);
        return new String(byteValue);
    }

    public static void writeStringTo(OutputStream out, String string, boolean isKey) throws IOException {
        // Strings are ASCII encoded
        byte[] byteValue = string.getBytes();
        // Write the STRING data type definition (except if this String is used as a key)
        if (!isKey) {
            out.write(AmfType.STRING.getValue());
        }
        // Write 2 bytes indicating string length
        Util.writeUnsignedInt16(out, byteValue.length);
        // Write string
        out.write(byteValue);
    }

    @Override
    public int getSize() {
        size = (isKey() ? 0 : 1) + 2 + value.getBytes().length;
        return size;
    }

    @Override
    public byte[] getBytes() {
        int size = getSize();
        ByteBuffer dataBuffer = ByteBuffer.allocate(size);
        if(!isKey()) {
            dataBuffer.put(AmfType.STRING.getValue());
        }
        dataBuffer.putShort((short) value.getBytes().length);
        dataBuffer.put(value.getBytes());
        return dataBuffer.array();
    }

    /** @return the byte size of the resulting AMF string of the specified value */
    public static int sizeOf(String string, boolean isKey) {
        int size = (isKey ? 0 : 1) + 2 + string.getBytes().length;
        return size;
    }
} public class RtmpPacker implements Packer, AnnexbHelper.AnnexbNaluListener{

    public static final int FIRST_VIDEO = 1;
    public static final int FIRST_AUDIO = 2;
    public static final int AUDIO = 3;
    public static final int KEY_FRAME = 4;
    public static final int INTER_FRAME = 5;
    public static final int CONFIGRATION = 6;

    private OnPacketListener packetListener;
    private boolean isHeaderWrite;
    private boolean isKeyFrameWrite;

    private int mAudioSampleRate, mAudioSampleSize;
    private boolean mIsStereo;

    private AnnexbHelper mAnnexbHelper;

    public RtmpPacker() {
        mAnnexbHelper = new AnnexbHelper();
    }

    @Override
    public void setPacketListener(OnPacketListener listener) {
        packetListener = listener;
    }

    @Override
    public void start() {
        mAnnexbHelper.setAnnexbNaluListener(this);
    }

    @Override
    public void onVideoData(ByteBuffer bb, MediaCodec.BufferInfo bi) {
        mAnnexbHelper.analyseVideoData(bb, bi);
    }

    @Override
    public void onAudioData(ByteBuffer bb, MediaCodec.BufferInfo bi) {
        if(packetListener == null || !isHeaderWrite || !isKeyFrameWrite) {
            return;
        }
        bb.position(bi.offset);
        bb.limit(bi.offset + bi.size);

        byte[] audio = new byte[bi.size];
        bb.get(audio);
        int size = AUDIO_HEADER_SIZE + audio.length;
        ByteBuffer buffer = ByteBuffer.allocate(size);
        FlvPackerHelper.writeAudioTag(buffer, audio, false, mAudioSampleSize);
        packetListener.onPacket(buffer.array(), AUDIO);
    }

    @Override
    public void stop() {
        isHeaderWrite = false;
        isKeyFrameWrite = false;
        mAnnexbHelper.stop();
    }

    @Override
    public void onVideo(byte[] video, boolean isKeyFrame) {
        if(packetListener == null || !isHeaderWrite) {
            return;
        }
        int packetType = INTER_FRAME;
        if(isKeyFrame) {
            isKeyFrameWrite = true;
            packetType = KEY_FRAME;
        }
        //确保第一帧是关键帧，避免一开始出现灰色模糊界面
        if(!isKeyFrameWrite) {
            return;
        }
        int size = VIDEO_HEADER_SIZE + video.length;
        ByteBuffer buffer = ByteBuffer.allocate(size);
        FlvPackerHelper.writeH264Packet(buffer, video, isKeyFrame);
        packetListener.onPacket(buffer.array(), packetType);
    }

    @Override
    public void onSpsPps(byte[] sps, byte[] pps) {
        if(packetListener == null) {
            return;
        }
        //写入第一个视频信息
        writeFirstVideoTag(sps, pps);
        //写入第一个音频信息
        writeFirstAudioTag();
        isHeaderWrite = true;
    }

    private void writeFirstVideoTag(byte[] sps, byte[] pps) {
        int size = VIDEO_HEADER_SIZE + VIDEO_SPECIFIC_CONFIG_EXTEND_SIZE + sps.length + pps.length;
        ByteBuffer buffer = ByteBuffer.allocate(size);
        FlvPackerHelper.writeFirstVideoTag(buffer, sps, pps);
        packetListener.onPacket(buffer.array(), FIRST_VIDEO);
    }

    private void writeFirstAudioTag() {
        int size = AUDIO_SPECIFIC_CONFIG_SIZE + AUDIO_HEADER_SIZE;
        ByteBuffer buffer = ByteBuffer.allocate(size);
        FlvPackerHelper.writeFirstAudioTag(buffer, mAudioSampleRate, mIsStereo, mAudioSampleSize);
        packetListener.onPacket(buffer.array(), FIRST_AUDIO);
    }

    public void initAudioParams(int sampleRate, int sampleSize, boolean isStereo) {
        mAudioSampleRate = sampleRate;
        mAudioSampleSize = sampleSize;
        mIsStereo = isStereo;
    }
} public class Abort extends Chunk {

    private int chunkStreamId;
    
    public Abort(ChunkHeader header) {
        super(header);
    }

    public Abort(int chunkStreamId) {
        super(new ChunkHeader(ChunkType.TYPE_0_FULL, SessionInfo.RTMP_CONTROL_CHANNEL, MessageType.SET_CHUNK_SIZE));
        this.chunkStreamId = chunkStreamId;
    }

    /** @return the ID of the chunk stream to be aborted */
    public int getChunkStreamId() {
        return chunkStreamId;
    }

    /** Sets the ID of the chunk stream to be aborted */
    public void setChunkStreamId(int chunkStreamId) {
        this.chunkStreamId = chunkStreamId;
    }

    @Override
    public void readBody(InputStream in) throws IOException {
        // Value is received in the 4 bytes of the body
        chunkStreamId = Util.readUnsignedInt32(in);
    }

    @Override
    protected void writeBody(OutputStream out) throws IOException {
        Util.writeUnsignedInt32(out, chunkStreamId);
    }
} public class Audio extends ContentData {

    public Audio(ChunkHeader header) {
        super(header);
    }

    public Audio() {
        super(new ChunkHeader(ChunkType.TYPE_0_FULL, SessionInfo.RTMP_AUDIO_CHANNEL, MessageType.AUDIO));
    }

    @Override
    public String toString() {
        return "RTMP Audio";
    }
} public abstract class Chunk {
     
    protected ChunkHeader header;

    public Chunk(ChunkHeader header) {
        this.header = header;
    }

    public ChunkHeader getChunkHeader() {
        return header;
    }
    
    public abstract void readBody(InputStream in) throws IOException;    
    
    protected abstract void writeBody(OutputStream out) throws IOException;
           
    public void writeTo(OutputStream out, final SessionInfo sessionInfo) throws IOException {
        int chunkSize = sessionInfo.getTxChunkSize();
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        writeBody(baos);        
        byte[] body = baos.toByteArray();        
        header.setPacketLength(body.length);
        // Write header for first chunk
        header.writeTo(out, ChunkType.TYPE_0_FULL, sessionInfo);
        int remainingBytes = body.length;
        int pos = 0;
        while (remainingBytes > chunkSize) {
            // Write packet for chunk
            out.write(body, pos, chunkSize);
            remainingBytes -= chunkSize;
            pos += chunkSize;
            // Write header for remain chunk
            header.writeTo(out, ChunkType.TYPE_3_NO_BYTE, sessionInfo);
        }
        out.write(body, pos, remainingBytes);
    }
} public class ChunkHeader {

    private static final String TAG = "ChunkHeader";
    private ChunkType chunkType;
    private int chunkStreamId;
    private int absoluteTimestamp;
    private int timestampDelta = -1;
    private int packetLength;
    private MessageType messageType;
    private int messageStreamId;
    private int extendedTimestamp;

    public ChunkHeader() {
    }

    public ChunkHeader(ChunkType chunkType, int chunkStreamId, MessageType messageType) {
        this.chunkType = chunkType;
        this.chunkStreamId = chunkStreamId;
        this.messageType = messageType;
    }

    public static ChunkHeader readHeader(InputStream in, SessionInfo sessionInfo) throws IOException {
        ChunkHeader rtmpHeader = new ChunkHeader();
        rtmpHeader.readHeaderImpl(in, sessionInfo);
        return rtmpHeader;
    }

    private void readHeaderImpl(InputStream in, SessionInfo sessionInfo) throws IOException {

        int basicHeaderByte = in.read();
        if (basicHeaderByte == -1) {
            throw new EOFException("Unexpected EOF while reading RTMP packet basic header");
        }
        // Read byte 0: chunk type and chunk stream ID
        parseBasicHeader((byte) basicHeaderByte);

        switch (chunkType) {
            case TYPE_0_FULL: { //  b00 = 12 byte header (full header) 
                // Read bytes 1-3: Absolute timestamp
                absoluteTimestamp = Util.readUnsignedInt24(in);
                timestampDelta = 0;
                // Read bytes 4-6: Packet length
                packetLength = Util.readUnsignedInt24(in);
                // Read byte 7: Message type ID
                messageType = MessageType.valueOf((byte) in.read());
                // Read bytes 8-11: Message stream ID (apparently little-endian order)
                byte[] messageStreamIdBytes = new byte[4];
                Util.readBytesUntilFull(in, messageStreamIdBytes);
                messageStreamId = Util.toUnsignedInt32LittleEndian(messageStreamIdBytes);
                // Read bytes 1-4: Extended timestamp
                extendedTimestamp = absoluteTimestamp >= 0xffffff ? Util.readUnsignedInt32(in) : 0;
                if (extendedTimestamp != 0) {
                    absoluteTimestamp = extendedTimestamp;
                }
                sessionInfo.putPreReceiveChunkHeader(chunkStreamId, this);
                break;
            }
            case TYPE_1_LARGE: { // b01 = 8 bytes - like type 0. not including message stream ID (4 last bytes)
                // Read bytes 1-3: Timestamp delta
                timestampDelta = Util.readUnsignedInt24(in);
                // Read bytes 4-6: Packet length
                packetLength = Util.readUnsignedInt24(in);
                // Read byte 7: Message type ID
                messageType = MessageType.valueOf((byte) in.read());
                // Read bytes 1-4: Extended timestamp delta
                extendedTimestamp = timestampDelta >= 0xffffff ? Util.readUnsignedInt32(in) : 0;
                ChunkHeader prevHeader = sessionInfo.getPreReceiveChunkHeader(chunkStreamId);
                if (prevHeader != null) {
                    messageStreamId = prevHeader.messageStreamId;
                    absoluteTimestamp = extendedTimestamp != 0 ? extendedTimestamp : prevHeader.absoluteTimestamp + timestampDelta;
                } else {
                    messageStreamId = 0;
                    absoluteTimestamp = extendedTimestamp != 0 ? extendedTimestamp : timestampDelta;
                }
                sessionInfo.putPreReceiveChunkHeader(chunkStreamId, this);
                break;
            }
            case TYPE_2_TIMESTAMP_ONLY: { // b10 = 4 bytes - Basic Header and timestamp (3 bytes) are included
                // Read bytes 1-3: Timestamp delta
                timestampDelta = Util.readUnsignedInt24(in);
                // Read bytes 1-4: Extended timestamp delta
                extendedTimestamp = timestampDelta >= 0xffffff ? Util.readUnsignedInt32(in) : 0;
                ChunkHeader prevHeader = sessionInfo.getPreReceiveChunkHeader(chunkStreamId);
                packetLength = prevHeader.packetLength;
                messageType = prevHeader.messageType;
                messageStreamId = prevHeader.messageStreamId;
                absoluteTimestamp = extendedTimestamp != 0 ? extendedTimestamp : prevHeader.absoluteTimestamp + timestampDelta;
                sessionInfo.putPreReceiveChunkHeader(chunkStreamId, this);
                break;
            }
            case TYPE_3_NO_BYTE: { // b11 = 1 byte: basic header only
                ChunkHeader prevHeader = sessionInfo.getPreReceiveChunkHeader(chunkStreamId);
                // Read bytes 1-4: Extended timestamp
                extendedTimestamp = prevHeader.timestampDelta >= 0xffffff ? Util.readUnsignedInt32(in) : 0;
                timestampDelta = extendedTimestamp != 0 ? 0xffffff : prevHeader.timestampDelta;
                packetLength = prevHeader.packetLength;
                messageType = prevHeader.messageType;
                messageStreamId = prevHeader.messageStreamId;
                absoluteTimestamp = extendedTimestamp != 0 ? extendedTimestamp : prevHeader.absoluteTimestamp + timestampDelta;
                sessionInfo.putPreReceiveChunkHeader(chunkStreamId, this);
                break;
            }
            default:
                Log.e(TAG, "readHeaderImpl(): Invalid chunk type; basic header byte was: " + Util.toHexString((byte) basicHeaderByte));
                throw new IOException("Invalid chunk type; basic header byte was: " + Util.toHexString((byte) basicHeaderByte));
        }
    }

    public void writeTo(OutputStream out, ChunkType chunkType, final SessionInfo sessionInfo) throws IOException {
        // Write basic header byte
        out.write(((byte) (chunkType.getValue() << 6) | chunkStreamId));
        switch (chunkType) {
            case TYPE_0_FULL: { //  b00 = 12 byte header (full header)
                absoluteTimestamp = (int) sessionInfo.markAbsoluteTimestampTx();
                Util.writeUnsignedInt24(out, absoluteTimestamp >= 0xffffff ? 0xffffff : absoluteTimestamp);
                Util.writeUnsignedInt24(out, packetLength);
                out.write(messageType.getValue());
                Util.writeUnsignedInt32LittleEndian(out, messageStreamId);
                if (absoluteTimestamp >= 0xffffff) {
                    extendedTimestamp = absoluteTimestamp;
                    Util.writeUnsignedInt32(out, extendedTimestamp);
                }
                break;
            }
            case TYPE_1_LARGE: { // b01 = 8 bytes - like type 0. not including message ID (4 last bytes)
                absoluteTimestamp = (int) sessionInfo.markAbsoluteTimestampTx();
                ChunkHeader preChunkHeader = sessionInfo.getPreSendChunkHeader(chunkStreamId);
                timestampDelta = absoluteTimestamp - preChunkHeader.absoluteTimestamp;
                Util.writeUnsignedInt24(out, absoluteTimestamp >= 0xffffff ? 0xffffff : timestampDelta);
                Util.writeUnsignedInt24(out, packetLength);
                out.write(messageType.getValue());
                if (absoluteTimestamp >= 0xffffff) {
                    extendedTimestamp = absoluteTimestamp;
                    Util.writeUnsignedInt32(out, absoluteTimestamp);
                }
                break;
            }
            case TYPE_2_TIMESTAMP_ONLY: { // b10 = 4 bytes - Basic Header and timestamp (3 bytes) are included
                absoluteTimestamp = (int) sessionInfo.markAbsoluteTimestampTx();
                ChunkHeader preChunkHeader = sessionInfo.getPreSendChunkHeader(chunkStreamId);
                timestampDelta = absoluteTimestamp - preChunkHeader.absoluteTimestamp;
                Util.writeUnsignedInt24(out, (absoluteTimestamp >= 0xffffff) ? 0xffffff : timestampDelta);
                if (absoluteTimestamp >= 0xffffff) {
                    extendedTimestamp = absoluteTimestamp;
                    Util.writeUnsignedInt32(out, extendedTimestamp);
                }
                break;
            }
            case TYPE_3_NO_BYTE: { // b11 = 1 byte: basic header only
                absoluteTimestamp = (int) sessionInfo.markAbsoluteTimestampTx();
                if (absoluteTimestamp >= 0xffffff) {
                    extendedTimestamp = absoluteTimestamp;
                    Util.writeUnsignedInt32(out, extendedTimestamp);
                }
                break;
            }
            default:
                throw new IOException("Invalid chunk type: " + chunkType);
        }
    }

    private void parseBasicHeader(byte basicHeaderByte) {
        chunkType = ChunkType.valueOf((byte) ((0xff & basicHeaderByte) >>> 6)); // 2 most significant bits define the chunk type
        chunkStreamId = basicHeaderByte & 0x3F; // 6 least significant bits define chunk stream ID
    }

    /** @return the RTMP chunk stream ID (channel ID) for this chunk */
    public int getChunkStreamId() {
        return chunkStreamId;
    }

    public int getPacketLength() {
        return packetLength;
    }

    public MessageType getMessageType() {
        return messageType;
    }

    /** Sets the RTMP chunk stream ID (channel ID) for this chunk */
    public void setChunkStreamId(int channelId) {
        this.chunkStreamId = channelId;
    }

    public void setMessageStreamId(int messageStreamId) {
        this.messageStreamId = messageStreamId;
    }

    public void setPacketLength(int packetLength) {
        this.packetLength = packetLength;
    }

    @Override
    public String toString() {
        return "ChunkType:" + chunkType + " ChunkStreamId:" + chunkStreamId + " absoluteTimestamp:" + absoluteTimestamp + " timestampDelta:" + timestampDelta +
                " messageLength:" + packetLength + " messageType:" + messageType +
                " messageStreamId:" + messageStreamId + " extendedTimestamp:" + extendedTimestamp;
    }
} public class Command extends VariableBodyRtmpPacket {

    private static final String TAG = "Command";

    private String commandName;
    private int transactionId;    

    public Command(ChunkHeader header) {
        super(header);
    }
    
    public Command(String commandName, int transactionId) {
        super(new ChunkHeader(ChunkType.TYPE_0_FULL, SessionInfo.RTMP_COMMAND_CHANNEL, MessageType.COMMAND_AMF0));
        this.commandName = commandName;
        this.transactionId = transactionId;
    }

    public String getCommandName() {
        return commandName;
    }

    public void setCommandName(String commandName) {
        this.commandName = commandName;
    }

    public int getTransactionId() {
        return transactionId;
    }

    public void setTransactionId(int transactionId) {
        this.transactionId = transactionId;
    }    

    @Override
    public void readBody(InputStream in) throws IOException {
        // The command name and transaction ID are always present (AMF string followed by number)
        commandName = AmfString.readStringFrom(in, false);
        transactionId = (int) AmfNumber.readNumberFrom(in);
        int bytesRead = AmfString.sizeOf(commandName, false) + AmfNumber.SIZE;
        readVariableData(in, bytesRead);
    }

    @Override
    protected void writeBody(OutputStream out) throws IOException {
        AmfString.writeStringTo(out, commandName, false);
        AmfNumber.writeNumberTo(out, transactionId);
        // Write body data
        writeVariableData(out);
    }
    
    @Override
    public String toString() {
        return "RTMP Command (command: " + commandName + ", transaction ID: " + transactionId + ")";
    }
} public class Data extends VariableBodyRtmpPacket {

    private String type;

    public Data(ChunkHeader header) {
        super(header);
    }

    public Data(String type) {
        super(new ChunkHeader(ChunkType.TYPE_0_FULL, SessionInfo.RTMP_COMMAND_CHANNEL, MessageType.DATA_AMF0));
        this.type = type;
    }

    public String getType() {
        return type;
    }

    public void setType(String type) {
        this.type = type;
    }

    @Override
    public void readBody(InputStream in) throws IOException {
        // Read notification type
        type = AmfString.readStringFrom(in, false);
        int bytesRead = AmfString.sizeOf(type, false);
        // Read data body
        readVariableData(in, bytesRead);
    }

    /** 
     * This method is public for Data to make it easy to dump its contents to 
     * another output stream
     */
    @Override
    public void writeBody(OutputStream out) throws IOException {
        AmfString.writeStringTo(out, type, false);
        writeVariableData(out);
    }
} public final class Handshake {
    private static final String TAG = "Handshake";
    /** S1 as sent by the server */
    private byte[] s1;
    private static final int PROTOCOL_VERSION = 0x03;
    private static final int HANDSHAKE_SIZE = 1536;
    private static final int SHA256_DIGEST_SIZE = 32;
    
    private static final int DIGEST_OFFSET_INDICATOR_POS = 772; // should either be byte 772 or byte 8
    
    private static final byte[] GENUINE_FP_KEY = {
        (byte) 0x47, (byte) 0x65, (byte) 0x6E, (byte) 0x75, (byte) 0x69, (byte) 0x6E, (byte) 0x65, (byte) 0x20,
        (byte) 0x41, (byte) 0x64, (byte) 0x6F, (byte) 0x62, (byte) 0x65, (byte) 0x20, (byte) 0x46, (byte) 0x6C,
        (byte) 0x61, (byte) 0x73, (byte) 0x68, (byte) 0x20, (byte) 0x50, (byte) 0x6C, (byte) 0x61, (byte) 0x79,
        (byte) 0x65, (byte) 0x72, (byte) 0x20, (byte) 0x30, (byte) 0x30, (byte) 0x31, // Genuine Adobe Flash Player 001
        (byte) 0xF0, (byte) 0xEE, (byte) 0xC2, (byte) 0x4A, (byte) 0x80, (byte) 0x68, (byte) 0xBE, (byte) 0xE8,
        (byte) 0x2E, (byte) 0x00, (byte) 0xD0, (byte) 0xD1, (byte) 0x02, (byte) 0x9E, (byte) 0x7E, (byte) 0x57,
        (byte) 0x6E, (byte) 0xEC, (byte) 0x5D, (byte) 0x2D, (byte) 0x29, (byte) 0x80, (byte) 0x6F, (byte) 0xAB,
        (byte) 0x93, (byte) 0xB8, (byte) 0xE6, (byte) 0x36, (byte) 0xCF, (byte) 0xEB, (byte) 0x31, (byte) 0xAE};

    /** Generates and writes the first handshake packet (C0) */
    public final void writeC0(OutputStream out) throws IOException {
        Log.d(TAG, "writeC0");
        out.write(PROTOCOL_VERSION);
    }

    public final void readS0(InputStream in) throws IOException {
        Log.d(TAG, "readS0");
        byte s0 = (byte) in.read();
        if (s0 != PROTOCOL_VERSION) {
            if (s0 == -1) {
                throw new IOException("InputStream closed");
            } else {
                throw new IOException("Invalid RTMP protocol version; expected " + PROTOCOL_VERSION + ", got " + s0);
            }
        }
    }

    /** Generates and writes the second handshake packet (C1) */
    public final void writeC1(OutputStream out) throws IOException {
        Log.d(TAG, "writeC1");
//        Util.writeUnsignedInt32(out, (int) (System.currentTimeMillis() / 1000)); // Bytes 0 - 3 bytes: current epoch (timestamp)
        //out.write(new byte[]{0x09, 0x00, 0x7c, 0x02}); // Bytes 4 - 7: Flash player version: 9.0.124.2

//        out.write(new byte[]{(byte) 0x80, 0x00, 0x07, 0x02}); // Bytes 4 - 7: Flash player version: 11.2.202.233


        Log.d(TAG, "writeC1(): Calculating digest offset");
        Random random = new Random();
        // Since we are faking a real Flash Player handshake, include a digest in C1        
        // Choose digest offset point (scheme 1; that is, offset is indicated by bytes 772 - 775 (4 bytes) )
        final int digestOffset = random.nextInt(HANDSHAKE_SIZE - DIGEST_OFFSET_INDICATOR_POS - 4 - 8 - SHA256_DIGEST_SIZE); //random.nextInt(DIGEST_OFFSET_INDICATOR_POS - SHA256_DIGEST_SIZE);

        final int absoluteDigestOffset = ((digestOffset % 728) + DIGEST_OFFSET_INDICATOR_POS + 4);
        Log.d(TAG, "writeC1(): (real value of) digestOffset: " + digestOffset);
        
        
        Log.d(TAG, "writeC1(): recalculated digestOffset: " + absoluteDigestOffset);

        int remaining = digestOffset;
        final byte[] digestOffsetBytes = new byte[4];
        for (int i = 3; i >= 0; i--) {            
            if (remaining > 255) {
                digestOffsetBytes[i] = (byte)255;
                remaining -= 255;
            } else {
                digestOffsetBytes[i] = (byte)remaining;
                remaining -= remaining;
            }
        }
        
        
        
        
        // Calculate the offset value that will be written
        //inal byte[] digestOffsetBytes = Util.unsignedInt32ToByteArray(digestOffset);// //((digestOffset - DIGEST_OFFSET_INDICATOR_POS) % 728)); // Thanks to librtmp for the mod 728                
        Log.d(TAG, "writeC1(): digestOffsetBytes: " + Util.toHexString(digestOffsetBytes));  //Util.unsignedInt32ToByteArray((digestOffset % 728))));

        // Create random bytes up to the digest offset point
        byte[] partBeforeDigest = new byte[absoluteDigestOffset];
        Log.d(TAG, "partBeforeDigest(): size: " + partBeforeDigest.length);
        random.nextBytes(partBeforeDigest);

        Log.d(TAG, "writeC1(): Writing timestamp and Flash Player version");
        byte[] timeStamp = Util.unsignedInt32ToByteArray((int) (System.currentTimeMillis() / 1000));
        System.arraycopy(timeStamp, 0, partBeforeDigest, 0, 4); // Bytes 0 - 3 bytes: current epoch timestamp
        System.arraycopy(new byte[]{(byte) 0x80, 0x00, 0x07, 0x02}, 0, partBeforeDigest, 4, 4); // Bytes 4 - 7: Flash player version: 11.2.202.233

        // Create random bytes for the part after the digest
        byte[] partAfterDigest = new byte[HANDSHAKE_SIZE - absoluteDigestOffset - SHA256_DIGEST_SIZE]; // subtract 8 because of initial 8 bytes already written
        Log.d(TAG, "partAfterDigest(): size: " + partAfterDigest.length);
        random.nextBytes(partAfterDigest);


        // Set the offset byte
//        if (digestOffset > 772) {                      
            Log.d(TAG, "copying digest offset bytes in partBeforeDigest");
            System.arraycopy(digestOffsetBytes, 0, partBeforeDigest, 772, 4);          
//        } else {
        // Implied offset of partAfterDigest is digestOffset + 32
///        Log.d(TAG, "copying digest offset bytes in partAfterDigest");
///        Log.d(TAG, " writing to location: " + (DIGEST_OFFSET_INDICATOR_POS - digestOffset - SHA256_DIGEST_SIZE - 8));
//        System.arraycopy(digestOffsetBytes, 0, partAfterDigest, (DIGEST_OFFSET_INDICATOR_POS - digestOffset - SHA256_DIGEST_SIZE - 8), 4);
//        }

        Log.d(TAG, "writeC1(): Calculating digest");
        byte[] tempBuffer = new byte[HANDSHAKE_SIZE - SHA256_DIGEST_SIZE];
        System.arraycopy(partBeforeDigest, 0, tempBuffer, 0, partBeforeDigest.length);
        System.arraycopy(partAfterDigest, 0, tempBuffer, partBeforeDigest.length, partAfterDigest.length);

        Crypto crypto = new Crypto();
        byte[] digest = crypto.calculateHmacSHA256(tempBuffer, GENUINE_FP_KEY, 30);

        // Now write the packet
        Log.d(TAG, "writeC1(): writing C1 packet");
        out.write(partBeforeDigest);
        out.write(digest);
        out.write(partAfterDigest);
    }

    public final void readS1(InputStream in) throws IOException {
        // S1 == 1536 bytes. We do not bother with checking the content of it
        Log.d(TAG, "readS1");
        s1 = new byte[HANDSHAKE_SIZE];

        // Read server time (4 bytes)
        int totalBytesRead = 0;
        int read;
        do {
            read = in.read(s1, totalBytesRead, (HANDSHAKE_SIZE - totalBytesRead));
            if (read != -1) {
                totalBytesRead += read;
            }
        } while (totalBytesRead < HANDSHAKE_SIZE);

        if (totalBytesRead != HANDSHAKE_SIZE) {
            throw new IOException("Unexpected EOF while reading S1, expected " + HANDSHAKE_SIZE + " bytes, but only read " + totalBytesRead + " bytes");
        } else {
            Log.d(TAG, "readS1(): S1 total bytes read OK");
        }
    }

    /** Generates and writes the third handshake packet (C2) */
    public final void writeC2(OutputStream out) throws IOException {
        Log.d(TAG, "readC2");
        // C2 is an echo of S1
        if (s1 == null) {
            throw new IllegalStateException("C2 cannot be written without S1 being read first");
        }
        out.write(s1);
    }

    public final void readS2(InputStream in) throws IOException {
        // S2 should be an echo of C1, but we are not too strict
        Log.d(TAG, "readS2");
        byte[] sr_serverTime = new byte[4];
        byte[] s2_serverVersion = new byte[4];
        byte[] s2_rest = new byte[HANDSHAKE_SIZE - 8]; // subtract 4+4 bytes for time and version

        // Read server time (4 bytes)
        int totalBytesRead = 0;
        int read;
        do {
            read = in.read(sr_serverTime, totalBytesRead, (4 - totalBytesRead));
            if (read == -1) {
                // End of stream reached - should not have happened at this point
                throw new IOException("Unexpected EOF while reading S2 bytes 0-3");
            } else {
                totalBytesRead += read;
            }
        } while (totalBytesRead < 4);

        // Read server version (4 bytes)
        totalBytesRead = 0;
        do {
            read = in.read(s2_serverVersion, totalBytesRead, (4 - totalBytesRead));
            if (read == -1) {
                // End of stream reached - should not have happened at this point
                throw new IOException("Unexpected EOF while reading S2 bytes 4-7");
            } else {
                totalBytesRead += read;
            }
        } while (totalBytesRead < 4);

        // Read 1528 bytes (to make up S1 total size of 1536 bytes)
        final int remainingBytes = HANDSHAKE_SIZE - 8;
        totalBytesRead = 0;
        do {
            read = in.read(s2_rest, totalBytesRead, (remainingBytes - totalBytesRead));
            if (read != -1) {
                totalBytesRead += read;
            }
        } while (totalBytesRead < remainingBytes && read != -1);

        if (totalBytesRead != remainingBytes) {
            throw new IOException("Unexpected EOF while reading remainder of S2, expected " + remainingBytes + " bytes, but only read " + totalBytesRead + " bytes");
        } else {
            Log.d(TAG, "readS2(): S2 total bytes read OK");
        }

        // Technically we should check that S2 == C1, but for now this is ignored
    }
} public enum MessageType {
    /**
     * Protocol control message 1
     * Set Chunkd Size, is used to notify the peer a new maximum chunk size to use.
     */
    SET_CHUNK_SIZE(0x01),
    /**
     * Protocol control message 2
     * Abort Message, is used to notify the peer if it is waiting for chunks
     * to complete a message, then to discard the partially received message
     * over a chunk stream and abort processing of that message.
     */
    ABORT(0x02),
    /**
     * Protocol control message 3
     * The client or the server sends the acknowledgment to the peer after
     * receiving bytes equal to the window size. The window size is the
     * maximum number of bytes that the sender sends without receiving
     * acknowledgment from the receiver.
     */
    ACKNOWLEDGEMENT(0x03),
    /**
     * Protocol control message 4
     * The client or the server sends this message to notify the peer about
     * the user control events. This message carries Event type and Event
     * data.
     * Also known as a PING message in some RTMP implementations.
     */
    USER_CONTROL_MESSAGE(0x04),
    /**
     * Protocol control message 5
     * The client or the server sends this message to inform the peer which
     * window size to use when sending acknowledgment.
     * Also known as ServerBW ("server bandwidth") in some RTMP implementations.
     */
    WINDOW_ACKNOWLEDGEMENT_SIZE(0x05),
    /**
     * Protocol control message 6
     * The client or the server sends this message to update the output
     * bandwidth of the peer. The output bandwidth value is the same as the
     * window size for the peer.
     * Also known as ClientBW ("client bandwidth") in some RTMP implementations.
     */
    SET_PEER_BANDWIDTH(0x06),
    /**
     * RTMP audio packet (0x08)
     * The client or the server sends this message to send audio data to the peer.
     */
    AUDIO(0x08),
    /**
     * RTMP video packet (0x09)
     * The client or the server sends this message to send video data to the peer.
     */
    VIDEO(0x09),
    /**
     * RTMP message type 0x0F
     * The client or the server sends this message to send Metadata or any
     * user data to the peer. Metadata includes details about the data (audio, video etc.)
     * like creation time, duration, theme and so on.
     * This is the AMF3-encoded version.
     */
    DATA_AMF3(0x0F),
    /**
     * RTMP message type 0x10
     * A shared object is a Flash object (a collection of name value pairs)
     * that are in synchronization across multiple clients, instances, and
     * so on.
     * This is the AMF3 version: kMsgContainerEx=16 for AMF3.
     */
    SHARED_OBJECT_AMF3(0x10),
    /**
     * RTMP message type 0x11
     * Command messages carry the AMF-encoded commands between the client
     * and the server.
     * A command message consists of command name, transaction ID, and command object that
     * contains related parameters.
     * This is the AMF3-encoded version.
     */
    COMMAND_AMF3(0x11),
    /**
     * RTMP message type 0x12
     * The client or the server sends this message to send Metadata or any
     * user data to the peer. Metadata includes details about the data (audio, video etc.)
     * like creation time, duration, theme and so on.
     * This is the AMF0-encoded version.
     */
    DATA_AMF0(0x12),
    /**
     * RTMP message type 0x14
     * Command messages carry the AMF-encoded commands between the client
     * and the server.
     * A command message consists of command name, transaction ID, and command object that
     * contains related parameters.
     * This is the common AMF0 version, also known as INVOKE in some RTMP implementations.
     */
    COMMAND_AMF0(0x14),
    /**
     * RTMP message type 0x13
     * A shared object is a Flash object (a collection of name value pairs)
     * that are in synchronization across multiple clients, instances, and
     * so on.
     * This is the AMF0 version: kMsgContainer=19 for AMF0.
     */
    SHARED_OBJECT_AMF0(0x13),
    /**
     * RTMP message type 0x16
     * An aggregate message is a single message that contains a list of sub-messages.
     */
    AGGREGATE_MESSAGE(0x16);
    private byte value;
    private static final Map<Byte, MessageType> quickLookupMap = new HashMap<>();

    static {
        for (MessageType messageTypId : MessageType.values()) {
            quickLookupMap.put(messageTypId.getValue(), messageTypId);
        }
    }

    MessageType(int value) {
        this.value = (byte) value;
    }

    /** Returns the value of this chunk type */
    public byte getValue() {
        return value;
    }

    public static MessageType valueOf(byte messageTypeId) {
        if (quickLookupMap.containsKey(messageTypeId)) {
            return quickLookupMap.get(messageTypeId);
        } else {
            throw new IllegalArgumentException("Unknown message type byte: " + Util.toHexString(messageTypeId));
        }
    }
} public class UserControl extends Chunk {

    /**
     * Control message type
     * Docstring adapted from the official Adobe RTMP spec, section 3.7
     */
    public static enum Type {

        /** 
         * Type: 0
         * The server sends this event to notify the client that a stream has become 
         * functional and can be used for communication. By default, this event
         * is sent on ID 0 after the application connect command is successfully
         * received from the client. 
         * 
         * Event Data:
         * eventData[0] (int) the stream ID of the stream that became functional
         */
        STREAM_BEGIN(0),
        /**
         * Type: 1
         * The server sends this event to notify the client that the playback of
         * data is over as requested on this stream. No more data is sent without
         * issuing additional commands. The client discards the messages received
         * for the stream.
         * 
         * Event Data:
         * eventData[0]: the ID of thestream on which playback has ended.
         */
        STREAM_EOF(1),
        /**
         * Type: 2
         * The server sends this event to notify the client that there is no 
         * more data on the stream. If the server does not detect any message for
         * a time period, it can notify the subscribed clients that the stream is 
         * dry. 
         * 
         * Event Data:
         * eventData[0]: the stream ID of the dry stream. 
         */
        STREAM_DRY(2),
        /**
         * Type: 3
         * The client sends this event to inform the server of the buffer size 
         * (in milliseconds) that is used to buffer any data coming over a stream.
         * This event is sent before the server starts  processing the stream.
         * 
         * Event Data:
         * eventData[0]: the stream ID and 
         * eventData[1]: the buffer length, in milliseconds.
         */
        SET_BUFFER_LENGTH(3),
        /**
         * Type: 4
         * The server sends this event to notify the client that the stream is a
         * recorded stream.
         * 
         * Event Data:
         * eventData[0]: the stream ID of the recorded stream.         
         */
        STREAM_IS_RECORDED(4),
        /**
         * Type: 6
         * The server sends this event to test whether the client is reachable. 
         * 
         * Event Data:
         * eventData[0]: a timestamp representing the local server time when the server dispatched the command. 
         * 
         * The client responds with PING_RESPONSE on receiving PING_REQUEST.
         */
        PING_REQUEST(6),
        /**
         * Type: 7
         * The client sends this event to the server in response to the ping request. 
         * 
         * Event Data:
         * eventData[0]: the 4-byte timestamp which was received with the PING_REQUEST.
         */
        PONG_REPLY(7),
        /**
         * Type: 31 (0x1F)
         * 
         * This user control type is not specified in any official documentation, but
         * is sent by Flash Media Server 3.5. Thanks to the rtmpdump devs for their
         * explanation: 
         * 
         * Buffer Empty (unofficial name): After the server has sent a complete buffer, and
         * sends this Buffer Empty message, it will wait until the play
         * duration of that buffer has passed before sending a new buffer.
         * The Buffer Ready message will be sent when the new buffer starts.
         *
         * (see also: http://repo.or.cz/w/rtmpdump.git/blob/8880d1456b282ee79979adbe7b6a6eb8ad371081:/librtmp/rtmp.c#l2787)         
         */
        BUFFER_EMPTY(31),
        /**
         * Type: 32 (0x20)
         * 
         * This user control type is not specified in any official documentation, but
         * is sent by Flash Media Server 3.5. Thanks to the rtmpdump devs for their
         * explanation: 
         * 
         * Buffer Ready (unofficial name): After the server has sent a complete buffer, and
         * sends a Buffer Empty message, it will wait until the play
         * duration of that buffer has passed before sending a new buffer.
         * The Buffer Ready message will be sent when the new buffer starts.
         * (There is no BufferReady message for the very first buffer;
         * presumably the Stream Begin message is sufficient for that
         * purpose.)
         *
         * (see also: http://repo.or.cz/w/rtmpdump.git/blob/8880d1456b282ee79979adbe7b6a6eb8ad371081:/librtmp/rtmp.c#l2787)         
         */
        BUFFER_READY(32);
        
        private int intValue;
        private static final Map<Integer, Type> quickLookupMap = new HashMap<Integer, Type>();

        static {
            for (Type type : Type.values()) {
                quickLookupMap.put(type.getIntValue(), type);
            }
        }

        private Type(int intValue) {
            this.intValue = intValue;
        }

        public int getIntValue() {
            return intValue;
        }

        public static Type valueOf(int intValue) {
            return quickLookupMap.get(intValue);
        }
    }
    private Type type;
    private int[] eventData;

    public UserControl(ChunkHeader header) {
        super(header);
    }

    public UserControl() {
        super(new ChunkHeader(ChunkType.TYPE_0_FULL, SessionInfo.RTMP_CONTROL_CHANNEL, MessageType.USER_CONTROL_MESSAGE));
    }

    /** Convenience construtor that creates a "pong" message for the specified ping */
    public UserControl(UserControl replyToPing) {
        this(Type.PONG_REPLY);
        this.eventData = replyToPing.eventData;
    }

    public UserControl(Type type) {
        this();
        this.type = type;
    }

    public Type getType() {
        return type;
    }

    public void setType(Type type) {
        this.type = type;
    }

    /** 
     * Convenience method for getting the first event data item, as most user control
     * message types only have one event data item anyway
     * This is equivalent to calling <code>getEventData()[0]</code>
     */
    public int getFirstEventData() {
        return eventData[0];
    }

    public int[] getEventData() {
        return eventData;
    }

    /** Used to set (a single) event data for most user control message types */
    public void setEventData(int eventData) {
        if (type == Type.SET_BUFFER_LENGTH) {
            throw new IllegalStateException("SET_BUFFER_LENGTH requires two event data values; use setEventData(int, int) instead");
        }
        this.eventData = new int[]{eventData};
    }

    /** Used to set event data for the SET_BUFFER_LENGTH user control message types */
    public void setEventData(int streamId, int bufferLength) {
        if (type != Type.SET_BUFFER_LENGTH) {
            throw new IllegalStateException("User control type " + type + " requires only one event data value; use setEventData(int) instead");
        }
        this.eventData = new int[]{streamId, bufferLength};
    }

    @Override
    public void readBody(InputStream in) throws IOException {
        // Bytes 0-1: first parameter: ping type (mandatory)
        type = Type.valueOf(Util.readUnsignedInt16(in));
        int bytesRead = 2;
        // Event data (1 for most types, 2 for SET_BUFFER_LENGTH)
        if (type == Type.SET_BUFFER_LENGTH) {
            setEventData(Util.readUnsignedInt32(in), Util.readUnsignedInt32(in));
            bytesRead += 8;
        } else {
            setEventData(Util.readUnsignedInt32(in));
            bytesRead += 4;
        }
        // To ensure some strange non-specified UserControl/ping message does not slip through
        assert header.getPacketLength() == bytesRead;
    }

    @Override
    protected void writeBody(OutputStream out) throws IOException {
        // Write the user control message type
        Util.writeUnsignedInt16(out, type.getIntValue());
        // Now write the event data
        Util.writeUnsignedInt32(out, eventData[0]);
        if (type == Type.SET_BUFFER_LENGTH) {
            Util.writeUnsignedInt32(out, eventData[1]);
        }
    }

    @Override
    public String toString() {
        return "RTMP User Control (type: " + type + ", event data: " + eventData + ")";
    }
} public class Video extends ContentData {

    public Video(ChunkHeader header) {
        super(header);
    }

    public Video() {
        super(new ChunkHeader(ChunkType.TYPE_0_FULL, SessionInfo.RTMP_VIDEO_CHANNEL, MessageType.VIDEO));
    }

    @Override
    public String toString() {
        return "RTMP Video";
    }
} public class WindowAckSize extends Chunk {

    private int acknowledgementWindowSize;

    public WindowAckSize(ChunkHeader header) {
        super(header);
    }
    
    public WindowAckSize(int acknowledgementWindowSize) {
        super(new ChunkHeader(ChunkType.TYPE_0_FULL, SessionInfo.RTMP_CONTROL_CHANNEL, MessageType.WINDOW_ACKNOWLEDGEMENT_SIZE));
        this.acknowledgementWindowSize = acknowledgementWindowSize;
    }


    public int getAcknowledgementWindowSize() {
        return acknowledgementWindowSize;
    }

    public void setAcknowledgementWindowSize(int acknowledgementWindowSize) {
        this.acknowledgementWindowSize = acknowledgementWindowSize;
    }

    @Override
    public void readBody(InputStream in) throws IOException {
        acknowledgementWindowSize = Util.readUnsignedInt32(in);
    }

    @Override
    protected void writeBody(OutputStream out) throws IOException {        
        Util.writeUnsignedInt32(out, acknowledgementWindowSize);
    }
    
    @Override
    public String toString() {
        return "RTMP Window Acknowledgment Size";
    }
} public interface ISendQueue {
    void start();
    void stop();
    void setBufferSize(int size);
    void putFrame(Frame frame);
    Frame takeFrame();
    void setSendQueueListener(SendQueueListener listener);
}
import android.util.Log;
import org.yeshen.video.librtmp.core.IRtmpConnection;
import org.yeshen.video.librtmp.unstable.net.Frame;
import org.yeshen.video.librtmp.unstable.net.amf.AmfMap;
import org.yeshen.video.librtmp.unstable.net.amf.AmfNull;
import org.yeshen.video.librtmp.unstable.net.amf.AmfNumber;
import org.yeshen.video.librtmp.unstable.net.amf.AmfObject;
import org.yeshen.video.librtmp.unstable.net.amf.AmfString;
import org.yeshen.video.librtmp.unstable.net.packer.rtmp.RtmpPacker;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.packets.Abort;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.packets.Audio;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.packets.Chunk;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.packets.ChunkHeader;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.packets.Command;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.packets.Data;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.packets.Handshake;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.packets.MessageType;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.packets.UserControl;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.packets.Video;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.packets.WindowAckSize;
import org.yeshen.video.librtmp.unstable.net.sender.sendqueue.ISendQueue;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
package org.yeshen.video.librtmp.unstable.net.sender.rtmp.io;




/**
 * Main RTMP connection implementation class
 *
 * @author francois, leoma
 */
public class RtmpConnection implements IRtmpConnection,OnReadListener, OnWriteListener {

    private static final String TAG = "RtmpConnection";
    private static final Pattern rtmpUrlPattern = Pattern.compile("^rtmp://([^/:]+)(:(\\d+))*/([^/]+)(/(.*))*$");

    private RtmpConnectListener listener;
    private Socket socket;
    private SessionInfo sessionInfo;
    private ReadThread readThread;
    private WriteThread writeThread;
    private State state = State.INIT;
    private int transactionIdCounter = 0;
    private int currentStreamId = -1;
    private ConnectData connectData;

    private int videoWidth, videoHeight;
    private int audioSampleRate, audioSampleSize;
    private boolean isAudioStereo;

    private boolean publishPermitted;
    private ISendQueue mSendQueue;

    private static class ConnectData {
        String appName;
        String streamName;
        String swfUrl;
        String tcUrl;
        String pageUrl;
        int port;
        String host;
    }

    public void setConnectListener(RtmpConnectListener listener) {
        this.listener = listener;
    }

    public void setSendQueue(ISendQueue sendQueue) {
        mSendQueue = sendQueue;
    }

    public void connect(String url) {
        state = State.INIT;
        connectData = parseRtmpUrl(url);
        if (connectData == null) {
            if (listener != null) {
                listener.onUrlInvalid();
            }
            return;
        }
        String host = connectData.host;
        int port = connectData.port;
        String appName = connectData.appName;
        String streamName = connectData.streamName;
        Log.d(TAG, "connect() called. Host: " + host + ", port: " + port + ", appName: " + appName + ", publishPath: " + streamName);
        socket = new Socket();
        SocketAddress socketAddress = new InetSocketAddress(host, port);
        try {
            socket.connect(socketAddress, 3000);
        } catch (IOException e) {
            e.printStackTrace();
            if (listener != null) {
                listener.onSocketConnectFail();
            }
            return;
        }
        if (listener != null) {
            listener.onSocketConnectSuccess();
        }
        BufferedInputStream in;
        BufferedOutputStream out;
        state = State.HANDSHAKE;
        try {
            Log.d(TAG, "connect(): socket connection established, doing handhake...");
            in = new BufferedInputStream(socket.getInputStream());
            out = new BufferedOutputStream(socket.getOutputStream());
            handshake(in, out);
        } catch (IOException e) {
            e.printStackTrace();
            state = State.INIT;
            clearSocket();
            if (listener != null) {
                listener.onHandshakeFail();
            }
            return;
        }
        if (listener != null) {
            listener.onHandshakeSuccess();
        }
        sessionInfo = new SessionInfo();
        readThread = new ReadThread(in, sessionInfo);
        writeThread = new WriteThread(out, sessionInfo);
        readThread.setOnReadListener(this);
        writeThread.setWriteListener(this);
        writeThread.setSendQueue(mSendQueue);
        readThread.start();
        writeThread.start();
        rtmpConnect();
    }

    public void publishAudioData(byte[] data, int type) {
        if (currentStreamId == -1) {
            return;
        }
        if (!publishPermitted) {
            return;
        }
        Audio audio = new Audio();
        audio.setData(data);
        audio.getChunkHeader().setMessageStreamId(currentStreamId);
        Frame frame;
        if (type == RtmpPacker.FIRST_AUDIO) {
            frame = new Frame(audio, type, Frame.FRAME_TYPE_CONFIGURATION);
        } else {
            frame = new Frame(audio, type, Frame.FRAME_TYPE_AUDIO);
        }
        mSendQueue.putFrame(frame);
    }

    public void publishVideoData(byte[] data, int type) {
        if (currentStreamId == -1) {
            return;
        }
        if (!publishPermitted) {
            return;
        }
        Video video = new Video();