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 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 interface Sender {
    void start();
    void onData(byte[] data, int type);
    void stop();
} public interface RtmpConnectListener {
    void onUrlInvalid();
    void onSocketConnectSuccess();
    void onSocketConnectFail();
    void onHandshakeSuccess();
    void onHandshakeFail();
    void onRtmpConnectSuccess();
    void onRtmpConnectFail();
    void onCreateStreamSuccess();
    void onCreateStreamFail();
    void onPublishSuccess();
    void onPublishFail();
    void onSocketDisconnect();
    void onStreamEnd();
} 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();
        video.setData(data);
        video.getChunkHeader().setMessageStreamId(currentStreamId);
        Frame frame;
        if (type == RtmpPacker.FIRST_VIDEO) {
            frame = new Frame(video, type, Frame.FRAME_TYPE_CONFIGURATION);
        } else if (type == RtmpPacker.KEY_FRAME) {
            frame = new Frame(video, type, Frame.FRAME_TYPE_KEY_FRAME);
        } else {
            frame = new Frame(video, type, Frame.FRAME_TYPE_INTER_FRAME);
        }
        mSendQueue.putFrame(frame);
    }

    public void stop() {
        closeStream();
        if (readThread != null) {
            readThread.setOnReadListener(null);
            readThread.shutdown();
        }
        if (writeThread != null) {
            writeThread.setWriteListener(null);
            writeThread.shutdown();
        }
        clearSocket();
        currentStreamId = -1;
        transactionIdCounter = 0;
        state = State.INIT;
    }

    private ConnectData parseRtmpUrl(String url) {
        ConnectData connectData = null;
        Matcher matcher = rtmpUrlPattern.matcher(url);
        if (matcher.matches()) {
            connectData = new ConnectData();
            connectData.tcUrl = url.substring(0, url.lastIndexOf('/'));
            connectData.swfUrl = "";
            connectData.pageUrl = "";
            connectData.host = matcher.group(1);
            String portStr = matcher.group(3);
            connectData.port = portStr != null ? Integer.parseInt(portStr) : 1935;
            connectData.appName = matcher.group(4);
            connectData.streamName = matcher.group(6);
        }
        return connectData;
    }

    private void handshake(InputStream in, OutputStream out) throws IOException {
        Handshake handshake = new Handshake();
        handshake.writeC0(out);
        handshake.writeC1(out); // Write C1 without waiting for S0
        out.flush();
        handshake.readS0(in);
        handshake.readS1(in);
        handshake.writeC2(out);
        handshake.readS2(in);
    }

    private void clearSocket() {
        if (socket != null && socket.isConnected()) {
            try {
                socket.close();
                socket = null;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void rtmpConnect() {
        SessionInfo.markSessionTimestampTx();
        Command invoke = new Command("connect", ++transactionIdCounter);

        AmfObject args = new AmfObject();
        args.setProperty("app", connectData.appName);
        args.setProperty("flashVer", "LNX 11,2,202,233"); // Flash player OS: Linux, version: 11.2.202.233
        args.setProperty("swfUrl", connectData.swfUrl);
        args.setProperty("tcUrl", connectData.tcUrl);
        args.setProperty("fpad", false);
        args.setProperty("capabilities", 239);
        args.setProperty("audioCodecs", 3575);
        args.setProperty("videoCodecs", 252);
        args.setProperty("videoFunction", 1);
        args.setProperty("pageUrl", connectData.pageUrl);
        args.setProperty("objectEncoding", 0);
        invoke.addData(args);
        Frame frame = new Frame(invoke, RtmpPacker.CONFIGRATION, Frame.FRAME_TYPE_CONFIGURATION);
        mSendQueue.putFrame(frame);
        state = State.CONNECTING;
    }

    private void handleRxCommandInvoke(Command command) {
        String commandName = command.getCommandName();
        switch (commandName) {
            case "_result": {
                String method = sessionInfo.takeInvokedCommand(command.getTransactionId());
                Log.d(TAG, "Got result for invoked method: " + method);
                if ("connect".equals(method)) {
                    if (listener != null) {
                        listener.onRtmpConnectSuccess();
                    }
                    createStream();
                } else if ("createStream".equals(method)) {
                    currentStreamId = (int) ((AmfNumber) command.getData().get(1)).getValue();
                    if (listener != null) {
                        listener.onCreateStreamSuccess();
                    }
                    fmlePublish();
                }
                break;
            }
            case "_error": {
                String method = sessionInfo.takeInvokedCommand(command.getTransactionId());
                Log.d(TAG, "Got error for invoked method: " + method);
                if ("connect".equals(method)) {
                    stop();
                    if (listener != null) {
                        listener.onRtmpConnectFail();
                    }
                } else if ("createStream".equals(method)) {
                    stop();
                    if (listener != null) {
                        listener.onCreateStreamFail();
                    }
                }
                break;
            }
            case "onStatus":
                String code = ((AmfString) ((AmfObject) command.getData().get(1)).getProperty("code")).getValue();
                if (code.equals("NetStream.Publish.Start")) {
                    Log.d(TAG, "Got publish start success");
                    state = State.LIVING;
                    if (listener != null) {
                        listener.onPublishSuccess();
                    }
                    onMetaData();
                    // We can now publish AV data
                    publishPermitted = true;
                } else {
                    Log.d(TAG, "Got publish start fail");
                    stop();
                    if (listener != null) {
                        listener.onPublishFail();
                    }
                }
                break;
            default:
                Log.d(TAG, "Got Command result: " + commandName);
                break;
        }
    }

    private void createStream() {
        state = State.CREATE_STREAM;
        Log.d(TAG, "createStream(): Sending releaseStream command...");
        // transactionId == 2
        Command releaseStream = new Command("releaseStream", ++transactionIdCounter);
        releaseStream.getChunkHeader().setChunkStreamId(SessionInfo.RTMP_STREAM_CHANNEL);
        releaseStream.addData(new AmfNull());  // command object: null for "createStream"
        releaseStream.addData(connectData.streamName);  // command object: null for "releaseStream"
        Frame frame1 = new Frame(releaseStream, RtmpPacker.CONFIGRATION, Frame.FRAME_TYPE_CONFIGURATION);
        mSendQueue.putFrame(frame1);

        Log.d(TAG, "createStream(): Sending FCPublish command...");
        // transactionId == 3
        Command FCPublish = new Command("FCPublish", ++transactionIdCounter);
        FCPublish.getChunkHeader().setChunkStreamId(SessionInfo.RTMP_STREAM_CHANNEL);
        FCPublish.addData(new AmfNull());  // command object: null for "FCPublish"
        FCPublish.addData(connectData.streamName);
        Frame frame2 = new Frame(FCPublish, RtmpPacker.CONFIGRATION, Frame.FRAME_TYPE_CONFIGURATION);
        mSendQueue.putFrame(frame2);

        Log.d(TAG, "createStream(): Sending createStream command...");
        // transactionId == 4
        Command createStream = new Command("createStream", ++transactionIdCounter);
        createStream.addData(new AmfNull());  // command object: null for "createStream"
        Frame frame3 = new Frame(createStream, RtmpPacker.CONFIGRATION, Frame.FRAME_TYPE_CONFIGURATION);
        mSendQueue.putFrame(frame3);
    }

    private void fmlePublish() {
        if (currentStreamId == -1 || connectData == null) {
            return;
        }
        state = State.PUBLISHING;
        Log.d(TAG, "fmlePublish(): Sending publish command...");
        // transactionId == 0
        Command publish = new Command("publish", 0);
        publish.getChunkHeader().setChunkStreamId(SessionInfo.RTMP_STREAM_CHANNEL);
        publish.getChunkHeader().setMessageStreamId(currentStreamId);
        publish.addData(new AmfNull());  // command object: null for "publish"
        publish.addData(connectData.streamName);
        publish.addData("live");
        Frame frame = new Frame(publish, RtmpPacker.CONFIGRATION, Frame.FRAME_TYPE_CONFIGURATION);
        mSendQueue.putFrame(frame);
    }

    private void onMetaData() {
        if (currentStreamId == -1) {
            return;
        }

        Log.d(TAG, "onMetaData(): Sending empty onMetaData...");
        Data metadata = new Data("@setDataFrame");
        metadata.getChunkHeader().setMessageStreamId(currentStreamId);
        metadata.addData("onMetaData");
        AmfMap ecmaArray = new AmfMap();
        ecmaArray.setProperty("duration", 0);
        ecmaArray.setProperty("width", videoWidth);
        ecmaArray.setProperty("height", videoHeight);
        ecmaArray.setProperty("videodatarate", 0);
        ecmaArray.setProperty("framerate", 0);
        ecmaArray.setProperty("audiodatarate", 0);
        ecmaArray.setProperty("audiosamplerate", audioSampleRate);
        ecmaArray.setProperty("audiosamplesize", audioSampleSize);
        ecmaArray.setProperty("stereo", isAudioStereo);
        ecmaArray.setProperty("filesize", 0);
        metadata.addData(ecmaArray);
        Frame frame = new Frame(metadata, RtmpPacker.CONFIGRATION, Frame.FRAME_TYPE_CONFIGURATION);
        mSendQueue.putFrame(frame);
    }

    public void setVideoParams(int width, int height) {
        videoWidth = width;
        videoHeight = height;
    }

    public void setAudioParams(int sampleRate, int sampleSize, boolean isStereo) {
        audioSampleRate = sampleRate;
        audioSampleSize = sampleSize;
        isAudioStereo = isStereo;
    }

    private void closeStream() throws IllegalStateException {
        if (currentStreamId == -1) {
            return;
        }
        if (!publishPermitted) {
            return;
        }
        Log.d(TAG, "closeStream(): setting current stream ID to -1");
        Command closeStream = new Command("closeStream", 0);
        closeStream.getChunkHeader().setChunkStreamId(SessionInfo.RTMP_STREAM_CHANNEL);
        closeStream.getChunkHeader().setMessageStreamId(currentStreamId);
        closeStream.addData(new AmfNull());
        Frame frame = new Frame(closeStream, RtmpPacker.CONFIGRATION, Frame.FRAME_TYPE_CONFIGURATION);
        mSendQueue.putFrame(frame);
    }


    @Override
    public void onChunkRead(Chunk chunk) {
        ChunkHeader chunkHeader = chunk.getChunkHeader();
        MessageType messageType = chunkHeader.getMessageType();
        switch (messageType) {
            case ABORT:
                readThread.clearStoredChunks(((Abort) chunk).getChunkStreamId());
                break;
            case USER_CONTROL_MESSAGE:
                UserControl ping = (UserControl) chunk;
                if (ping.getType() == UserControl.Type.PING_REQUEST) {
                    Log.d(TAG, "Sending PONG reply..");
                    UserControl pong = new UserControl();
                    pong.setType(UserControl.Type.PONG_REPLY);
                    pong.setEventData(ping.getEventData()[0]);
                    Frame frame = new Frame(pong, RtmpPacker.CONFIGRATION, Frame.FRAME_TYPE_CONFIGURATION);
                    mSendQueue.putFrame(frame);
                } else if (ping.getType() == UserControl.Type.STREAM_EOF) {
                    Log.d(TAG, "Stream EOF reached");
                }
                break;
            case WINDOW_ACKNOWLEDGEMENT_SIZE:
                WindowAckSize windowAckSize = (WindowAckSize) chunk;
                int size = windowAckSize.getAcknowledgementWindowSize();
                Log.d(TAG, "Setting acknowledgement window size: " + size);
                sessionInfo.setAcknowledgmentWindowSize(size);
                // Set socket option
                try {
                    socket.setSendBufferSize(size);
                } catch (SocketException e) {
                    e.printStackTrace();
                }
                break;
            case SET_PEER_BANDWIDTH:
                int acknowledgementWindowsize = sessionInfo.getAcknowledgementWindowSize();
                Log.d(TAG, "Send acknowledgement window size: " + acknowledgementWindowsize);
                Chunk setPeerBandwidth = new WindowAckSize(acknowledgementWindowsize);
                Frame frame = new Frame(setPeerBandwidth, RtmpPacker.CONFIGRATION, Frame.FRAME_TYPE_CONFIGURATION);
                mSendQueue.putFrame(frame);
                break;
            case COMMAND_AMF0:
                handleRxCommandInvoke((Command) chunk);
                break;
            default:
                Log.w(TAG, "Not handling unimplemented/unknown packet of type: " + chunkHeader.getMessageType());
                break;
        }
    }

    @Override
    public void onStreamEnd() {
        stop();
        if (listener != null) {
            listener.onStreamEnd();
        }
    }

    @Override
    public void onDisconnect() {
        stop();
        if (listener != null) {
            listener.onSocketDisconnect();
        }
    }

} public interface ISendQueue {
    void start();
    void stop();
    void setBufferSize(int size);
    void putFrame(Frame frame);
    Frame takeFrame();
    void setSendQueueListener(SendQueueListener listener);
} public class NormalSendQueue implements ISendQueue {
    private static final int NORMAL_FRAME_BUFFER_SIZE = 800;
    private ArrayBlockingQueue<Frame> mFrameBuffer;
    private int mFullQueueCount = NORMAL_FRAME_BUFFER_SIZE;
    private AtomicInteger mTotalFrameCount = new AtomicInteger(0);  //总个数
    private AtomicInteger mGiveUpFrameCount = new AtomicInteger(0);  //总个数

    private AtomicInteger mInFrameCount = new AtomicInteger(0);  //进入总个数
    private AtomicInteger mOutFrameCount = new AtomicInteger(0);  //输出总个数
    private volatile boolean mScanFlag;
    private SendQueueListener mSendQueueListener;

    public NormalSendQueue() {
        mFrameBuffer = new ArrayBlockingQueue<>(mFullQueueCount, true);
    }

    @Override
    public void start() {
        mScanFlag = true;
        ScanThread mScanThread = new ScanThread();
        mScanThread.start();
    }

    @Override
    public void stop() {
        mScanFlag = false;
        mInFrameCount.set(0);
        mOutFrameCount.set(0);
        mTotalFrameCount.set(0);
        mGiveUpFrameCount.set(0);
        mFrameBuffer.clear();
    }

    public void setSendQueueListener(SendQueueListener listener) {
        mSendQueueListener = listener;
    }

    @Override
    public void setBufferSize(int size) {
        mFullQueueCount = size;
    }

    @Override
    public void putFrame(Frame frame) {
        if (mFrameBuffer == null) {
            return;
        }
        abandonData();
        try {
            mFrameBuffer.put(frame);
            mInFrameCount.getAndIncrement();
            mTotalFrameCount.getAndIncrement();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Frame takeFrame() {
        if (mFrameBuffer == null) {
            return null;
        }
        Frame frame = null;
        try {
            frame = mFrameBuffer.take();
            mOutFrameCount.getAndIncrement();
            mTotalFrameCount.getAndDecrement();
        } catch (InterruptedException e) {
            //do nothing
        }
        return frame;
    }

    private void abandonData() {
        if (mTotalFrameCount.get() >= (mFullQueueCount / 2)) {
            //从队列头部开始搜索，删除最早发现的连续P帧
            boolean pFrameDelete = false;
            boolean start = false;
            for (Frame frame : mFrameBuffer) {
                if (frame.frameType == FRAME_TYPE_INTER_FRAME) {
                    start = true;
                }
                if (start) {
                    if (frame.frameType == FRAME_TYPE_INTER_FRAME) {
                        mFrameBuffer.remove(frame);
                        mTotalFrameCount.getAndDecrement();
                        mGiveUpFrameCount.getAndIncrement();
                        pFrameDelete = true;
                    } else if (frame.frameType == FRAME_TYPE_KEY_FRAME) {
                        break;
                    }
                }
            }
            boolean kFrameDelete = false;
            //从队列头部开始搜索，删除最早发现的I帧
            if (!pFrameDelete) {
                for (Frame frame : mFrameBuffer) {
                    if (frame.frameType == FRAME_TYPE_KEY_FRAME) {
                        mFrameBuffer.remove(frame);
                        mTotalFrameCount.getAndDecrement();
                        mGiveUpFrameCount.getAndIncrement();
                        kFrameDelete = true;
                        break;
                    }
                }
            }
            //从队列头部开始搜索，删除音频
            if (!pFrameDelete && !kFrameDelete) {
                for (Frame frame : mFrameBuffer) {
                    if (frame.frameType == FRAME_TYPE_AUDIO) {
                        mFrameBuffer.remove(frame);
                        mTotalFrameCount.getAndDecrement();
                        mGiveUpFrameCount.getAndIncrement();
                        break;
                    }
                }
            }
        }
    }

    private class ScanThread extends Thread {
        private static final int SCAN_MAX_TIME = 6;
        private int mCurrentScanTime = 0;
        private ArrayList<ScanSnapShot> mScanSnapShotList = new ArrayList<>();

        @Override
        public void run() {
            while (mScanFlag) {
                //达到仲裁次数了
                if (mCurrentScanTime == SCAN_MAX_TIME) {
                    int averageDif = 0;
                    int negativeCounter = 0;
                    String strLog = "";
                    for (int i = 0; i < SCAN_MAX_TIME; i++) {
                        int dif = mScanSnapShotList.get(i).outCount - mScanSnapShotList.get(i).inCount;
                        if (dif < 0) {
                            negativeCounter++;
                        }
                        averageDif += dif;
                        strLog = strLog + String.format(Locale.CHINA, "n%d:%d  ", i, dif);
                    }
                    Lg.d(strLog);
                    if (negativeCounter >= 4 || averageDif < -100) {
                        //坏
                        Lg.d("Bad Send Speed.");
                        if (mSendQueueListener != null) {
                            mSendQueueListener.bad();
                        }
                    } else {
                        //好
                        Lg.d("Good Send Speed.");
                        if (mSendQueueListener != null) {
                            mSendQueueListener.good();
                        }
                    }
                    //清空
                    mScanSnapShotList.clear();
                    mCurrentScanTime = 0;
                }
                mScanSnapShotList.add(new ScanSnapShot(mInFrameCount.get(), mOutFrameCount.get()));
                mInFrameCount.set(0);
                mOutFrameCount.set(0);
                mCurrentScanTime++;
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private class ScanSnapShot {
        int inCount;
        int outCount;

        ScanSnapShot(int inCount, int outCount) {
            this.inCount = inCount;
            this.outCount = outCount;
        }
    }
} public interface SendQueueListener {
    void good();
    void bad();
} public class GlobalAsyncThread {

    private static GlobalAsyncThread sInstance = new GlobalAsyncThread();

    public static GlobalAsyncThread get() {
        return sInstance;
    }

    public static void post(final Runnable runnable){
        sInstance.posts(runnable);
    }

    public static void start(){
        sInstance.starts();
    }

    public static void stop(){
        sInstance.stops();
    }

    private GlobalAsyncThread() {
    }

    private HandlerThread workThread;
    private Handler workHandler;

    private void posts(final Runnable runnable) {
        if(workHandler != null){
            workHandler.post(runnable);
        }else{
            starts();
            workHandler.post(runnable);
        }
    }

    private void starts(){
        stops();
        workThread = new HandlerThread(
                GlobalAsyncThread.class.getSimpleName(),
                Process.THREAD_PRIORITY_BACKGROUND);
        workThread.start();
        workHandler  = new Handler(workThread.getLooper());
    }

    private void stops(){
        if(workHandler != null)workHandler.removeCallbacksAndMessages(null);
        if(workThread != null)workThread.quit();
    }

} public class Options {

    public static Options ins = new Options();

    public static Options getInstance() {
        return ins;
    }

    private Options() {
    }

    //网络请求配置
    public String protocol = "rtmp";
    public String service = "192.168.43.174";
    public int port = 1935;
    public String app = "live";
    public String stream = "ta";
    public boolean isProvider = true;

    //视频录制配置
    public int fps = DEFAULT_FPS;
    public int bufferSize = DEFAULT_BUFFER_SIZE;
    public boolean front = true;

    //configuration
    public Audio audio = new Audio();
    public Camera camera = new Camera();
    public Video video = new Video();


    //默认配置
    private static final int DEFAULT_BUFFER_SIZE = 1024;
    //audio
    public static final int DEFAULT_FREQUENCY = 44100;
    public static final int DEFAULT_MAX_BPS = 64;
    public static final int DEFAULT_MIN_BPS = 32;
    public static final int DEFAULT_ADTS = 0;
    public static final String DEFAULT_MIME = "audio/mp4a-latm";
    public static final int DEFAULT_AUDIO_ENCODING = AudioFormat.ENCODING_PCM_16BIT;
    public static final int DEFAULT_AAC_PROFILE = MediaCodecInfo.CodecProfileLevel.AACObjectLC;
    public static final int DEFAULT_CHANNEL_COUNT = 1;
    public static final boolean DEFAULT_AEC = false;
    //camera
    public static final int DEFAULT_HEIGHT = 1280;
    public static final int DEFAULT_WIDTH = 720;
    public static final int DEFAULT_FPS = 15;
    public static final Facing DEFAULT_FACING = Facing.FRONT;
    public static final Orientation DEFAULT_ORIENTATION = Orientation.PORTRAIT;
    public static final FocusMode DEFAULT_FOCUSMODE = FocusMode.AUTO;
    //video
    public static final int VIDEO_DEFAULT_HEIGHT = 640;
    public static final int VIDEO_DEFAULT_WIDTH = 360;
    public static final int VIDEO_DEFAULT_FPS = 15;
    public static final int VIDEO_DEFAULT_MAX_BPS = 1300;
    public static final int VIDEO_DEFAULT_MIN_BPS = 400;
    public static final int VIDEO_DEFAULT_IFI = 2;
    public static final String VIDEO_DEFAULT_MIME = "video/avc";

    public enum  Facing {
        FRONT,
        BACK
    }

    public enum  Orientation {
        LANDSCAPE,
        PORTRAIT
    }

    public enum  FocusMode {
        AUTO,
        TOUCH
    }

    public void setFacing(Facing facing) {
        camera.facing = facing;
    }

    public void setOrientation(Orientation orientation) {
        camera.orientation = orientation;
    }

    public void setSize(int width, int height) {
        video.width = width;
        video.height = height;
    }

    @Override
    public String toString() {
        return protocol + "://" + service + ":" + port + "/" + app + "/" + stream;
    }

    public static class Audio{
        public int minBps = DEFAULT_MIN_BPS;
        public int maxBps = DEFAULT_MAX_BPS;
        public int frequency = DEFAULT_FREQUENCY;
        public int encoding = DEFAULT_AUDIO_ENCODING;
        public int channelCount = DEFAULT_CHANNEL_COUNT;
        public int adts = DEFAULT_ADTS;
        public String mime = DEFAULT_MIME;
        public int aacProfile = DEFAULT_AAC_PROFILE;
        public boolean aec = DEFAULT_AEC;
    }

    public static class Camera{
        public int height = DEFAULT_HEIGHT;
        public int width = DEFAULT_WIDTH;
        public int fps = DEFAULT_FPS;
        public Facing facing = DEFAULT_FACING;
        public Orientation orientation = DEFAULT_ORIENTATION;
        public FocusMode focusMode = DEFAULT_FOCUSMODE;
    }

    public static class Video{
        public int height = VIDEO_DEFAULT_HEIGHT;
        public int width = VIDEO_DEFAULT_WIDTH;
        public int minBps = VIDEO_DEFAULT_MIN_BPS;
        public int maxBps = VIDEO_DEFAULT_MAX_BPS;
        public int fps = VIDEO_DEFAULT_FPS;
        public int ifi = VIDEO_DEFAULT_IFI;
        public String mime = VIDEO_DEFAULT_MIME;
    }
} @SuppressWarnings("unused")
public class WeakHandler {
    private final Handler.Callback mCallback; // hard reference to Callback. We need to keep callback in memory
    private final ExecHandler mExec;
    private Lock mLock = new ReentrantLock();
    @SuppressWarnings("ConstantConditions")
    final ChainedRef mRunnables = new ChainedRef(mLock, null);

    /**
     * Default constructor associates this handler with the {@link Looper} for the
     * current thread.
     *
     * If this thread does not have a looper, this handler won't be able to receive messages
     * so an exception is thrown.
     */
    public WeakHandler() {
        mCallback = null;
        mExec = new ExecHandler();
    }

    /**
     * Constructor associates this handler with the {@link Looper} for the
     * current thread and takes a callback interface in which you can handle
     * messages.
     *
     * If this thread does not have a looper, this handler won't be able to receive messages
     * so an exception is thrown.
     *
     * @param callback The callback interface in which to handle messages, or null.
     */
    public WeakHandler(Handler.Callback callback) {
        mCallback = callback; // Hard referencing body
        mExec = new ExecHandler(new WeakReference<>(callback)); // Weak referencing inside ExecHandler
    }

    /**
     * Use the provided {@link Looper} instead of the default one.
     *
     * @param looper The looper, must not be null.
     */
    public WeakHandler(Looper looper) {
        mCallback = null;
        mExec = new ExecHandler(looper);
    }

    /**
     * Use the provided {@link Looper} instead of the default one and take a callback
     * interface in which to handle messages.
     *
     * @param looper The looper, must not be null.
     * @param callback The callback interface in which to handle messages, or null.
     */
    public WeakHandler(Looper looper, Handler.Callback callback) {
        mCallback = callback;
        mExec = new ExecHandler(looper, new WeakReference<>(callback));
    }

    /**
     * Causes the Runnable r to be added to the message queue.
     * The runnable will be run on the thread to which this handler is
     * attached.
     *
     * @param r The Runnable that will be executed.
     *
     * @return Returns true if the Runnable was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.
     */
    public final boolean post(Runnable r) {
        return mExec.post(wrapRunnable(r));
    }

    /**
     * Causes the Runnable r to be added to the message queue, to be run
     * at a specific time given by <var>uptimeMillis</var>.
     * <b>The time-base is {@link android.os.SystemClock#uptimeMillis}.</b>
     * The runnable will be run on the thread to which this handler is attached.
     *
     * @param r The Runnable that will be executed.
     * @param uptimeMillis The absolute time at which the callback should run,
     *         using the {@link android.os.SystemClock#uptimeMillis} time-base.
     *
     * @return Returns true if the Runnable was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.  Note that a
     *         result of true does not mean the Runnable will be processed -- if
     *         the looper is quit before the delivery time of the message
     *         occurs then the message will be dropped.
     */
    public final boolean postAtTime(Runnable r, long uptimeMillis) {
        return mExec.postAtTime(wrapRunnable(r), uptimeMillis);
    }

    /**
     * Causes the Runnable r to be added to the message queue, to be run
     * at a specific time given by <var>uptimeMillis</var>.
     * <b>The time-base is {@link android.os.SystemClock#uptimeMillis}.</b>
     * The runnable will be run on the thread to which this handler is attached.
     *
     * @param r The Runnable that will be executed.
     * @param uptimeMillis The absolute time at which the callback should run,
     *         using the {@link android.os.SystemClock#uptimeMillis} time-base.
     *
     * @return Returns true if the Runnable was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.  Note that a
     *         result of true does not mean the Runnable will be processed -- if
     *         the looper is quit before the delivery time of the message
     *         occurs then the message will be dropped.
     *
     * @see android.os.SystemClock#uptimeMillis
     */
    public final boolean postAtTime(Runnable r, Object token, long uptimeMillis) {
        return mExec.postAtTime(wrapRunnable(r), token, uptimeMillis);
    }

    /**
     * Causes the Runnable r to be added to the message queue, to be run
     * after the specified amount of time elapses.
     * The runnable will be run on the thread to which this handler
     * is attached.
     *
     * @param r The Runnable that will be executed.
     * @param delayMillis The delay (in milliseconds) until the Runnable
     *        will be executed.
     *
     * @return Returns true if the Runnable was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.  Note that a
     *         result of true does not mean the Runnable will be processed --
     *         if the looper is quit before the delivery time of the message
     *         occurs then the message will be dropped.
     */
    public final boolean postDelayed(Runnable r, long delayMillis) {
        return mExec.postDelayed(wrapRunnable(r), delayMillis);
    }

    /**
     * Posts a message to an object that implements Runnable.
     * Causes the Runnable r to executed on the next iteration through the
     * message queue. The runnable will be run on the thread to which this
     * handler is attached.
     * <b>This method is only for use in very special circumstances -- it
     * can easily starve the message queue, cause ordering problems, or have
     * other unexpected side-effects.</b>
     *
     * @param r The Runnable that will be executed.
     *
     * @return Returns true if the message was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.
     */
    public final boolean postAtFrontOfQueue(Runnable r) {
        return mExec.postAtFrontOfQueue(wrapRunnable(r));
    }

    /**
     * Remove any pending posts of Runnable r that are in the message queue.
     */
    public final void removeCallbacks(Runnable r) {
        final WeakRunnable runnable = mRunnables.remove(r);
        if (runnable != null) {
            mExec.removeCallbacks(runnable);
        }
    }

    /**
     * Remove any pending posts of Runnable <var>r</var> with Object
     * <var>token</var> that are in the message queue.  If <var>token</var> is null,
     * all callbacks will be removed.
     */
    public final void removeCallbacks(Runnable r, Object token) {
        final WeakRunnable runnable = mRunnables.remove(r);
        if (runnable != null) {
            mExec.removeCallbacks(runnable, token);
        }
    }

    /**
     * Pushes a message onto the end of the message queue after all pending messages
     * before the current time. It will be received in callback,
     * in the thread attached to this handler.
     *
     * @return Returns true if the message was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.
     */
    public final boolean sendMessage(Message msg) {
        return mExec.sendMessage(msg);
    }

    /**
     * Sends a Message containing only the what value.
     *
     * @return Returns true if the message was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.
     */
    public final boolean sendEmptyMessage(int what) {
        return mExec.sendEmptyMessage(what);
    }

    /**
     * Sends a Message containing only the what value, to be delivered
     * after the specified amount of time elapses.
     * @see #sendMessageDelayed(android.os.Message, long)
     *
     * @return Returns true if the message was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.
     */
    public final boolean sendEmptyMessageDelayed(int what, long delayMillis) {
        return mExec.sendEmptyMessageDelayed(what, delayMillis);
    }

    /**
     * Sends a Message containing only the what value, to be delivered
     * at a specific time.
     * @see #sendMessageAtTime(android.os.Message, long)
     *
     * @return Returns true if the message was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.
     */
    public final boolean sendEmptyMessageAtTime(int what, long uptimeMillis) {
        return mExec.sendEmptyMessageAtTime(what, uptimeMillis);
    }

    /**
     * Enqueue a message into the message queue after all pending messages
     * before (current time + delayMillis). You will receive it in
     * callback, in the thread attached to this handler.
     *
     * @return Returns true if the message was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.  Note that a
     *         result of true does not mean the message will be processed -- if
     *         the looper is quit before the delivery time of the message
     *         occurs then the message will be dropped.
     */
    public final boolean sendMessageDelayed(Message msg, long delayMillis) {
        return mExec.sendMessageDelayed(msg, delayMillis);
    }

    /**
     * Enqueue a message into the message queue after all pending messages
     * before the absolute time (in milliseconds) <var>uptimeMillis</var>.
     * <b>The time-base is {@link android.os.SystemClock#uptimeMillis}.</b>
     * You will receive it in callback, in the thread attached
     * to this handler.
     *
     * @param uptimeMillis The absolute time at which the message should be
     *         delivered, using the
     *         {@link android.os.SystemClock#uptimeMillis} time-base.
     *
     * @return Returns true if the message was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.  Note that a
     *         result of true does not mean the message will be processed -- if
     *         the looper is quit before the delivery time of the message
     *         occurs then the message will be dropped.
     */
    public boolean sendMessageAtTime(Message msg, long uptimeMillis) {
        return mExec.sendMessageAtTime(msg, uptimeMillis);
    }

    /**
     * Enqueue a message at the front of the message queue, to be processed on
     * the next iteration of the message loop.  You will receive it in
     * callback, in the thread attached to this handler.
     * <b>This method is only for use in very special circumstances -- it
     * can easily starve the message queue, cause ordering problems, or have
     * other unexpected side-effects.</b>
     *
     * @return Returns true if the message was successfully placed in to the
     *         message queue.  Returns false on failure, usually because the
     *         looper processing the message queue is exiting.
     */
    public final boolean sendMessageAtFrontOfQueue(Message msg) {
        return mExec.sendMessageAtFrontOfQueue(msg);
    }

    /**
     * Remove any pending posts of messages with code 'what' that are in the
     * message queue.
     */
    public final void removeMessages(int what) {
        mExec.removeMessages(what);
    }

    /**
     * Remove any pending posts of messages with code 'what' and whose obj is
     * 'object' that are in the message queue.  If <var>object</var> is null,
     * all messages will be removed.
     */
    public final void removeMessages(int what, Object object) {
        mExec.removeMessages(what, object);
    }

    /**
     * Remove any pending posts of callbacks and sent messages whose
     * <var>obj</var> is <var>token</var>.  If <var>token</var> is null,
     * all callbacks and messages will be removed.
     */
    public final void removeCallbacksAndMessages(Object token) {
        mExec.removeCallbacksAndMessages(token);
    }

    /**
     * Check if there are any pending posts of messages with code 'what' in
     * the message queue.
     */
    public final boolean hasMessages(int what) {
        return mExec.hasMessages(what);
    }

    /**
     * Check if there are any pending posts of messages with code 'what' and
     * whose obj is 'object' in the message queue.
     */
    public final boolean hasMessages(int what, Object object) {
        return mExec.hasMessages(what, object);
    }

    public final Looper getLooper() {
        return mExec.getLooper();
    }

    private WeakRunnable wrapRunnable(Runnable r) {
        //noinspection ConstantConditions
        if (r == null) {
            throw new NullPointerException("Runnable can't be null");
        }
        final ChainedRef hardRef = new ChainedRef(mLock, r);
        mRunnables.insertAfter(hardRef);
        return hardRef.wrapper;
    }

    private static class ExecHandler extends Handler {
        private final WeakReference<Handler.Callback> mCallback;

        ExecHandler() {
            mCallback = null;
        }

        ExecHandler(WeakReference<Handler.Callback> callback) {
            mCallback = callback;
        }

        ExecHandler(Looper looper) {
            super(looper);
            mCallback = null;
        }

        ExecHandler(Looper looper, WeakReference<Handler.Callback> callback) {
            super(looper);
            mCallback = callback;
        }

        @Override
        public void handleMessage(Message msg) {
            if (mCallback == null) {
                return;
            }
            final Handler.Callback callback = mCallback.get();
            if (callback == null) { // Already disposed
                return;
            }
            callback.handleMessage(msg);
        }
    }

    static class WeakRunnable implements Runnable {
        private final WeakReference<Runnable> mDelegate;
        private final WeakReference<ChainedRef> mReference;

        WeakRunnable(WeakReference<Runnable> delegate, WeakReference<ChainedRef> reference) {
            mDelegate = delegate;
            mReference = reference;
        }

        @Override
        public void run() {
            final Runnable delegate = mDelegate.get();
            final ChainedRef reference = mReference.get();
            if (reference != null) {
                reference.remove();
            }
            if (delegate != null) {
                delegate.run();
            }
        }
    }

    static class ChainedRef {
        ChainedRef next;
        ChainedRef prev;
        final Runnable runnable;
        final WeakRunnable wrapper;

        Lock lock;

        public ChainedRef(Lock lock, Runnable r) {
            this.runnable = r;
            this.lock = lock;
            this.wrapper = new WeakRunnable(new WeakReference<>(r), new WeakReference<>(this));
        }

        public WeakRunnable remove() {
            lock.lock();
            try {
                if (prev != null) {
                    prev.next = next;
                }
                if (next != null) {
                    next.prev = prev;
                }
                prev = null;
                next = null;
            } finally {
                lock.unlock();
            }
            return wrapper;
        }

        public void insertAfter(ChainedRef candidate) {
            lock.lock();
            try {
                if (this.next != null) {
                    this.next.prev = candidate;
                }

                candidate.next = this.next;
                this.next = candidate;
                candidate.prev = this;
            } finally {
                lock.unlock();
            }
        }

        public WeakRunnable remove(Runnable obj) {
            lock.lock();
            try {
                ChainedRef curr = this.next; // Skipping head
                while (curr != null) {
                    if (curr.runnable == obj) { // We do comparison exactly how Handler does inside
                        return curr.remove();
                    }
                    curr = curr.next;
                }
            } finally {
                lock.unlock();
            }
            return null;
        }
    }
}
import org.yeshen.video.librtmp.core.IRtmpConnection;
import org.yeshen.video.librtmp.unstable.net.packer.rtmp.RtmpPacker;
import org.yeshen.video.librtmp.unstable.net.sender.Sender;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.io.RtmpConnectListener;
import org.yeshen.video.librtmp.unstable.net.sender.rtmp.io.RtmpConnection;
import org.yeshen.video.librtmp.unstable.net.sender.sendqueue.ISendQueue;
import org.yeshen.video.librtmp.unstable.net.sender.sendqueue.NormalSendQueue;
import org.yeshen.video.librtmp.unstable.net.sender.sendqueue.SendQueueListener;
import org.yeshen.video.librtmp.unstable.tools.GlobalAsyncThread;
import org.yeshen.video.librtmp.unstable.tools.Options;
import org.yeshen.video.librtmp.unstable.tools.WeakHandler;
package org.yeshen.video.librtmp.unstable.net.sender.rtmp;


/**
 * @Title: RtmpSender
 * @Package org.yeshen.video.librtmp.afix.net.sender.rtmp
 * @Description:
 * @Author Jim
 * @Date 16/9/21
 * @Time 上午11:16
 * @Version
 */
public class RtmpSender implements Sender, SendQueueListener {

    public static RtmpSender getRtmpSender() {
        return new RtmpSender();
    }

    private IRtmpConnection rtmpConnection;
    private String mRtmpUrl;
    private OnSenderListener mListener;
    private WeakHandler mHandler = new WeakHandler();
    private ISendQueue mSendQueue = new NormalSendQueue();