广播相关代码，先上传，功能待调试

conversion_box/src/main/java/com/wdkl/app/ncs/conversion_box/activity/MainActivity.kt

@@ -50,6 +50,8 @@ import com.wdkl.ncs.android.middleware.tcp.enums.TcpType
 import com.wdkl.ncs.android.middleware.udp.ServerInfoUtil
 import com.wdkl.ncs.android.middleware.utils.AppUtil
 import com.wdkl.ncs.android.middleware.utils.CommonUtils
+import com.wdkl.ncs.janus.client.JanusClient
+import com.wdkl.ncs.janus.client.StreamingCallback
 import com.wdkl.ncs.janus.rtc.WebRTCEngine
 import com.wdkl.ncs.janus.util.JanusConstant
 import kotlinx.android.synthetic.main.conversion_box_main_lay.*
@@ -124,6 +126,11 @@ class MainActivity :BaseActivity<MainActivityPresenter, MainActivityLayoutBindin
     private var copyDone = false
     private var testing = false
 
+    private var janusClient: JanusClient? = null
+    private var broadcastOn = false
+    private var broadcastChannelId: String = ""
+    private var playing = false
+
     private var serverSuccess = false
     private val handler by lazy { Handler(Looper.getMainLooper()) }
 
@@ -638,6 +645,17 @@ class MainActivity :BaseActivity<MainActivityPresenter, MainActivityLayoutBindin
         if (!BuildConfig.DEBUG) {
             checkAppVersion()
         }
+
+
+        //test
+        /*handler.postDelayed({
+            broadcastOn = true
+            broadcastChannelId = "9"
+            //如果当前有呼叫或通话则等待
+            if (Constant.CALL_STATE == Constant.CALL_STANDBY) {
+                startBroadcast()
+            }
+        }, 5000)*/
     }
 
     private fun initCallTimer() {
@@ -1590,7 +1608,25 @@ class MainActivity :BaseActivity<MainActivityPresenter, MainActivityLayoutBindin
                                 Constant.LATER_RESTART = true
                             }
                         }
-                    } else if (tcpModel.type == TcpType.SIDE) {
+                    } /*else if (tcpModel.type == TcpType.BROADCAST) {
+                        if (tcpModel.action == TcpAction.BroadcastAction.START) {
+                            if (Constant.TCP_CONNECTED && !TextUtils.isEmpty(Constant.SIP_ID)) {
+                                if (tcpModel.data != null && !broadcastOn) {
+                                    broadcastOn = true
+                                    broadcastChannelId = tcpModel.data.toString()
+                                    //如果当前有呼叫或通话则等待
+                                    if (Constant.CALL_STATE == Constant.CALL_STANDBY) {
+                                        startBroadcast()
+                                    }
+                                }
+                            } else {
+                                showMessage("服务未连接或sipId为空")
+                            }
+                        } else if (tcpModel.action == TcpAction.BroadcastAction.STOP) {
+                            broadcastOn = false
+                            stopBroadcast(true)
+                        }
+                    }*/ else if (tcpModel.type == TcpType.SIDE) {
                         //门灯控制
                         controlDoorLight(tcpModel)
                     } else if (tcpModel.action == TcpAction.DataAction.REFRESH) {
@@ -1633,6 +1669,58 @@ class MainActivity :BaseActivity<MainActivityPresenter, MainActivityLayoutBindin
         }
     }
 
+    private fun startBroadcast() {
+        if (TextUtils.isEmpty(broadcastChannelId)) {
+            broadcastOn = false
+            return
+        }
+        playing = true
+
+        //初始化 engine
+        WebRTCEngine.getInstance().init(true, this)
+        //初始化 janusClient
+        janusClient = JanusClient(JanusConstant.JANUS_URL, Constant.SIP_ID.toBigInteger())
+        //初始化 StreamingCallback
+        val streamingCallback = StreamingCallback(janusClient, Constant.SIP_ID.toBigInteger(), broadcastChannelId.toBigInteger())
+        janusClient?.setJanusCallback(streamingCallback)
+        janusClient?.connect()
+
+        SerialPortHelper.openBroadcast()
+
+        runOnUiThread {
+            showMessage("广播开始播放")
+            //tv_broadcast_text.text = "广播正在播放......"
+            //tv_broadcast_state.setBackgroundResource(R.drawable.ic_baseline_pause)
+        }
+    }
+
+    private fun stopBroadcast(showMsg: Boolean) {
+        playing = false
+        SerialPortHelper.closeSoundChannel("FFFF")
+        if (showMsg) {
+            showMsgMain("广播停止")
+        }
+
+        if (janusClient != null) {
+            janusClient?.destroySession()
+            janusClient = null
+        }
+    }
+
+    private fun pauseBroadcast() {
+        playing = false
+        SerialPortHelper.closeSoundChannel("FFFF")
+
+        if (janusClient != null) {
+            janusClient?.destroySession()
+            janusClient = null
+        }
+
+        runOnUiThread {
+            showMessage("广播暂停播放...")
+        }
+    }
+
     fun inCalling() {
         Constant.CALL_STATE = Constant.CALL_CALLING
         updateCallText("通话中")

conversion_box/src/main/java/com/wdkl/app/ncs/conversion_box/helper/SerialPortHelper.java

@@ -24,7 +24,7 @@ public class SerialPortHelper {
         SerialPortUtil.getInstance().send(SerialPortUtil.C_HEARD + "4" + SerialPortUtil.C_SEPARATE + address + "F" + SerialPortUtil.C_END);
     }
 
-    //SIP挂断（主机挂断分机） address为FFFFF时，为关闭全部通道
+    //SIP挂断（主机挂断分机） address为FFFF时，为关闭全部通道
     public static void closeSoundChannel(String address) {
         SerialPortUtil.getInstance().send(SerialPortUtil.C_HEARD + "5" + SerialPortUtil.C_SEPARATE + address + "F" + SerialPortUtil.C_END);
     }

janus/src/main/java/com/wdkl/ncs/janus/client/JanusClient.java

@@ -5,6 +5,7 @@ import android.util.Log;
 import com.wdkl.ncs.janus.rtc.WebRTCEngine;
 import com.wdkl.ncs.janus.util.EnumType;
 
+import org.json.JSONArray;
 import org.json.JSONException;
 import org.json.JSONObject;
 import org.webrtc.IceCandidate;
@@ -631,6 +632,147 @@ public class JanusClient implements WebSocketChannel.WebSocketCallback {
     }
 
 
+    //---------------------------------------------------- 广播用 开始
+    public static final int ERROR_STREAMING_REQUEST = 0x04;
+
+    public void requestList(){
+        String tid = randomString(12);
+        transactions.put(tid, new Transaction(tid) {
+            @Override
+            public void onSuccess(JSONObject msg) throws Exception {
+                JSONObject data = msg.getJSONObject("plugindata").getJSONObject("data");
+                JSONArray array = data.getJSONArray("list");
+                /*
+                audio_age_ms: 236984209
+                description: "Opus live stream coming from external source"
+                enabled: true
+                id: 112         //来自数据库
+                metadata: "12"  //partId,
+                type: "live"
+                 */
+            }
+
+            @Override
+            public void onError() {
+                janusCallback.onError(ERROR_STREAMING_REQUEST, "获取Streaming List失败");
+            }
+        });
+
+        JSONObject message = new JSONObject();
+        JSONObject body = new JSONObject();
+        try {
+            body.putOpt("request", "list");
+            message.put("body", body);
+
+            message.putOpt("janus", "message");
+            message.putOpt("transaction", tid);
+            message.putOpt("session_id", sessionId);
+            message.putOpt("handle_id", currentHandleId);
+        } catch (JSONException e) {
+            e.printStackTrace();
+        }
+
+        sendMessage(message.toString());
+    }
+
+    public void requestInfo(BigInteger channelId){
+        String tid = randomString(12);
+        transactions.put(tid, new Transaction(tid) {
+            @Override
+            public void onSuccess(JSONObject msg) throws Exception {
+                JSONObject data = msg.getJSONObject("plugindata").getJSONObject("data");
+                JSONObject array = data.getJSONObject("info");
+                /*
+                audio: true
+                audio_age_ms: 243502007
+                audioport: 5004
+                audiopt: 111
+                audiortcpport: 0
+                audiortpmap: "opus/48000/2"
+                description: "Opus live stream coming from external source"
+                enabled: true
+                id: 112
+                metadata: "测试一下测试一下"
+                name: "rtp-audio"
+                type: "live"
+                viewers: 1
+                 */
+            }
+
+            @Override
+            public void onError() {
+                janusCallback.onError(ERROR_STREAMING_REQUEST, "获取Streaming Info失败");
+            }
+        });
+
+        JSONObject message = new JSONObject();
+        JSONObject body = new JSONObject();
+        try {
+            body.putOpt("request", "info");
+            body.putOpt("id", channelId);
+            message.put("body", body);
+
+            message.putOpt("janus", "message");
+            message.putOpt("transaction", tid);
+            message.putOpt("session_id", sessionId);
+            message.putOpt("handle_id", currentHandleId);
+        } catch (JSONException e) {
+            e.printStackTrace();
+        }
+
+        sendMessage(message.toString());
+    }
+
+    public void requestWatch(BigInteger channelId){
+        String tid = randomString(12);
+        JSONObject message = new JSONObject();
+        JSONObject body = new JSONObject();
+        try {
+            body.putOpt("request", "watch");
+            body.putOpt("id", channelId);
+            message.put("body", body);
+
+            message.putOpt("janus", "message");
+            message.putOpt("transaction", tid);
+            message.putOpt("session_id", sessionId);
+            message.putOpt("handle_id", currentHandleId);
+        } catch (JSONException e) {
+            e.printStackTrace();
+        }
+
+        sendMessage(message.toString());
+    }
+
+    public void requestStart(BigInteger channelId, SessionDescription sdp){
+        String tid = randomString(12);
+
+        JSONObject message = new JSONObject();
+        JSONObject body = new JSONObject();
+        try {
+            body.putOpt("request", "start");
+            body.putOpt("id", channelId);
+            message.put("body", body);
+
+            message.putOpt("janus", "message");
+            message.putOpt("transaction", tid);
+            message.putOpt("session_id", sessionId);
+            message.putOpt("handle_id", currentHandleId);
+
+            if (sdp != null) {
+                JSONObject jsep = new JSONObject();
+                jsep.putOpt("type", sdp.type);
+                jsep.putOpt("sdp", sdp.description);
+                message.putOpt("jsep", jsep);
+            }
+        } catch (JSONException e) {
+            e.printStackTrace();
+        }
+
+        sendMessage(message.toString());
+    }
+    //---------------------------------------------------- 广播用 结束
+
+
     private synchronized void sendMessage(String message){
         if(webSocketChannel!=null){
             webSocketChannel.sendMessage(message);

janus/src/main/java/com/wdkl/ncs/janus/client/StreamingCallback.java

@@ -0,0 +1,197 @@
+package com.wdkl.ncs.janus.client;
+
+import android.util.Log;
+
+import com.wdkl.ncs.janus.rtc.Peer;
+import com.wdkl.ncs.janus.rtc.WebRTCEngine;
+import com.wdkl.ncs.janus.rtc.observer.CreateAnswerCallback;
+import com.wdkl.ncs.janus.rtc.observer.CreatePeerConnectionCallback;
+import com.wdkl.ncs.janus.util.EnumType;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.webrtc.IceCandidate;
+import org.webrtc.MediaStream;
+import org.webrtc.SdpObserver;
+import org.webrtc.SessionDescription;
+
+import java.math.BigInteger;
+
+public class StreamingCallback implements JanusClient.JanusCallback {
+
+    private final static String TAG = StreamingCallback.class.getSimpleName();
+
+    private JanusClient janusClient;
+    private Peer peer;
+    private BigInteger userId;
+    private BigInteger currentHandleId;
+    private BigInteger channelId;
+
+    public StreamingCallback(JanusClient janusClient, BigInteger userId, BigInteger channelId){
+        this.janusClient = janusClient;
+        this.userId = userId;
+        this.channelId = channelId;
+    }
+
+    @Override
+    public void onCreateSession(BigInteger sessionId) {
+        janusClient.attachPlugin("janus.plugin.streaming");
+    }
+
+    @Override
+    public void onCreateRoom(BigInteger handleId) {
+
+    }
+
+    @Override
+    public void onAttached(BigInteger handleId) {
+        Log.d(TAG, "onAttached");
+        this.currentHandleId = handleId;
+        janusClient.requestWatch(channelId);
+        janusClient.requestInfo(channelId);
+    }
+
+    @Override
+    public void onSubscribeAttached(BigInteger subscribeHandleId, BigInteger feedId) {
+
+    }
+
+    @Override
+    public void onDetached(BigInteger handleId) {
+
+    }
+
+    @Override
+    public void onHangup(BigInteger handleId) {
+
+    }
+
+    //接收 JanusClient 810 行处理
+    @Override
+    public void onMessage(BigInteger sender, BigInteger handleId, JSONObject data, JSONObject jsep) {
+        if (!data.has("streaming")) {
+            return;
+        }
+
+        try {
+            String type = data.getString("streaming");
+            switch (type){
+                case "event":
+                    String status = data.getJSONObject("result").getString("status");
+                    //频道准备中
+                    if (status.equals("preparing")){
+                        // sdp 协商成功，收到网关转发来的sdp offer，开始接收
+                        String sdp = jsep.getString("sdp");
+                        //创建本地PeerConnection
+                        WebRTCEngine.getInstance().createLocalPeer(userId, peerConnectionCallback);
+                        this.peer = WebRTCEngine.getInstance().getPeer(userId);
+
+                        this.peer.getPeerConnection().setRemoteDescription(new SdpObserver() {
+                            @Override
+                            public void onCreateSuccess(SessionDescription sdp) {
+                                Log.d(TAG, "setRemoteDescription onCreateSuccess");
+                            }
+
+                            @Override
+                            public void onSetSuccess() {
+                                Log.d(TAG, "setRemoteDescription onSetSuccess");
+                                // 回复网关一个 start ，附带自己的 sdp answer
+                                WebRTCEngine.getInstance().getPeer(userId).createAnswer(new CreateAnswerCallback() {
+                                    @Override
+                                    public void onSetAnswerSuccess(SessionDescription sdp) {
+                                        janusClient.requestStart(channelId, sdp);
+                                    }
+
+                                    @Override
+                                    public void onSetAnswerFailed(String error) {
+
+                                    }
+                                });
+                            }
+
+                            @Override
+                            public void onCreateFailure(String error) {
+                                Log.d(TAG, "setRemoteDescription onCreateFailure " + error);
+                            }
+
+                            @Override
+                            public void onSetFailure(String error) {
+                                Log.d(TAG, "setRemoteDescription onSetFailure " + error);
+                            }
+                        }, new SessionDescription(SessionDescription.Type.OFFER, sdp));
+                    } else if (status.equals("starting")){
+
+                    } else if (status.equals("started")){
+
+                    }
+                    break;
+            }
+        } catch (JSONException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Override
+    public void onIceCandidate(BigInteger handleId, JSONObject candidate) {
+
+    }
+
+    @Override
+    public void onDestroySession(BigInteger sessionId) {
+        this.currentHandleId = null;
+        this.channelId = null;
+        janusClient.disConnect();
+    }
+
+    @Override
+    public void onError(int errorCode, String error) {
+
+    }
+
+    //收集candidate并发送
+    private CreatePeerConnectionCallback peerConnectionCallback = new CreatePeerConnectionCallback() {
+        @Override
+        public void onIceGatheringComplete() {
+            janusClient.trickleCandidateComplete(currentHandleId);
+        }
+
+        @Override
+        public void onIceCandidate(IceCandidate candidate) {
+            janusClient.trickleCandidate(currentHandleId, candidate);
+        }
+
+        @Override
+        public void onIceCandidatesRemoved(IceCandidate[] candidates) {
+            peer.getPeerConnection().removeIceCandidates(candidates);
+        }
+
+        @Override
+        public void onAddStream(MediaStream stream) {
+            if (stream.videoTracks.size() > 0) {
+                stream.audioTracks.get(0).setEnabled(true);
+                WebRTCEngine.getInstance().getPeer(userId)._remoteStream = stream;
+                janusClient.setCallState(EnumType.CallState.Connected);
+            }
+        }
+
+        @Override
+        public void onRemoveStream(MediaStream stream) {
+            WebRTCEngine.getInstance().getPeer(userId)._remoteStream = null;
+        }
+
+        @Override
+        public void onIceConnected() {
+
+        }
+
+        @Override
+        public void onIceDisconnected() {
+
+        }
+
+        @Override
+        public void onIceConnectFail() {
+
+        }
+    };
+}