流媒体增加webrtc和rtmp协议推拉流

‘liusuyi’

2023-08-26 76817b8c752b12030ab285bcb5b2effebfa9a248

流媒体增加webrtc和rtmp协议推拉流

 ard-work/src/main/java/com/ruoyi/media/controller/MediaController.java

@@ -17,6 +17,9 @@
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @Description:
@@ -38,17 +41,16 @@
    @PreAuthorize("@ss.hasPermi('media:stream:add')")
    @ApiOperationSupport(includeParameters = {"streamInfo.name", "streamInfo.rtspSource", "streamInfo.mode"})
    public AjaxResult addPath(@RequestBody StreamInfo streamInfo) {
        if(StringUtils.isEmpty(streamInfo.getName()))
        {
        if (StringUtils.isEmpty(streamInfo.getName())) {
            return AjaxResult.error("通道名称不能为空");
        }
        if(StringUtils.isEmpty(streamInfo.getRtspSource()))
        {
        if (StringUtils.isEmpty(streamInfo.getRtspSource())) {
            return AjaxResult.error("rtsp地址不能为空");
        }
        String rtsp = mediaService.addPath(streamInfo.getName(), streamInfo.getRtspSource(), streamInfo.getMode(),streamInfo.getIsCode());
        return AjaxResult.success(rtsp);
    }

    /**
     * èŽ·å–转码详细信息
     */
@@ -57,6 +59,7 @@
    public AjaxResult getInfo(@PathVariable("name") String name) {
        return success(mediaService.getPathInfo(name));
    }

    /**
     * ä¿®æ”¹è½¬ç 
     */
@@ -128,7 +131,19 @@
    @PreAuthorize("@ss.hasPermi('media:stream:remove')")
    @DeleteMapping("/{id}")
    public AjaxResult removePullStreamSession(@PathVariable String id) {
        List<StreamInfo> pullStreamList = mediaService.getPullStreamList();
        StreamInfo streamInfo = pullStreamList.stream()
                .filter(object -> object.getId().equals(id))
                .collect(Collectors.toList()).get(0);
        switch (streamInfo.getSessionType()) {
            case "rtsp":
        return AjaxResult.success(mediaService.kickRtspSession(id));
            case "webrtc":
                return AjaxResult.success(mediaService.kickWebrtcSession(id));
            case "rtmp":
                return AjaxResult.success(mediaService.kickRtmpSession(id));
        }
        return AjaxResult.error();
    }

    @PreAuthorize("@ss.hasPermi('media:stream:list')")

 ard-work/src/main/java/com/ruoyi/media/domain/RtmpSession.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,24 @@
package com.ruoyi.media.domain;

import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;

import java.util.Date;

/**
 * @Description:
 * @ClassName: WebrtcSession
 * @Author: åˆ˜è‹ä¹‰
 * @Date: 2023å¹´08月26日10:16:21
 **/
@Data
public class RtmpSession {
    private String name;
    private String id;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    private Date created;
    private String remoteAddr;
    private String state;
    private long bytesReceived;
    private long bytesSent;
}

 ard-work/src/main/java/com/ruoyi/media/domain/StreamInfo.java

@@ -33,6 +33,14 @@
     */
    String rtspUrl;
    /**
     * rtmp播放地址
     */
    String rtmpUrl;
    /**
     * webrtc播放地址
     */
    String webrtcUrl;
    /**
     * ä¼ è¾“方式
     */
    String protocol;
@@ -67,5 +75,9 @@
     * æ˜¯å¦è½¬ç 
     */
    String isCode;
    /**
     * åª’体类型（webRTCSession/rtspSession）
     */
    String sessionType;

}

 ard-work/src/main/java/com/ruoyi/media/domain/WebrtcSession.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,24 @@
package com.ruoyi.media.domain;

import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;

import java.util.Date;

/**
 * @Description:
 * @ClassName: WebrtcSession
 * @Author: åˆ˜è‹ä¹‰
 * @Date: 2023å¹´08月26日10:16:21
 **/
@Data
public class WebrtcSession {
    private String name;
    private String id;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    private Date created;
    private String remoteAddr;
    private String state;
    private long bytesReceived;
    private long bytesSent;
}

 ard-work/src/main/java/com/ruoyi/media/service/IMediaService.java

@@ -1,8 +1,6 @@
package com.ruoyi.media.service;

import com.ruoyi.media.domain.Items;
import com.ruoyi.media.domain.RtspSession;
import com.ruoyi.media.domain.StreamInfo;
import com.ruoyi.media.domain.*;

import java.util.List;

@@ -29,7 +27,8 @@
    public List<Items> rtspsessions();

    RtspSession getRtspSessionById(String sessionId);

    WebrtcSession getWebrtcSessionById(String sessionId);
    RtmpSession getRtmpSessionById(String sessionId);
    List<RtspSession> getPushStreams();

    List<RtspSession> getPullStreams();
@@ -39,5 +38,6 @@
    List<StreamInfo> getPullStreamList();

    Boolean kickRtspSession(String sessionId);

    Boolean kickRtmpSession(String sessionId);
    Boolean kickWebrtcSession(String sessionId);
}

 ard-work/src/main/java/com/ruoyi/media/service/impl/MediaService.java

@@ -130,9 +130,7 @@
        if (matcher.find()) {
            info.setRtspSource(matcher.group());
            info.setIsCode("1");
        }
        else
        {
        } else {
            info.setRtspSource(item.getConf().getSource());
            info.setIsCode("0");
        }
@@ -212,6 +210,20 @@
    }

    @Override
    public WebrtcSession getWebrtcSessionById(String sessionId) {
        String list = mediaClient.getWebrtcsessionById(sessionId);
        WebrtcSession webrtcSession = JSONObject.parseObject(list, WebrtcSession.class);
        return webrtcSession;
    }

    @Override
    public RtmpSession getRtmpSessionById(String sessionId) {
        String list = mediaClient.getRtmpsessionById(sessionId);
        RtmpSession rtmpSession = JSONObject.parseObject(list, RtmpSession.class);
        return rtmpSession;
    }

    @Override
    public List<RtspSession> getPushStreams() {
        List<RtspSession> rtspSessions = new ArrayList<>();

@@ -255,9 +267,15 @@
            //ID
            String name = item.getName();
            info.setName(name);
            //RTMP播放地址
            String rtmpUrl = "rtmp://" + mediamtxHost + ":1935/" + name;
            info.setRtmpUrl(rtmpUrl);
            //RTSP播放地址
            String rtspUrl = "rtsp://" + mediamtxHost + ":7554/" + name;
            info.setRtspUrl(rtspUrl);
            //WEBRTC播放地址
            String webrtcUrl = "http://" + mediamtxHost + ":8889/" + name;
            info.setWebrtcUrl(webrtcUrl);
            Source source = item.getSource();
            if (source==null||source.getId().equals("")) {
                //会话ID
@@ -296,18 +314,10 @@
            Matcher matcher = pattern.matcher(runoninit);
            if (matcher.find()) {
                info.setRtspSource(matcher.group());
            }
            else
            {
            } else {
                info.setRtspSource(item.getConf().getSource());
            }
            //传输协议
//            regex = "-rtsp_transport\\s+(\\w+)";
//            pattern = Pattern.compile(regex);
//            matcher = pattern.matcher(runoninit);
//            if (matcher.find()) {
//                info.setProtocol(matcher.group(1));
//            }
            info.setProtocol(item.getConf().getSourceprotocol());
            //拉流数量
            List<Readers> readers = item.getReaders();
@@ -332,6 +342,57 @@
                //ID
                String name = item.getName();
                info.setName(name);
                //传输协议
                info.setProtocol(item.getConf().getSourceprotocol());

                String type = reader.getType();
                switch (type) {
                    case "rtmpConn":
                        info.setSessionType("rtmp");
                        //webrtc播放地址
                        String url = "rtmp://" + mediamtxHost + ":1935/" + name;
                        info.setRtspUrl(url);
                        RtmpSession rtmpSession = getRtmpSessionById(reader.getId());
                        //会话ID
                        info.setId(rtmpSession.getId());
                        //开始拉流时间
                        info.setBeginTime(rtmpSession.getCreated());
                        //上行流量
                        long bytesReceived = rtmpSession.getBytesReceived();
                        String formatReceivedSize = ArdTool.formatFileSize(bytesReceived);
                        info.setUpTraffic(formatReceivedSize);
                        //下行流量
                        long bytesSent = rtmpSession.getBytesSent();
                        String formatSentSize = ArdTool.formatFileSize(bytesSent);
                        info.setDownTraffic(formatSentSize);
                        //拉流服务器
                        info.setRemoteAddr(rtmpSession.getRemoteAddr());
                        PullStreamInfoList.add(info);
                        break;
                    case "webRTCSession":
                        info.setSessionType("webrtc");
                        //webrtc播放地址
                        url = "http://" + mediamtxHost + ":8889/" + name;
                        info.setRtspUrl(url);
                        WebrtcSession webrtcSession = getWebrtcSessionById(reader.getId());
                        //会话ID
                        info.setId(webrtcSession.getId());
                        //开始拉流时间
                        info.setBeginTime(webrtcSession.getCreated());
                        //上行流量
                         bytesReceived = webrtcSession.getBytesReceived();
                         formatReceivedSize = ArdTool.formatFileSize(bytesReceived);
                        info.setUpTraffic(formatReceivedSize);
                        //下行流量
                         bytesSent = webrtcSession.getBytesSent();
                         formatSentSize = ArdTool.formatFileSize(bytesSent);
                        info.setDownTraffic(formatSentSize);
                        //拉流服务器
                        info.setRemoteAddr(webrtcSession.getRemoteAddr());
                        PullStreamInfoList.add(info);
                        break;
                    case "rtspSession":
                        info.setSessionType("rtsp");
                //RTSP播放地址
                String rtspUrl = "rtsp://" + mediamtxHost + ":8554/" + name;
                info.setRtspUrl(rtspUrl);
@@ -341,26 +402,18 @@
                //开始拉流时间
                info.setBeginTime(rtspSession.getCreated());
                //上行流量
                long bytesReceived = rtspSession.getBytesReceived();
                String formatReceivedSize = ArdTool.formatFileSize(bytesReceived);
                        bytesReceived = rtspSession.getBytesReceived();
                        formatReceivedSize = ArdTool.formatFileSize(bytesReceived);
                info.setUpTraffic(formatReceivedSize);
                //下行流量
                long bytesSent = rtspSession.getBytesSent();
                String formatSentSize = ArdTool.formatFileSize(bytesSent);
                        bytesSent = rtspSession.getBytesSent();
                        formatSentSize = ArdTool.formatFileSize(bytesSent);
                info.setDownTraffic(formatSentSize);

                //传输协议
//                String runoninit = item.getConf().getRunondemand();
//                String regex = "-rtsp_transport\\s+(\\w+)";
//                Pattern pattern = Pattern.compile(regex);
//                Matcher matcher = pattern.matcher(runoninit);
//                if (matcher.find()) {
//                    info.setProtocol(matcher.group(1));
//                }
                info.setProtocol(item.getConf().getSourceprotocol());
                //拉流服务器
                info.setRemoteAddr(rtspSession.getRemoteAddr());
                PullStreamInfoList.add(info);
                        break;
                }
            }
        }
        Comparator<StreamInfo> comparator = Comparator.comparing(streamInfo ->streamInfo.getBeginTime() ); // ä½¿ç”¨Collections.sort方法进行排序 Collections.sort(personList, comparator);
@@ -371,7 +424,27 @@
    @Override
    public Boolean kickRtspSession(String sessionId) {
        try {
            mediaClient.kick(sessionId);
            mediaClient.kickRtspSessions(sessionId);
            return true;
        } catch (Exception ex) {
            return false;
        }
    }

    @Override
    public Boolean kickRtmpSession(String sessionId) {
        try {
            mediaClient.kickRtmpSessions(sessionId);
            return true;
        } catch (Exception ex) {
            return false;
        }
    }

    @Override
    public Boolean kickWebrtcSession(String sessionId) {
        try {
            mediaClient.kickWebrtcSessions(sessionId);
            return true;
        } catch (Exception ex) {
            return false;

 ard-work/src/main/java/com/ruoyi/utils/forest/MediaClient.java

@@ -53,11 +53,29 @@
     */
    @Get("/rtspsessions/get/{sessionId}")
    String getRtspsessionById(@Var("sessionId") String sessionId);

    /**
     * æŒ‰sessionId查询webrtc会话
     */
    @Get("/webrtcsessions/get/{sessionId}")
    String getWebrtcsessionById(@Var("sessionId") String sessionId);
    /**
     * æŒ‰sessionId查询rtmp会话
     */
    @Get("/rtmpconns/get/{sessionId}")
    String getRtmpsessionById(@Var("sessionId") String sessionId);
    /**
     * æŒ‰sessionId删除rtsp会话
     */
    @Post("/rtspsessions/kick/{sessionId}")
    String kick(@Var("sessionId") String sessionId);

    String kickRtspSessions(@Var("sessionId") String sessionId);
    /**
     * æŒ‰sessionId删除rtmp连接
     */
    @Post("/rtmpconns/kick/{sessionId}")
    String kickRtmpSessions(@Var("sessionId") String sessionId);
    /**
     * æŒ‰sessionId删除webrtc会话
     */
    @Post("/webrtcsessions/kick/{sessionId}")
    String kickWebrtcSessions(@Var("sessionId") String sessionId);
}

 ard-work/src/main/resources/static/js/WHEPClient.js

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,76 @@
import negotiateConnectionWithClientOffer from "./negotiateConnectionWithClientOffer.js";
/**
 * Example implementation of a client that uses WHEP to playback video over WebRTC
 *
 * https://www.ietf.org/id/draft-murillo-whep-00.html
 */
export default class WHEPClient {
    constructor(endpoint, videoElement) {
        this.endpoint = endpoint;
        this.videoElement = videoElement;
        this.stream = new MediaStream();
        /**
         * Create a new WebRTC connection, using public STUN servers with ICE,
         * allowing the client to disover its own IP address.
         * https://developer.mozilla.org/en-US/docs/Web/API/WebRTC_API/Protocols#ice
         */
        this.peerConnection = new RTCPeerConnection({
            iceServers: [
                {
                    urls: "stun:stun.cloudflare.com:3478",
                },
            ],
            bundlePolicy: "max-bundle",
        });
        /** https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection/addTransceiver */
        this.peerConnection.addTransceiver("video", {
            direction: "recvonly",
        });
        this.peerConnection.addTransceiver("audio", {
            direction: "recvonly",
        });
        /**
         * When new tracks are received in the connection, store local references,
         * so that they can be added to a MediaStream, and to the <video> element.
         *
         * https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection/track_event
         */
        this.peerConnection.ontrack = (event) => {
            const track = event.track;
            const currentTracks = this.stream.getTracks();
            const streamAlreadyHasVideoTrack = currentTracks.some(
                (track) => track.kind === "video"
            );
            const streamAlreadyHasAudioTrack = currentTracks.some(
                (track) => track.kind === "audio"
            );
            switch (track.kind) {
                case "video":
                    if (streamAlreadyHasVideoTrack) {
                        break;
                    }
                    this.stream.addTrack(track);
                    break;
                case "audio":
                    if (streamAlreadyHasAudioTrack) {
                        break;
                    }
                    this.stream.addTrack(track);
                    break;
                default:
                    console.log("got unknown track " + track);
            }
        };
        this.peerConnection.addEventListener("connectionstatechange", (ev) => {
            if (this.peerConnection.connectionState !== "connected") {
                return;
            }
            if (!this.videoElement.srcObject) {
                this.videoElement.srcObject = this.stream;
            }
        });
        this.peerConnection.addEventListener("negotiationneeded", (ev) => {
            negotiateConnectionWithClientOffer(this.peerConnection, this.endpoint);
        });
    }
}

 ard-work/src/main/resources/static/js/negotiateConnectionWithClientOffer.js

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,80 @@
/**
 * Performs the actual SDP exchange.
 *
 * 1. Constructs the client's SDP offer
 * 2. Sends the SDP offer to the server,
 * 3. Awaits the server's offer.
 *
 * SDP describes what kind of media we can send and how the server and client communicate.
 *
 * https://developer.mozilla.org/en-US/docs/Glossary/SDP
 * https://www.ietf.org/archive/id/draft-ietf-wish-whip-01.html#name-protocol-operation
 */
export default async function negotiateConnectionWithClientOffer(
    peerConnection,
    endpoint
) {
    /** https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection/createOffer */
    const offer = await peerConnection.createOffer();
    /** https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection/setLocalDescription */
    await peerConnection.setLocalDescription(offer);
    /** Wait for ICE gathering to complete */
    let ofr = await waitToCompleteICEGathering(peerConnection);
    if (!ofr) {
        throw Error("failed to gather ICE candidates for offer");
    }
    /**
     * As long as the connection is open, attempt to...
     */
    while (peerConnection.connectionState !== "closed") {
        /**
         * This response contains the server's SDP offer.
         * This specifies how the client should communicate,
         * and what kind of media client and server have negotiated to exchange.
         */
        let response = await postSDPOffer(endpoint, ofr.sdp);
        if (response.status === 201) {
            let answerSDP = await response.text();
            await peerConnection.setRemoteDescription(
                new RTCSessionDescription({ type: "answer", sdp: answerSDP })
            );
            return response.headers.get("Location");
        } else if (response.status === 405) {
            console.error("Update the URL passed into the WHIP or WHEP client");
        } else {
            const errorMessage = await response.text();
            console.error(errorMessage);
        }
        /** Limit reconnection attempts to at-most once every 5 seconds */
        await new Promise((r) => setTimeout(r, 5000));
    }
}
async function postSDPOffer(endpoint, data) {
    return await fetch(endpoint, {
        method: "POST",
        mode: "cors",
        headers: {
            "content-type": "application/sdp",
        },
        body: data,
    });
}
/**
 * Receives an RTCPeerConnection and waits until
 * the connection is initialized or a timeout passes.
 *
 * https://www.ietf.org/archive/id/draft-ietf-wish-whip-01.html#section-4.1
 * https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection/iceGatheringState
 * https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection/icegatheringstatechange_event
 */
async function waitToCompleteICEGathering(peerConnection) {
    return new Promise((resolve) => {
        /** Wait at most 1 second for ICE gathering. */
        setTimeout(function () {
            resolve(peerConnection.localDescription);
        }, 1000);
        peerConnection.onicegatheringstatechange = (ev) =>
            peerConnection.iceGatheringState === "complete" &&
            resolve(peerConnection.localDescription);
    });
}

 ard-work/src/main/resources/templates/test.html

@@ -13,6 +13,20 @@
            margin-top: 10px;
        }
    </style>
    <!-- å¯¼å…¥ ECMAScript æ¨¡å— -->
    <script th:type="module" th:src="@{/js/WHEPClient.js}"></script>
    <!-- å†…联脚本块，可以使用模块中的内容 -->
    <script th:inline="javascript">
        /*<![CDATA[*/
        // åœ¨è¿™é‡Œä½¿ç”¨ WHEPClient æ¨¡å—中的内容
        $("#play").click(function webrtcClient(){
            const  url = "http://127.0.0.1:8889/165"; // add the webRTC URL from your live input here
            console.log(url)
            const  videoElement = document.getElementById("remote-video");
            const  client = new WHEPClient(url, videoElement);
        })
        /*]]>*/
    </script>
<body>
<div class="container">
    <div class="row ">
@@ -127,11 +141,15 @@
            <div class="row top-buffer">
                <video id="video" muted autoplay loop controls style="width: 800px; height: 100%; object-fit: fill;"/>
            </div>
            <div class="row top-buffer">
                <button id="play" type="button" class="btn btn-default" >播放</button>
            <video id="remote-video" controls autoplay muted></video>
            </div>
        </div>
    </div>
</div>
<script th:inline="javascript">
    var cameraId, opt, token;
    var cameraId, opt,optOpen,optClose, token;
    window.onload = function () {
        $.ajax({
            url: "../hik/list",
@@ -828,5 +846,6 @@
        webRtcServer.disconnect();
    }
</script>

</body>
</html>

 lib/mediamtx/mediamtx.yml

@@ -105,7 +105,7 @@
# RTMP parameters

# Disable support for the RTMP protocol.
rtmpDisable: yes
rtmpDisable: no
# Address of the RTMP listener. This is needed only when encryption is "no" or "optional".
rtmpAddress: :1935
# Encrypt connections with TLS (RTMPS).
@@ -181,7 +181,7 @@
# WebRTC parameters

# Disable support for the WebRTC protocol.
webrtcDisable: yes
webrtcDisable: no
# Address of the WebRTC listener.
webrtcAddress: :8889
# Enable TLS/HTTPS on the WebRTC server.