package com.cooleshow.teacher.presenter.live;

import android.text.TextUtils;
import android.util.Log;

import com.cooleshow.base.data.net.BaseResponse;
import com.cooleshow.base.presenter.BasePresenter;
import com.cooleshow.base.rx.BaseObserver;
import com.cooleshow.base.utils.RequestBodyUtil;
import com.cooleshow.base.utils.Utils;
import com.cooleshow.teacher.api.APIService;
import com.cooleshow.teacher.bean.LiveRoomInfoBean;
import com.cooleshow.teacher.contract.LiveRoomContract;
import com.cooleshow.usercenter.helper.UserHelper;
import com.daya.live_teaching.common.ErrorCode;
import com.daya.live_teaching.common.ResultCallback;
import com.google.gson.Gson;
import com.rong.io.live.LiveRoomMsgConstants;
import com.rong.io.live.bean.ImUserState;
import com.rong.io.live.bean.User;
import com.rong.io.live.callback.ClickCallback;
import com.rong.io.live.callback.IRoomCallBack;
import com.rong.io.live.helper.LiveEventHelper;
import com.rong.io.live.helper.LiveRTCEngineInitHelper;
import com.rong.io.live.manager.RCChatRoomMessageManager;
import com.rong.io.live.message.RCChatJoinRoomMessage;
import com.rong.io.live.message.RCChatRoomMemberNumMessage;
import com.rong.io.live.message.RCChatroomLocationMessage;
import com.rong.io.live.message.RCOnSnappingUpMessage;
import com.rong.io.live.message.RCPauseLiveMessage;
import com.rong.io.live.message.RCUserAddLikeMessage;
import com.rong.io.live.message.RCUserSeatApplyMessage;
import com.rong.io.live.message.RCUserSeatResponseMessage;
import com.rong.io.live.message.RCUserSyncAddLikeCountMessage;

import java.util.ArrayList;
import java.util.List;

import cn.rongcloud.rtc.api.RCRTCEngine;
import cn.rongcloud.rtc.api.RCRTCRemoteUser;
import cn.rongcloud.rtc.api.RCRTCRoom;
import cn.rongcloud.rtc.api.callback.IRCRTCResultCallback;
import cn.rongcloud.rtc.api.callback.IRCRTCResultDataCallback;
import cn.rongcloud.rtc.api.callback.IRCRTCRoomEventsListener;
import cn.rongcloud.rtc.api.stream.RCRTCAudioInputStream;
import cn.rongcloud.rtc.api.stream.RCRTCCameraOutputStream;
import cn.rongcloud.rtc.api.stream.RCRTCInputStream;
import cn.rongcloud.rtc.api.stream.RCRTCLiveInfo;
import cn.rongcloud.rtc.api.stream.RCRTCOutputStream;
import cn.rongcloud.rtc.api.stream.RCRTCVideoInputStream;
import cn.rongcloud.rtc.api.stream.RCRTCVideoOutputStream;
import cn.rongcloud.rtc.api.stream.RCRTCVideoStreamConfig;
import cn.rongcloud.rtc.api.stream.RCRTCVideoView;
import cn.rongcloud.rtc.base.RCRTCLiveRole;
import cn.rongcloud.rtc.base.RCRTCMediaType;
import cn.rongcloud.rtc.base.RCRTCParamsType;
import cn.rongcloud.rtc.base.RTCErrorCode;
import cn.rongcloud.rtc.core.CameraVideoCapturer;
import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.core.Scheduler;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.observers.DisposableObserver;
import io.reactivex.rxjava3.schedulers.Schedulers;
import io.rong.imkit.IMCenter;
import io.rong.imlib.IRongCoreCallback;
import io.rong.imlib.RongIMClient;
import io.rong.imlib.model.Message;
import io.rong.imlib.model.MessageContent;
import okhttp3.RequestBody;

/**
 * Author by pq, Date on 2022/6/5.
 */
public class LiveRoomPresenter extends BasePresenter<LiveRoomContract.LiveRoomView> implements LiveRoomContract.Presenter {
    public boolean isInRoom = false;
    private LiveRoomInfoBean currentRoomInfo;
    private List<Disposable> disposablesManager = new ArrayList<>();//监听管理器
    private final IRCRTCRoomEventsListener mRoomEventsListener = new IRCRTCRoomEventsListener() {
        @Override
        public void onRemoteUserPublishResource(RCRTCRemoteUser rcrtcRemoteUser, List<RCRTCInputStream> list) {
            Log.i("pq", "onRemoteUserPublishResource");
//            subscribeAVStream();
        }

        @Override
        public void onRemoteUserMuteAudio(RCRTCRemoteUser rcrtcRemoteUser, RCRTCInputStream rcrtcInputStream, boolean b) {
            Log.i("pq", "onRemoteUserMuteAudio:" + b);
            if (getView() != null) {
//                getView().syncAudioStatus(b);
            }
        }

        @Override
        public void onRemoteUserMuteVideo(RCRTCRemoteUser rcrtcRemoteUser, RCRTCInputStream rcrtcInputStream, boolean b) {
        }

        @Override
        public void onRemoteUserUnpublishResource(RCRTCRemoteUser rcrtcRemoteUser, List<RCRTCInputStream> list) {
            Log.i("pq", "onRemoteUserUnpublishResource");
            if (getView() != null) {
//                getView().showCloseVideoView();
            }
        }

        @Override
        public void onUserJoined(RCRTCRemoteUser rcrtcRemoteUser) {
            //当房间中用户使用RCRTCLocalUser#switchToBroadcaster 方法上麦成功时
            // ，房间中观众会收到IRCRTCRoomEventsListener#onUserJoined 回调。
            if (null != getView()) {
//                getView().onUserJoinRoom(rcrtcRemoteUser);
            }
        }

        @Override
        public void onSwitchRole(String userId, RCRTCLiveRole role) {
            super.onSwitchRole(userId, role);
            //当房间内的用户使用 RCRTCLocalUser#switchToBroadcaster 或 RCRTCLocalUser#switchToAudience 方法上下麦时
            // ，同房间内的其他主播会收到 IRCRTCRoomEventsListener#onSwitchRole 回调。
            if (getView() != null) {
//                getView().onSwitchRole(userId, role);
            }
        }

        @Override
        public void onUserLeft(RCRTCRemoteUser rcrtcRemoteUser) {
            if (getView() != null) {
//                getView().onUserLeftRoomMic(rcrtcRemoteUser);
            }
        }

        @Override
        public void onUserOffline(RCRTCRemoteUser rcrtcRemoteUser) {
            if (getView() != null) {
//                getView().onUserOfflineRoomMic(rcrtcRemoteUser);
            }
        }

        @Override
        public void onPublishLiveStreams(List<RCRTCInputStream> list) {
            Log.i("pq", "onPublishLiveStreams");
        }

        @Override
        public void onUnpublishLiveStreams(List<RCRTCInputStream> list) {
            Log.i("pq", "onUnpublishLiveStreams");
            if (getView() != null) {
//                getView().showRestView();
            }
        }


        /**
         * 自己退出房间。 例如断网退出等
         * @param i 状态码
         */
        @Override
        public void onLeaveRoom(int i) {
            Log.i("pq", "onLeaveRoom:" + i);
        }
    };


    /**
     * init
     *
     * @param roomId
     * @param isCreate
     */
    public void init(String roomId, boolean isCreate) {
        isInRoom = TextUtils.equals(LiveEventHelper.getInstance().getRoomId(), roomId);
        RongIMClient.ConnectionStatusListener.ConnectionStatus currentConnectionStatus = RongIMClient.getInstance().getCurrentConnectionStatus();
        if (currentConnectionStatus == RongIMClient.ConnectionStatusListener.ConnectionStatus.CONNECTED) {
            Log.i("pq", "LiveRoomActivity init getRoomInfo");
            getRoomInfo(roomId);
        } else {
            connectIM();
        }
    }


    /**
     * 获取房间信息
     *
     * @param roomId
     */
    public void getRoomInfo(String roomId) {
        addSubscribe(create(APIService.class).getLiveRoomInfo(roomId), new BaseObserver<LiveRoomInfoBean>(getView()) {
            @Override
            protected void onSuccess(LiveRoomInfoBean data) {
                Log.i("pq", "getRoomInfo success" + data);
                currentRoomInfo = data;
                setObMessageListener(data.roomUid);
                if (getView() != null) {
                    getView().getRoomInfoSuccess(data);
                }
            }

            @Override
            public void onError(Throwable e) {
                super.onError(e);
                e.printStackTrace();
                if (getView() != null) {
                    getView().getRoomInfoError(e);
                }
            }
        });
    }

    /**
     * 初始化配置以及预览模式
     */
    public void openVideoPreview() {
        initPublishConfig();
        startVideoPreview();
    }

    /**
     * 切换摄像头
     */
    public void reverseCamera() {
        RCRTCEngine.getInstance().getDefaultVideoStream().switchCamera(new CameraVideoCapturer.CameraSwitchHandler() {
            @Override
            public void onCameraSwitchDone(boolean isFrontCamera) {
                // 切换成功
            }

            @Override
            public void onCameraSwitchError(String errorDescription) {
                // 切换失败
            }
        });
    }

    public void prepareJoinRoom(String roomId, boolean isCreate) {
        //先离开当前房间
        LiveEventHelper.getInstance().leaveRoom(new IRoomCallBack() {
            @Override
            public void onSuccess() {
                if (getView() != null && getView().getContentView() != null) {
                    getView().getContentView().postDelayed(new Runnable() {
                        @Override
                        public void run() {
                            joinRoom(roomId, isCreate);
                        }
                    }, 1000);
                }
            }

            @Override
            public void onError(int code, String message) {
                Log.i("pq", "leaveRoom onError:" + code + "-msg:" + message);
                joinRoom(roomId, isCreate);
            }
        });
    }

    /**
     * 取消发布流
     */
    public void cancelPublish(IRCRTCResultCallback resultCallback) {
        RCRTCRoom rcrtcRoom = LiveEventHelper.getInstance().getRtcRoom();
        if (rcrtcRoom == null) {
            return;
        }
        rcrtcRoom.getLocalUser().unpublishDefaultLiveStreams(resultCallback);
    }


    /**
     * 开启预览
     *
     * @param
     */
    private void startVideoPreview() {
        // 创建本地视频显示视图
        RCRTCVideoView rcrtcVideoView = new RCRTCVideoView(Utils.getApp().getApplicationContext());
        RCRTCCameraOutputStream rcrtcCameraOutputStream = RCRTCEngine.getInstance().getDefaultVideoStream();
        if (rcrtcCameraOutputStream == null) {
            return;
        }
        RCRTCEngine.getInstance().getDefaultVideoStream().setVideoView(rcrtcVideoView);
        if (getView() != null) {
            getView().addVideoPreview(rcrtcVideoView);
        }
        RCRTCEngine.getInstance().getDefaultVideoStream().startCamera(new IRCRTCResultDataCallback<Boolean>() {
            @Override
            public void onSuccess(Boolean data) {
                if (getView() != null) {
                    getView().openCameraSuccess(data);
                }
            }

            @Override
            public void onFailed(RTCErrorCode errorCode) {
                if (getView() != null) {
                    getView().openCameraError(errorCode.getReason());
                }
            }
        });
    }

    public void initPublishConfig() {
        try {
            RCRTCVideoStreamConfig config =
                    RCRTCVideoStreamConfig.Builder.create()
                            .setVideoFps(RCRTCParamsType.RCRTCVideoFps.Fps_30)
                            .setVideoResolution(RCRTCParamsType.RCRTCVideoResolution.RESOLUTION_1080_1920)
                            .build();
            //音乐教学场景。不建议在会议、语聊房、直播场景中使用 MUSIC_CLASSROOM，否则有出现回声问题的风险。
            RCRTCEngine.getInstance().getDefaultVideoStream().setVideoConfig(config);
            RCRTCEngine.getInstance().getDefaultAudioStream().setAudioQuality(RCRTCParamsType.AudioQuality.MUSIC_HIGH, RCRTCParamsType.AudioScenario.MUSIC_CHATROOM);
            RCRTCEngine.getInstance().getDefaultVideoStream().enableTinyStream(false);//只发布大流
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void handleAction(int action, Object... params) {
        //处理教师端直播间操作
        if (action == LiveRoomMsgConstants.ACTION_AGREE_MIC_APPLY) {
            //主讲人同意连麦申请
            Object param = params[0];
            if (param instanceof User) {
                User user = (User) param;
                RCUserSeatResponseMessage rcUserSeatResponseMessage = new RCUserSeatResponseMessage();
                rcUserSeatResponseMessage.setTeacherId(UserHelper.getUserId());
                rcUserSeatResponseMessage.setTeacherName(UserHelper.getUserName());
                rcUserSeatResponseMessage.setAudienceId(user.getUserId());
                rcUserSeatResponseMessage.setAudienceName(user.getUserName());
                rcUserSeatResponseMessage.setType(LiveRoomMsgConstants.MIC_RESPONSE_AGREE);
                sendMessage(rcUserSeatResponseMessage, false);
            }
            return;
        }

        if (action == LiveRoomMsgConstants.ACTION_SEND_ADD_LIKE_COUNT) {
            //发送点赞数量总数
            int count = (int) params[0];
            RCUserSyncAddLikeCountMessage syncAddLikeCountMessage = new RCUserSyncAddLikeCountMessage();
            syncAddLikeCountMessage.setCount(count);
            sendMessage(syncAddLikeCountMessage, false);
            return;
        }

        if (action == LiveRoomMsgConstants.ACTION_SEND_PAUSE_LIVE) {
            //暂停直播消息
            RCPauseLiveMessage pauseLiveMessage = new RCPauseLiveMessage();
            sendMessage(pauseLiveMessage, false);
            return;
        }
    }

    /**
     * 加入房间
     *
     * @param roomId
     * @param isCreate
     */
    private void joinRoom(String roomId, boolean isCreate) {
        //如果是观众就直接加入房间
        Log.i("pq", "joinRoom");
        LiveEventHelper.getInstance().joinRoom(roomId, RCRTCLiveRole.BROADCASTER, new ClickCallback<Boolean>() {
            @Override
            public void onResult(Boolean result, String msg) {
                Log.i("pq", "joinRoom onResult:" + result);
                if (result) {
                    setCurrentRoom(currentRoomInfo);
                } else {
                    //加入直播房间失败
                    if (getView() != null) {
                        getView().showFinishView();
                    }
                    leaveRoom();
                }
            }
        });
    }

    /**
     * 设置当前房间
     *
     * @param roomInfo
     */
    public void setCurrentRoom(LiveRoomInfoBean roomInfo) {
        initLiveRoomListener(roomInfo.roomUid);
        //设置创建者id
        LiveEventHelper.getInstance().setCreateUserId(roomInfo.speakerId);
//        getShield();获取敏感词汇
        if (getView() != null) {
            getView().setRoomData(currentRoomInfo);
        }
    }

    /**
     * 加入房间后，开始摄像头采集并发布音视频流。
     */
    public void startPublish() {
        RCRTCRoom rcrtcRoom = LiveEventHelper.getInstance().getRtcRoom();
        RCRTCEngine.getInstance().getDefaultVideoStream().startCamera(null);
        rcrtcRoom.getLocalUser().publishDefaultLiveStreams(new IRCRTCResultDataCallback<RCRTCLiveInfo>() {
            @Override
            public void onSuccess(RCRTCLiveInfo liveInfo) {
                if (getView() != null) {
                    getView().onPublishSuccess();
                }
            }

            @Override
            public void onFailed(RTCErrorCode code) {
                Log.i("pq", "publishDefaultLiveStreamError:" + code);
            }
        });
    }

    /**
     * 设置直播房的各种监听
     *
     * @param roomId
     */
    public void initLiveRoomListener(String roomId) {
        LiveEventHelper.getInstance().getRtcRoom().registerRoomListener(mRoomEventsListener);
        //添加消息监听
    }

    /**
     * 进入房间后发送默认的消息
     */
    public void sendDefaultMessage() {
        if (currentRoomInfo != null) {
            RCChatroomLocationMessage tips = new RCChatroomLocationMessage();
            tips.setContent("欢迎进入直播课堂，请遵守相关法规，禁止传播低俗、暴力等不良信息。为孩子创造健康绿色的学习环境。");
            sendMessage(tips);
        }
    }

    /**
     * 发送消息
     * 默认显示在本地
     *
     * @param messageContent
     */
    public void sendMessage(MessageContent messageContent) {
        sendMessage(messageContent, true);
    }

    /**
     * 发送消息
     *
     * @param messageContent 消息体
     * @param isShowLocation 是否显示在本地
     */
    public void sendMessage(MessageContent messageContent, boolean isShowLocation) {
        if (!isContainsShield(messageContent)) {
            LiveEventHelper.getInstance().sendMessage(messageContent, isShowLocation);
        }
    }

    /**
     * 是否包含屏蔽词
     *
     * @return
     */
    private boolean isContainsShield(MessageContent messageContent) {
        boolean isContains = false;
//        if (shields != null) {
//            for (String shield : shields) {
//                if (messageContent instanceof RCChatroomBarrage) {
//                    if (((RCChatroomBarrage) messageContent).getContent().contains(shield)) {
//                        isContains = true;
//                        break;
//                    }
//                }
//            }
//            if (isContains) {
//                //如果是包含了敏感词'
//                mView.addMessageContent(messageContent, false);
//                return true;
//            }
//        }
        return false;
    }


    /**
     * 获得当前视频流
     */
    public void getVideoStream(List<RCRTCVideoOutputStream> outputStreams, List<RCRTCVideoInputStream> inputStreams, List<RCRTCAudioInputStream> audioinputstream) {
        RCRTCRoom mRtcRoom = LiveEventHelper.getInstance().getRtcRoom();
        if (mRtcRoom == null) {
            return;
        }
        for (final RCRTCRemoteUser remoteUser : mRtcRoom.getRemoteUsers()) {
            if (remoteUser.getStreams().size() == 0) {
                continue;
            }
            List<RCRTCInputStream> userStreams = remoteUser.getStreams();
            for (RCRTCInputStream i : userStreams) {
                if (i.getMediaType() == RCRTCMediaType.VIDEO) {
                    inputStreams.add((RCRTCVideoInputStream) i);
                }
                if (i.getMediaType() == RCRTCMediaType.AUDIO) {
                    audioinputstream.add((RCRTCAudioInputStream) i);
                }
            }
        }

        for (RCRTCOutputStream o : mRtcRoom.getLocalUser().getStreams()) {
            if (o.getMediaType() == RCRTCMediaType.VIDEO) {
                outputStreams.add((RCRTCVideoOutputStream) o);
            }
        }
    }

    /**
     * 离开房间
     */
    public void leaveRoom() {
        LiveEventHelper.getInstance().leaveRoom(new IRoomCallBack() {
            @Override
            public void onSuccess() {
            }

            @Override
            public void onError(int code, String message) {
            }
        });
    }

    /**
     * 监听接收房间的所有信息
     *
     * @param roomId
     */
    public void setObMessageListener(String roomId) {
        if (disposablesManager.size() != 0) {
            return;
        }
        Disposable subscribe = RCChatRoomMessageManager.
                obMessageReceiveByRoomId(roomId)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Message>() {
                    @Override
                    public void accept(Message message) {
                        //将消息显示到列表上
                        if (LiveEventHelper.getInstance().isShowingMessage(message.getContent())) {
                            if (null != getView()) {
                                getView().addMessageContent(message, false);
                            }
                        }
                        if (TextUtils.equals(message.getObjectName(), LiveRoomMsgConstants.TAG_CHAT_ROOM_ENTER)) {
                            //加入房间消息
                            if (null != getView() && message.getContent() != null) {
                                RCChatJoinRoomMessage joinRoomMessage = (RCChatJoinRoomMessage) message.getContent();
                                getView().receiveJoinMessage(joinRoomMessage);
                            }
                            return;
                        }

                        if (TextUtils.equals(message.getObjectName(), LiveRoomMsgConstants.TAG_LIVE_ON_SNAP_UP)) {
                            //抢购消息
                            if (null != getView() && message.getContent() != null) {
                                RCOnSnappingUpMessage rcOnSnappingUpMessage = (RCOnSnappingUpMessage) message.getContent();
                                getView().receiveSnapUpMessage(rcOnSnappingUpMessage);
                            }
                            return;
                        }

                        if (TextUtils.equals(message.getObjectName(), LiveRoomMsgConstants.TAG_CHAT_ROOM_ADD_LIKE)) {
                            if (message.getContent() != null) {
                                RCUserAddLikeMessage addLikeMessage = (RCUserAddLikeMessage) message.getContent();
                                if (getView() != null) {
                                    getView().onAddLikeMessage(addLikeMessage);
                                }
                            }
                            return;
                        }
                        if (TextUtils.equals(message.getObjectName(), LiveRoomMsgConstants.TAG_CHAT_ROOM_SEAT_APPLY)) {
                            //学生连麦申请
                            if (message.getContent() != null) {
                                RCUserSeatApplyMessage rcUserSeatApplyMessage = (RCUserSeatApplyMessage) message.getContent();
                                if (getView() != null) {
                                    getView().onSeatApplyMessage(rcUserSeatApplyMessage);
                                }
                            }
                            return;
                        }

                        if (TextUtils.equals(message.getObjectName(), LiveRoomMsgConstants.TAG_LIVE_MEMBER_COUNT_SYNC)) {
                            //同步成员数量
                            if (message.getContent() != null && null != getView()) {
                                RCChatRoomMemberNumMessage rcChatRoomMemberNumMessage = (RCChatRoomMemberNumMessage) message.getContent();
                                getView().syncMemberCount(rcChatRoomMemberNumMessage.getCount());
                            }
                            return;
                        }

                        if (TextUtils.equals(message.getObjectName(), LiveRoomMsgConstants.TAG_LIVE_FINISH_MSG)) {
                            //关闭直播间
                            if (null != getView()) {
                                getView().liveRoomOffline();
                            }
                            return;
                        }
                    }
                });
        disposablesManager.add(subscribe);
    }

    public void joinChartRoom(String chatroomId, final IRongCoreCallback.OperationCallback callback) {
        LiveEventHelper.getInstance().joinChatRoom(chatroomId, callback);
    }

    /**
     * 连接融云IM
     *
     * @param
     */
    public void connectIM() {
        String imtoken = UserHelper.getUserIMToken();
        Log.i("pq", "im未连接,token:" + imtoken);
        if (TextUtils.isEmpty(imtoken)) {
            return;
        }
        if (getView() != null) {
            getView().showLoading();
        }
        IMCenter.getInstance().connect(imtoken, new RongIMClient.ConnectCallback() {
            @Override
            public void onSuccess(String t) {
                Log.i("pq", "连接成功");
            }

            @Override
            public void onError(RongIMClient.ConnectionErrorCode e) {
                Log.i("pq", "connect error" + e);
            }

            @Override
            public void onDatabaseOpened(RongIMClient.DatabaseOpenStatus code) {

            }
        });
    }

    @Override
    public void notifyJoinRoomAction(String roomId, String userId) {
        //主进程调用,因为BaseObserver onStart show了loading控件,防止在子线程创建
        addSubscribe(create(APIService.class).notifyJoinRoomAction(roomId, userId), new DisposableObserver() {
            @Override
            public void onNext(Object o) {
                //通知服务端加入成功即可,无须关注结果
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onComplete() {

            }
        });
    }

    /**
     * 通知服务端离开直播间
     */
    @Override
    public void notifyLeaveRoomAction() {
        handleAction(LiveRoomMsgConstants.ACTION_SEND_LEAVE_ROOM);
        ImUserState imUserState = new ImUserState();
        imUserState.setStatus(ImUserState.ACTION_LEAVE_LIVE_ROOM);
        imUserState.setUserid(UserHelper.getUserId());
        ArrayList<ImUserState> imUserStates = new ArrayList();
        imUserStates.add(imUserState);
        RequestBody body = RequestBodyUtil.convertToRequestBodyJson(new Gson().toJson(imUserStates));
        create(APIService.class).notifyLeaveRoomAction(body)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<BaseResponse<Object>>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(BaseResponse<Object> objectBaseResponse) {
                        //通知服务端离开即可,无须关注结果
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        //通知服务端离开即可,无须关注结果
                    }
                });
    }


    /**
     * 通知(开启/关闭)直播间录像
     *
     * @param roomId
     * @param type
     * @param userId
     */
    public void notifyOpenOpsLiveVideoAction(String roomId, String type, String userId) {
        addSubscribe(create(APIService.class).notifyOpenOpsLiveVideoAction(roomId, type, userId, "1080x1920"), new DisposableObserver() {
            @Override
            public void onNext(Object o) {
                //通知服务端加入成功即可,无须关注结果
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onComplete() {

            }
        });
    }

    /**
     * 通知关闭直播间
     *
     * @param roomId
     */
    public void notifyCloseLiveRoomAction(String roomId) {
        addSubscribe(create(APIService.class).notifyCloseLiveRoomAction(roomId), new DisposableObserver() {
            @Override
            public void onNext(Object o) {
                //通知服务端加入成功即可,无须关注结果
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onComplete() {

            }
        });
    }

    /**
     * 同步点赞数量
     *
     * @param addLikeNum
     * @param roomUid
     */
    public void syncAddLikeNum(String addLikeNum, String roomUid) {
        addSubscribe(create(APIService.class).syncAddLikeNum(addLikeNum, roomUid), new DisposableObserver() {
            @Override
            public void onNext(Object o) {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    /**
     * 取消订阅指定的流
     *
     * @param userId
     */
    public void unSubscribeStream(final String userId, final ResultCallback<String> callback) {
        RCRTCRoom rcrtcRoom = LiveEventHelper.getInstance().getRtcRoom();
        if (rcrtcRoom == null) {
            Log.i("pq", "unSubscribeStream failed , rcrtcRoom is null");
            if (callback != null) {
                callback.onFail(ErrorCode.RTC_ERROR.getCode(), null);
            }
            return;
        }
        RCRTCRemoteUser remoteUser = rcrtcRoom.getRemoteUser(userId);

        if (remoteUser == null || remoteUser.getStreams() == null) {
            Log.i("pq", "unSubscribeStream failed , remoteUser  = " + remoteUser);
            if (callback != null) {
                callback.onFail(ErrorCode.RTC_ERROR.getCode(), null);
            }
            return;
        }

        List<RCRTCInputStream> inputStreams = remoteUser.getStreams();
        Log.i("pq", "unSubscribeStream , inputStreams  = " + inputStreams);
        rcrtcRoom.getLocalUser().unsubscribeStreams(inputStreams, new IRCRTCResultCallback() {
            @Override
            public void onSuccess() {
                Log.i("pq", "unSubscribeStream success,user = " + userId);
                if (callback != null) {
                    callback.onSuccess(userId);
                }
            }

            @Override
            public void onFailed(RTCErrorCode rtcErrorCode) {
                Log.i("pq", "unSubscribeStream  error - " + rtcErrorCode.getReason());
                if (callback != null) {
                    callback.onFail(ErrorCode.RTC_ERROR.getCode(), null);
                }
            }
        });

    }
}
