kurento-one2many-broadcast V6.0源码分析

kurento-one2many-call V6.0 源码分析

 

一、Web页面

后台服务用命令行启动:

   $ mvn clean compile exec:java

启动成功后，在chorme浏览器的地址栏输入:

  

即可看到如下页面

 

二、系统分析

在这个应用程序员有两类用户:

一个发送媒体的(我们称之为表演者)，

N个从表演者处接收媒体流(我们称之为观看者)。

因此，这个媒体管道是由1+N个互联的WebRtcEndpoint组成。上面的图片显示了表演者的页面。

 

为了实现这个应用程序功能，我们需要创建一个由1+N个WebRtcEndpoint组成的媒体管道。

表演者端发送它的流到其它观看者。观看者都配置成只接收模式。这个媒体管道的示例图如下：

2.1 示例程序的框架

这是一个Web应用程序，遵从客户-服务端的架构，主要分成三个部分:

?  页面客户端，包括HTML页面，页面按键调用的JavaScript函数，以及支持前两者的Web服务器----Tomcat，

?  应用程序服务端，使用Java EE应用程序服务框架，调用Kurento Java Client API。

?  Kurento Media Server， 是真正进行媒体处理的多媒体服务器。

 

这三者之间的通信都是使用WebSocket+JSON实现。

 

2.2. 系统通信的时序图

系统通信的时序图如下:

三、 页面客户端

3.1. Web服务器的实现

这个页面应用程序的Web服务器使用了基于Spring Boot的Tomcat.

Web服务的pom.xml中相关的设置如下：

. . .

    8083

 

   

    org.kurento.tutorial.one2manycall.One2ManyCallApp

   

       

            org.springframework.boot

            spring-boot-starter-web

       

        . . .

   

. . .

 

在上面的配置中，

spring-boot-starter-web 告诉spring boot，我们要启动web应用；

org.kurento.tutorial.helloworld.HelloWorldApp 告诉 Spring Boot Maven在启动Tomcat后要执行的主类所在java文件位置；

 

3.2 处理页面请求的主类

接着看主类所在的Java文件

  “src/main/java/org/kurento/tutorial/helloworld/HelloWorldApp.java”

代码如下:

package org.kurento.tutorial.helloworld;

. . .

@Configuration

@EnableWebSocket

@EnableAutoConfiguration

public class One2ManyCallApp implements WebSocketConfigurer {

 

    final static String DEFAULT_KMS_WS_URI = "ws://localhost:8888/kurento";

    @Bean

    public CallHandler callHandler() {

        return new CallHandler();

    }

    @Bean

    public KurentoClient kurentoClient() {

        return KurentoClient.create(System.getProperty("kms.ws.uri",DEFAULT_KMS_WS_URI));

    }

    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {

        registry.addHandler(callHandler(), "/call");

    }

    public static void main(String[] args) throws Exception {

        new SpringApplication(One2ManyCallApp.class).run(args);

    }

 

}其中，

@EnableAutoConfiguration 注解告诉Spring Boot根据添加的jar依赖猜测你想如何配置Spring。

由于 spring-boot-starter-web 添加了Tomcat和Spring MVC，所以auto-configuration将假定你正在开发一个web应用并相应地对Spring进行设置。

 

这个Java程序最后部分是main方法”public static void main(String[] args) throws Exception {”。

这只是一个标准的方法，它遵循Java对于一个应用程序入口点的约定。

我们的main方法通过调用run，将业务委托给了Spring Boot的SpringApplication类。SpringApplication将引导我们的应用，启动Spring，相应地启动被自动配置的Tomcat web服务器。

我们需要将 One2ManyCallApp.class作为参数传递给run方法来告诉SpringApplication谁是主要的Spring组件。为了暴露任何的命令行参数，args数组也会被传递过去。

主类One2ManyCallApp 实现了接口WebSocketConfigurer，

并通过注册函数

    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {

        registry.addHandler(callHandler(), "/call");

    }

注册了一个 WebSocketHandler, 即:

    @Bean

    public CallHandler callHandler() {

        return new CallHandler();

}

CallHandler类实现了TextWebSocketHandler来处理文本WebSocket的请求，它就是WebSocket的服务端。

通过这个类来处理对路径 “/call” 下的WebSocket请求。

我们可以看到，在对应页面客户端 src/main/resources/static/js/index.js中，

WebSocket客户端的路径设置如下:

    var  ws = new WebSocket('ws://' + location.host + '/call');

 

3.3 页面客户端代码分析

为了和服务端的WebSocket服务通信，

我们需要在客户端页面中使用 JavaScript类 WebSocket。

我们还需要使用 Kurento JavaScript 库，叫做 kurento-util.js,

来简化和服务端的WebRTC交互。

这个库依赖于 adapter.js, 它是谷歌开发JavaScript WebRTC库，它抽象并封装了各个浏览器间的差异，能方便页面程序开发人员的开发。

页面客户端还需要 jquery.js；

 

这些库的都在 src/main/resources/static/index.html 而在中被链接:

. . .

 

 

. . .

在 src/main/resources/static/js/index.js中被使用。

 

【NOTE】换句话说，我们在开发自己的应用时，可以不需要使用基于spring boot + tomcat的框架，可以直接使用自己的Web服务框架，如Nginx,  python+tornado等，链接上这些JavaScript库也能正常工作。

 

页面客户端工作逻辑

3.3.1 >>> 首先，

需要创建一个对路径”/call”通信的WebSocket:

 

var ws = new WebSocket('ws://' + location.host + '/call');

 

当点击页面的presenter按键后，将调用src/main/resources/static/js/index.js的presenter()函数，开始直播发布者的WebRTC通信:

var ws = new WebSocket('ws://' + location.host + '/call');

var video;

var webRtcPeer;

 

function presenter() {

    if (!webRtcPeer) {

        showSpinner(video);

 

        var options = {

            localVideo : video,

            onicecandidate : onIceCandidate

        }

        webRtcPeer = new kurentoUtils.WebRtcPeer.WebRtcPeerSendonly(options,

                function(error) {

                    if (error) {

                        return console.error(error);

                    }

                    webRtcPeer.generateOffer(onOfferPresenter);

                });

    }

}

function onOfferPresenter(error, offerSdp) {

    if (error)

        return console.error('Error generating the offer');

    console.info('Invoking SDP offer callback function ' + location.host);

    var message = {

        id : 'presenter',

        sdpOffer : offerSdp

    }

    sendMessage(message);

}

 

其中，kurentoUtils.WebRtcPeer.WebRtcPeerSendonly()是js/kurento-utils.js定义的函数，它抽象了WebRTC的内部细节(如PeerConnection和getUserStream)，

并且，它使用HTML视频标签 video ---- 显示视频摄像头(本地流)来启动一个单工的WebRTC通信。

在这个函数中，将会调用generateOffer()函数。

这个函数(指generateOffer())会根据参数offeSDP生成客户端SDP请求，然后调用函数onOfferPresenter()。

在onOfferPresenter()函数中，会通过WebSocket把SDP和id信息发送到WebSocket信令服务端；

 

3.3.2 >>> 然后，

WebSocket的监听函数onmessage() 用来处理从WebSocket服务端发送到客户端的JSON信令消息。

ws.onmessage = function(message) {

    var parsedMessage = JSON.parse(message.data);

    console.info('Received message: ' + message.data);

 

    switch (parsedMessage.id) {

    case 'presenterResponse':

        presenterResponse(parsedMessage);

        break;

    case 'viewerResponse':

        viewerResponse(parsedMessage);

        break;

    case 'iceCandidate':

        webRtcPeer.addIceCandidate(parsedMessage.candidate, function(error) {

            if (error)

                return console.error('Error adding candidate: ' + error);

        });

        break;

    case 'stopCommunication':

        dispose();

        break;

    default:

        console.error('Unrecognized message', parsedMessage);

    }

}

从服务端发送来的消息有四种，presenterResponse, viewerResponse, stopCommunicate和iceCandidate，每种消息都有对应的处理函数。

对应的presenter成功返回的消息处理函数如下:

function presenterResponse(message) {

    if (message.response != 'accepted') {

        var errorMsg = message.message ? message.message : 'Unknow error';

        console.info('Call not accepted for the following reason: ' + errorMsg);

        dispose();

    } else {

        webRtcPeer.processAnswer(message.sdpAnswer, function(error) {

            if (error)

                return console.error(error);

        });

    }

}

它调用webRtcPeer.processAnswer()函数来处理返回的消息。

 

3.3.3 >>> 接着，

打一个新的页面，点击页面的viewer按键后，将调用src/main/resources/static/js/index.js的viewer()函数，开始直播观看者的WebRTC通信:

function viewer() {

    if (!webRtcPeer) {

        showSpinner(video);

 

        var options = {

            remoteVideo : video,

            onicecandidate : onIceCandidate

        }

        webRtcPeer = new kurentoUtils.WebRtcPeer.WebRtcPeerRecvonly(options,

                function(error) {

                    if (error) {

                        return console.error(error);

                    }

                    this.generateOffer(onOfferViewer);

                });

    }

}

function onOfferViewer(error, offerSdp) {

    if (error)

        return console.error('Error generating the offer');

    console.info('Invoking SDP offer callback function ' + location.host);

    var message = {

        id : 'viewer',

        sdpOffer : offerSdp

    }

    sendMessage(message);

}

其中，kurentoUtils.WebRtcPeer.WebRtcPeerRecvonly()是js/kurento-utils.js定义的函数，它抽象了WebRTC的内部细节(如PeerConnection和getUserStream)，

并且，它使用HTML视频标签 video ---- 显示远端媒体流(本地流)来启动一个单工的WebRTC通信。

在这个函数中，将会调用generateOffer()函数。

这个函数(指generateOffer())会根据参数offeSDP生成客户端SDP请求，然后调用函数onOfferViewerr()。

在onOfferViewer()函数中，会通过WebSocket把SDP和id信息发送到WebSocket信令服务端；

 

3.3.4 >>> 再然后，

WebSocket的监听函数onmessage() 用来处理从WebSocket服务端发送到客户端的JSON信令消息。

ws.onmessage = function(message) {

    var parsedMessage = JSON.parse(message.data);

    console.info('Received message: ' + message.data);

 

    switch (parsedMessage.id) {

    case 'presenterResponse':

        presenterResponse(parsedMessage);

        break;

    case 'viewerResponse':

        viewerResponse(parsedMessage);

        break;

    case 'iceCandidate':

        webRtcPeer.addIceCandidate(parsedMessage.candidate, function(error) {

            if (error)

                return console.error('Error adding candidate: ' + error);

        });

        break;

    case 'stopCommunication':

        dispose();

        break;

    default:

        console.error('Unrecognized message', parsedMessage);

    }

}

从服务端发送来的消息有四种，presenterResponse, viewerResponse, stopCommunicate和iceCandidate，每种消息都有对应的处理函数。

对应的viewer成功返回的消息处理函数如下:

function viewerResponse(message) {

    if (message.response != 'accepted') {

        var errorMsg = message.message ? message.message : 'Unknow error';

        console.info('Call not accepted for the following reason: ' + errorMsg);

        dispose();

    } else {

        webRtcPeer.processAnswer(message.sdpAnswer, function(error) {

            if (error)

                return console.error(error);

        });

    }

}

它调用webRtcPeer.processAnswer()函数来处理返回的消息。

 

3.3.5 >>> 最后，

用户点击页面stop按键后，将会调用src/main/resources/static/js/index.js中的stop()函数，

function stop() {

    var message = {

        id : 'stop'

    }

    sendMessage(message);

    dispose();

}

它会通过WebSocket客户端，发送停止消息到WebSocket服务端，结束服务。

 

四、应用程序服务端

4.1 应用程序服务端主类

如前所述，由页面客户端通过WebSocket发送的信令消息会被应用程序服务端的

 src/main/java/org/kurento/tutorial/helloworld/HelloWorldHandler.java的HelloWorldHandler类处理。

 

同时，他还需要创建一个和Kurento Media Server进行WebSocket通信的客户端:

src/main/java/org/kurento/tutorial/one2manycall/One2ManyCallApp.java 下

public class One2ManyCallApp implements WebSocketConfigurer {

 

    final static String DEFAULT_KMS_WS_URI = "ws://localhost:8888/kurento";

    @Bean

    public CallHandler callHandler() {

        return new CallHandler();

    }

 

    @Bean

    public KurentoClient kurentoClient() {

        return KurentoClient.create(System.getProperty("kms.ws.uri", DEFAULT_KMS_WS_URI));

    }

 

    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {

        registry.addHandler(callHandler(), "/call");

    }

 

    public static void main(String[] args) throws Exception {

        new SpringApplication(One2ManyCallApp.class).run(args);

    }

}

 

4.2 应用程序服务端处理逻辑

当页面客户端通过websocket发送过来信令后，会在CallHandler类的 handlerTextMessage方法中进行分别处理:

public  void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {

JsonObject jsonMessage = gson.fromJson(message.getPayload(), JsonObject.class);

log.debug("Incoming message from session '{}': {}", session.getId(), jsonMessage);

 

switch (jsonMessage.get("id").getAsString()) {

case "presenter":

    try {

       presenter(session, jsonMessage);

    } catch (Throwable t) {

       stop(session);

       log.error(t.getMessage(), t);

       JsonObject response = new JsonObject();

       response.addProperty("id", "presenterResponse");

       response.addProperty("response", "rejected");

       response.addProperty("message", t.getMessage());

       session.sendMessage(new TextMessage(response.toString()));

    }

    break;

    case "viewer":

        try {

           viewer(session, jsonMessage);

        } catch (Throwable t) {

           stop(session);

           log.error(t.getMessage(), t);

           JsonObject response = new JsonObject();

           response.addProperty("id", "viewerResponse");

           response.addProperty("response", "rejected");

           response.addProperty("message", t.getMessage());

           session.sendMessage(new TextMessage(response.toString()));

        }

        break;

    case "onIceCandidate": {

        JsonObject candidate = jsonMessage.get("candidate").getAsJsonObject();

 

        UserSession user = null;

        if (presenterUserSession.getSession() == session) {

           user = presenterUserSession;

        } else {

            user = viewers.get(session.getId());

        }

        if (user != null) {

           IceCandidate cand = new IceCandidate(candidate.get("candidate").getAsString(),

                   candidate.get("sdpMid").getAsString(), candidate.get("sdpMLineIndex").getAsInt());

           user.addCandidate(cand);

       }

       break;

   }

   case "stop":

       stop(session);

       break;

   default:

       break;

   }

}

 

4.2.1 >>>首先,

当页面客户端发送了”presenter”信令后，将调用presenter方法进行处理:

package org.kurento.tutorial.one2manycall;

           

import java.io.IOException;

import java.util.concurrent.ConcurrentHashMap;

 

import org.kurento.client.EventListener;

import org.kurento.client.IceCandidate;

import org.kurento.client.KurentoClient;

import org.kurento.client.MediaPipeline;

import org.kurento.client.OnIceCandidateEvent;

import org.kurento.client.WebRtcEndpoint;

import org.kurento.jsonrpc.JsonUtils;

. . .

public class CallHandler extends TextWebSocketHandler {

   . . .

private synchronized void presenter(final WebSocketSession session, JsonObject jsonMessage)

throws IOException {

        if (presenterUserSession == null) {

            //1.  Create user session

presenterUserSession = new UserSession(session);

           

            // 2. Media logic

            pipeline = kurento.createMediaPipeline();

            presenterUserSession.setWebRtcEndpoint(new WebRtcEndpoint.Builder(pipeline).build());

            WebRtcEndpoint presenterWebRtc = presenterUserSession.getWebRtcEndpoint();

           

            // 3. Gather ICE candidates

            presenterWebRtc.addOnIceCandidateListener(new EventListener() {

               

                @Override

                public void onEvent(OnIceCandidateEvent event) {

                    JsonObject response = new JsonObject();

                    response.addProperty("id", "iceCandidate");

                    response.add("candidate", JsonUtils.toJsonObject(event.getCandidate()));

                    try {

                        synchronized (session) {

                            session.sendMessage(new TextMessage(response.toString()));

                        }

                    } catch (IOException e) {

                        log.debug(e.getMessage());

                    }

                }

            });

 

            // 4. SDP negotiation

            String sdpOffer = jsonMessage.getAsJsonPrimitive("sdpOffer").getAsString();

            String sdpAnswer = presenterWebRtc.processOffer(sdpOffer);

 

            JsonObject response = new JsonObject();

            response.addProperty("id", "presenterResponse");

            response.addProperty("response", "accepted");

            response.addProperty("sdpAnswer", sdpAnswer);

 

            synchronized (session) {

                presenterUserSession.sendMessage(response);

            }

            presenterWebRtc.gatherCandidates();

 

        } else {

            JsonObject response = new JsonObject();

            response.addProperty("id", "presenterResponse");

            response.addProperty("response", "rejected");

            response.addProperty("message", "Another user is currently acting as sender. Try again later ...");

            session.sendMessage(new TextMessage(response.toString()));

        }

}

presenter()方法执行了下列动作:

如果当前还没有表演者，

1. 创建用户session

用户session的定义 src/main/java/org/kurento/tutorial/one2manycall/UserSession.java如下:

/**

 * User session.

 *

 * @author Boni Garcia ([e-mail])

 * @since 5.0.0

 */

public class UserSession {

   

    private static final Logger log = LoggerFactory.getLogger(UserSession.class);                                   

   

    private final WebSocketSession session;

    private WebRtcEndpoint webRtcEndpoint;

   

    public UserSession(WebSocketSession session) {

        this.session = session;

    }  

   

    public WebSocketSession getSession() {

        return session;

    }  

   

    public void sendMessage(JsonObject message) throws IOException {

        log.debug("Sending message from user with session Id '{}': {}",session.getId(), message);

        session.sendMessage(new TextMessage(message.toString()));

    }

 

    public WebRtcEndpoint getWebRtcEndpoint() {

        return webRtcEndpoint;

    }

 

    public void setWebRtcEndpoint(WebRtcEndpoint webRtcEndpoint) {

        this.webRtcEndpoint = webRtcEndpoint;

    }

 

    public void addCandidate(IceCandidate i) {

        webRtcEndpoint.addIceCandidate(i);

    }

}

 

为了释放向kurento Media Server请求的资源，我们需要把用户session(例如MediaPipeline和WebRtcEndpoint)存储起来，以在用户调用stop时释放掉。

 

2. 配置媒体处理逻辑

在这部分中，应用程序服务端配置了Kurento 如何处理媒体。

首先，使用KurentoClient对象kurento创建一个MediaPipeline对象，通过它，我们可以再创建我们想要的媒体元件并连接。

在这里，我们创建了一个WebRtcEndpoint媒体元件实例来接收WebRTC流。

 

3. 收集ICE候选者

在Version 6中，Kurento完全支持 Trickle ICE协议。

基于这个原因，WebRtcEndpoint可以接收异步的ICE候选者。为了处理它，每个WebRtcEndpoint提供了一个监听器(addOnIceGatheringDoneListener)来监听当ICE收集处理已完成的事件。

 

4. WebRTC SDP协商

在WebRTC中，SDP用来在端间实现媒体数据交换的协商。它是基于SDP提交和回答的数据交换机制。

这个协商是在方法processRequest的第三部分完成的，它使用了浏览器客户端的SDP提交，然后返回由WebRtcEndpoint生成的SDP回答。

 

4.2.2 >>> 然后

当页面客户端通过WebSocket发送了”viewer”信令后，将调用viewer方法进行处理:

private synchronized void viewer(final WebSocketSession session, JsonObject jsonMessage) throws IOException {

    // Check “presenter”

    if (presenterUserSession == null || presenterUserSession.getWebRtcEndpoint() == null) {

        JsonObject response = new JsonObject();

        response.addProperty("id", "viewerResponse");

        response.addProperty("response", "rejected");

        response.addProperty("message", "No active sender now. Become sender or . Try again later ...");

        session.sendMessage(new TextMessage(response.toString()))

     } else {

        // Check “viewer”

        if (viewers.containsKey(session.getId())) {

            JsonObject response = new JsonObject();

            response.addProperty("id", "viewerResponse");

            response.addProperty("response", "rejected");

            response.addProperty("message",

               "You are already viewing in this session. Use a different browser to add additional viewers.");

            session.sendMessage(new TextMessage(response.toString()));

            return;

        }

 

        // Create and strore user session

        UserSession viewer = new UserSession(session);

        viewers.put(session.getId(), viewer);

 

        // Media logic

        WebRtcEndpoint nextWebRtc = new WebRtcEndpoint.Builder(pipeline).build();

 

        // Gather ICE candidates

        nextWebRtc.addOnIceCandidateListener(new EventListener() {

            @Override

            public void onEvent(OnIceCandidateEvent event) {

                JsonObject response = new JsonObject();

                response.addProperty("id", "iceCandidate");

                response.add("candidate", JsonUtils.toJsonObject(event.getCandidate()));

                try {

                    synchronized (session) {

                        session.sendMessage(new TextMessage(response.toString()));

                    }

                } catch (IOException e) {

                    log.debug(e.getMessage());

                }

            }

        });

        viewer.setWebRtcEndpoint(nextWebRtc);

        presenterUserSession.getWebRtcEndpoint().connect(nextWebRtc);

 

// SDP negotiation

        String sdpOffer = jsonMessage.getAsJsonPrimitive("sdpOffer").getAsString();

 String sdpAnswer = nextWebRtc.processOffer(sdpOffer);

 

        JsonObject response = new JsonObject();

        response.addProperty("id", "viewerResponse");

        response.addProperty("response", "accepted");

        response.addProperty("sdpAnswer", sdpAnswer);

        synchronized (session) {

            viewer.sendMessage(response);

        }

        nextWebRtc.gatherCandidates();

    }

}

 

viewer()方法执行了如下动作:

首先，进行判断动作

1. 判断是否有presenter

   如果没有presenter,则通过websocket回复拒绝的消息；

2. 判断是否用同一个浏览器打开了多个viewer

   如果在同一个浏览器中有已打开的viewer,则回复拒绝的消息；

 

然后，如下上述两个判断都不成立，则进行入正式的处理阶段；

1. 创建并存储用户session

创建viewer的用户session,并把它存储到viewer的HashMap中，以用于后续的资源释放；

 

2. 配置媒体处理逻辑

新建一个WebRtcEndpoint并插入到已连接上了presenter媒体管道，

然后连接当前的viewer和presenter

 

3. SDP协商

在WebRTC中，SDP用来在端间实现媒体数据交换的协商。

它是基于SDP提交和回答的数据交换机制。

这个协商是在方法processRequest的第三部分完成的，它使用了浏览器客户端的SDP提交，然后返回由WebRtcEndpoint生成的SDP回答。

 

4.2.3 >>> 最后

当页面客户端通过websocket发送过来 “stop”信令后，将调用stop()方法处理:

 

    private synchronized void stop(WebSocketSession session) throws IOException {

        String sessionId = session.getId();

       

        if (presenterUserSession != null && presenterUserSession.getSession().getId().equals(sessionId)) {

    // Stop by “presenter”

            for (UserSession viewer : viewers.values()) {

                JsonObject response = new JsonObject();

                response.addProperty("id", "stopCommunication");

                viewer.sendMessage(response);

            }

 

            log.info("Releasing media pipeline");

            if (pipeline != null) {

                pipeline.release();

            }

            pipeline = null;

            presenterUserSession = null;

        } else if (viewers.containsKey(sessionId)) {

    // Stop by “viewer”

            if (viewers.get(sessionId).getWebRtcEndpoint() != null) {

                viewers.get(sessionId).getWebRtcEndpoint().release();

            }

            viewers.remove(sessionId);

        }

    }

 

这个方法实现的动作如下:

依据session ID来判断是presenter还是viewer发送来的”stop”信令。

1. 如果是 presenter发送的，

则对所有的viewer发送结束通信的信令，

然后通知kurenot media server释放媒体管道，

并释放presenter的用户session.

 

2. 如果是 viewer发送的，

则释放viewer申请的WebRtcEndpoint资源，

再释放viewer的用户session.