Kurento应用开发指南(以Kurento 5.0为模板) 之四：示例教程 一对一视频呼叫

5.3 教程四-一对一的视频呼叫

这个页面应用程序使用WebRTC技术实现了一个一对一的呼叫，换言话说，这个应用提供了一个简单的视频电话

5.3.1 运行示例程序

运行这个DEMO之前，你需要先安装Kurento Media Server.可以看前面的介绍。

另外，你还需要先安装好 JDK (at least version 7), Maven, Git, 和 Bower。

在Ubuntu上安装这些的命令如下:

sudo apt-get install curl

curl -sL https://deb.nodesource.com/setup | sudo bash -

sudo apt-get install -y nodejs

sudo npm install -g bower

启动应用程序之前，需要先下载源，并编译运行，命令如下:

git clone https://github.com/Kurento/kurento-tutorial-java.git
cd kurento-tutorial-java/kurento-one2one-call

mvn clean compile exec:java

默认地，这个应用程序部署在8080端口上，可以使用兼容WebRTC的浏览器打开URL http://localhost:8080

5.3.2 Understanding this example

下面的图片显示了在浏览上运行这个DEMO时截图。

这个应用程序(一个HTML页面)的接口是由两个HTML5视频标签组成的:

一个用来显示本地流；

另一个用来显示远端的流；

如果有两用户，A和B都使用这个应用程序，则媒体流的工作方式如下:

A的摄像头的流发送到Kurento Media Server，Kurento Media Server会将这个流发送给B；

同样地，B也会将流发送到Kurento Media Server，它再发给A。

这意味着，KMS提供了一个B2B (back-to-back) 的呼叫服务。



Figure 9.1: One to one video call screenshot

为了实现上述的工作方式，需要创建一个由两个WebRtc端点以B2B方式连接的媒体管道，媒体管道的示例图如下:



Figure 9.2: One to one video call Media Pipeline

客户端和服务端的通信是通过基于WebSocket上的JSON消息的信令协议实现的，客户端和服务端的工作时序如下:

1. 用户A在服务器上注册他的名字

2. 用户B在服务器注册他的名字

3. 用户A呼叫用户B

4. 用户B接受呼叫

5. 通信已建立，媒体在用户A与用户B之间流动

6. 其中一个用户结束这次通信

时序流程的细节如下图所示:



Figure 9.3: One to many one call signaling protocol

如图中所示，为了在浏览器和Kurento之间建立WebRTC连接，需要在客户端和服务端之间进行SDP交互。

特别是，SDP协商连接了浏览器的WebRtcPeer和服务端的WebRtcEndpoint。

下面的章节描述了服务端和客户端的细节，以及DEMO是如何运行的。源码可以从GitHub上下载；

5.3.3 应用程序服务端逻辑

这个DEMO的服务端是使用Java的Spring Boot框架开发的。这个技术可以嵌入到Tomcat页面服务器中，从而简化开发流程。

Note: You can use whatever Java server side technology you prefer to build

web applications with Kurento. For example, a pure Java EE application, SIP Servlets,

Play, Vertex, etc. We have choose Spring Boot for convenience.

下面的图显示了服务端的类图。

这个DEMO的主类为One2OneCallApp, 如代码中所见，KurentoClient作为Spring Bean在类中进行了实例化。



Figure 9.4: Server-side class diagram of the one to one video call app

@Configuration

@EnableWebSocket

@EnableAutoConfiguration

public class One2OneCallApp implements WebSocketConfigurer {

@Bean

public CallHandler callHandler() {

return
new CallHandler();

}

@Bean

public UserRegistry registry() {

return
new UserRegistry();

}

@Bean

public KurentoClient kurentoClient() {

return
KurentoClient.create("ws://localhost:8888/kurento");

}

public void registerWebSocketHandlers(WebSocketHandlerRegistry
registry) {

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

}

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

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

}

}

这个页面应用程序使用了单页面应用程序架构(SPA：Single Page Application architecture )，

并使用了WebSocket来作为客户端与服务端通信的请求与响应。

特别地，主app类实现了WebSocketConfigurer接口来注册一个WebSocketHandler来处理WebSocket请求。

CallHandler类实现了TextWebSocketHandler，用来处理文本WebSocket的请求。

这个类的主要实现的方法就是handleTextMessage, 这个方法实现了对请求的动作:

通过WebSocket返回对请求的响应。换句话说，它实现前面的时序图中的信令协议的服务端部分。

在设计的协议中，有三种类型的输入消息: 注册，呼叫， incomingCallResponse和stop。

这些消息对应的处理都在switch中。

public class CallHandler extends TextWebSocketHandler {

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

private static final Gson gson = new GsonBuilder().create();

private ConcurrentHashMap<String, CallMediaPipeline>
pipelines =

new ConcurrentHashMap<String,
CallMediaPipeline>();

@Autowired

private KurentoClient kurento;

@Autowired

private UserRegistry registry;

@Override

public void handleTextMessage(WebSocketSession
session, TextMessage message)

throws Exception {

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

JsonObject.class);

UserSession
user = registry.getBySession(session);

if
(user != null) {

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

}
else {

log.debug("Incoming message from new user: {}", jsonMessage);

}

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

case "register":

try
{

register(session, jsonMessage);

}
catch (Throwable t) {

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

JsonObject response = new JsonObject();

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

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

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

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

}

break;

case "call":

try
{

call(user, jsonMessage);

}
catch (Throwable t) {

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

JsonObject response = new JsonObject();

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

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

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

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

}

break;

case
"incomingCallResponse":

incomingCallResponse(user, jsonMessage);

break;

case
"stop":

stop(session);

break;

default:

break;

}

}

private void register(WebSocketSession session, JsonObject jsonMessage)

throws IOException {

...

}

private void call(UserSession caller, JsonObject jsonMessage)

throws IOException {

...

}

private void incomingCallResponse(UserSession callee, JsonObject jsonMessage)

throws IOException {

...

}

public void stop(WebSocketSession session) throws IOException {

...

}

@Override

public void afterConnectionClosed(WebSocketSession session,

CloseStatus status) throws Exception {

registry.removeBySession(session);

}

}

在下面的代码片断中，我们可以看到注册方法，基本上，它包含了从注册信息中得到的名字属性，并检测它是否被注册过。

如果没有，则新用户被注册且有一个接受的消息发送给它；

private void register(WebSocketSession session, JsonObject jsonMessage)

throws IOException {

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

UserSession caller = new UserSession(session,
name);

String responseMsg = "accepted";

if (name.isEmpty()) {

responseMsg
= "rejected: empty user name";

} else if (registry.exists(name)) {

responseMsg
= "rejected: user '" + name + "' already registered";

} else {

registry.register(caller);

}

JsonObject response = new JsonObject();

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

response.addProperty("response", responseMsg);

caller.sendMessage(response);

}

在call方法中，服务端会检查在消息属性栏中的名字是否已注册，然后发送一个incomingCall消息给它。

或者，如果这个名字未注册，则会有一个callResponse消息发送给呼叫者以拒绝这次呼叫。

private void call(UserSession caller, JsonObject jsonMessage)

throws IOException {

String to = jsonMessage.get("to").getAsString();

String from = jsonMessage.get("from").getAsString();

JsonObject response = new JsonObject();

if (registry.exists(to)) {

UserSession
callee = registry.getByName(to);

caller.setSdpOffer(jsonMessage.getAsJsonPrimitive("sdpOffer").getAsString());

caller.setCallingTo(to);

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

response.addProperty("from",
from);

callee.sendMessage(response);

callee.setCallingFrom(from);

} else {

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

response.addProperty("response",
"rejected: user '" + to+ "' is not registered");

caller.sendMessage(response);

}

}

stop方法结束这次呼叫。这个过程会被呼叫者和被叫者在通信中被调用。

结果是这两端会释放媒体管道并结束通信:

public void stop(WebSocketSession session) throws IOException {

String sessionId = session.getId();

if (pipelines.containsKey(sessionId))
{

pipelines.get(sessionId).release();

CallMediaPipeline
pipeline = pipelines.remove(sessionId);

pipeline.release();

//
Both users can stop the communication. A 'stopCommunication'

//
message will be sent to the other peer.

UserSession
stopperUser = registry.getBySession(session);

UserSession
stoppedUser = (stopperUser.getCallingFrom() != null) ? registry

.getByName(stopperUser.getCallingFrom()) : registry

.getByName(stopperUser.getCallingTo());

JsonObject
message = new JsonObject();

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

stoppedUser.sendMessage(message);

}

}

在 incomingCallResponse方法中，如果被叫用户接受了这个呼叫，那么就会以B2B方式创建媒体元素并连接呼叫者与被叫者。

通常，服务端会创建一个 CallMediaPipeline对象，用来封装媒体管道的创建和管理。

然后，这个对象就用来在用户浏览器间进行媒体交互协商。

浏览器上WebRTC端点与Kurento Media Server的WebRtcEndpoint间的协商

是通过客户端生成的SDP(提交)与服务端生成的SDP(回答)实现的。

这个SDP的回答是由类CallMediaPipeline中Kurento Java Client生成的。

用于生成SDP的方法为generateSdpAnswerForCallee(calleeSdpOffer) 和 generateSdpAnswerForCaller(callerSdpOffer):

private void incomingCallResponse(UserSession callee, JsonObject jsonMessage)

throws IOException {

String callResponse = jsonMessage.get("callResponse").getAsString();

String from = jsonMessage.get("from").getAsString();

UserSession calleer = registry.getByName(from);

String to = calleer.getCallingTo();

if ("accept".equals(callResponse)) {

log.debug("Accepted
call from '{}' to '{}'", from, to);

CallMediaPipeline
pipeline = null;

try
{

pipeline = new CallMediaPipeline(kurento);

pipelines.put(calleer.getSessionId(), pipeline);

pipelines.put(callee.getSessionId(), pipeline);

String calleeSdpOffer = jsonMessage.get("sdpOffer").getAsString();

String calleeSdpAnswer = pipeline.generateSdpAnswerForCallee(calleeSdpOffer);

String callerSdpOffer = registry.getByName(from).getSdpOffer();

String callerSdpAnswer = pipeline.generateSdpAnswerForCaller(callerSdpOffer);

JsonObject startCommunication = new JsonObject();

startCommunication.addProperty("id", "startCommunication");

startCommunication.addProperty("sdpAnswer", calleeSdpAnswer);

callee.sendMessage(startCommunication);

JsonObject response = new JsonObject();

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

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

response.addProperty("sdpAnswer", callerSdpAnswer);

calleer.sendMessage(response);

}
catch (Throwable t) {

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

if (pipeline != null) {

pipeline.release();

}

pipelines.remove(calleer.getSessionId());

pipelines.remove(callee.getSessionId());

JsonObject response = new JsonObject();

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

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

calleer.sendMessage(response);

response = new JsonObject();

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

callee.sendMessage(response);

}

} else {

JsonObject
response = new JsonObject();

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

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

calleer.sendMessage(response);

}

}

这个DEMO的媒体逻辑是在类CallMediaPipeline中实现的，如上图所见，媒体管道的组成很简单:

由两个WebRtcEndpoint直接相连组成。需要注意的WebRtcEndpoints需要做两次连接，每次连接一个方向的。

public class CallMediaPipeline {

private MediaPipeline pipeline;

private WebRtcEndpoint callerWebRtcEP;

private WebRtcEndpoint calleeWebRtcEP;

public CallMediaPipeline(KurentoClient
kurento) {

try
{

this.pipeline = kurento.createMediaPipeline();

this.callerWebRtcEP = new WebRtcEndpoint.Builder(pipeline).build();

this.calleeWebRtcEP = new WebRtcEndpoint.Builder(pipeline).build();

this.callerWebRtcEP.connect(this.calleeWebRtcEP);

this.calleeWebRtcEP.connect(this.callerWebRtcEP);

}
catch (Throwable t) {

if(this.pipeline != null){

pipeline.release();

}

}

}

public String generateSdpAnswerForCaller(String
sdpOffer) {

return
callerWebRtcEP.processOffer(sdpOffer);

}

public String generateSdpAnswerForCallee(String
sdpOffer) {

return
calleeWebRtcEP.processOffer(sdpOffer);

}

public void release() {

if
(pipeline != null) {

pipeline.release();

}

}

}

在这个类中，我们可以看到方法generateSdpAnswerForCaller 和 generateSdpAnswerForCallee的实现，

这些方法引导WebRtc端点创建合适的回答。

5.3.4 客户端

现在来看应用程序客户端的代码。为了调用前面提到的服务端的WebSocket服务，我们使用了JavaScript类WebSocket。

我们使用了特殊的Kurento JavaScript库，叫做kurento-utils.js来简化WebRTC的交互，

这个库依赖于adapter.js，它是一个JavaScript WebRTC设备，由Google维护，用来抽象浏览器之间的差异。

最后，这个应用程序还需要jquery.js.

这些库都链接到了index.html页面中，并都在index.js中被使用。

在下面的代码片断中，我们可以看到在path /call下WebSocket(变量ws)的创建，

然后，WebSocket的监听者onmessage被用来实现在客户端的JSON信令协议。

.

注意，在客户端有四个输入信息:resgisterResponse, callResponse,incomingCall, 和startCommunication，

用来实现通信中的各个步骤。

例如，在函数 call and incomingCall (for caller and callee respectively)中，

kurento-utils.js的函数WebRtcPeer.startSendRecv用来启动WebRTC通信。

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

ws.onmessage = function(message) {

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

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

switch (parsedMessage.id) {

case 'resgisterResponse':

resgisterResponse(parsedMessage);

break;

case 'callResponse':

callResponse(parsedMessage);

break;

case 'incomingCall':

incomingCall(parsedMessage);

break;

case 'startCommunication':

startCommunication(parsedMessage);

break;

case 'stopCommunication':

console.info("Communication
ended by remote peer");

stop(true);

break;

default:

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

}

}

function incomingCall(message) {

//If bussy just reject without disturbing
user

if(callState != NO_CALL){

var
response = {

id : 'incomingCallResponse',

from : message.from,

callResponse : 'reject',

message : 'bussy'

};

return
sendMessage(response);

}

setCallState(PROCESSING_CALL);

if (confirm('User ' + message.from + '
is calling you. Do you accept the call?')) {

showSpinner(videoInput,
videoOutput);

webRtcPeer
= kurentoUtils.WebRtcPeer.startSendRecv(videoInput, videoOutput,

function(sdp,
wp) {

var
response = {

id : 'incomingCallResponse',

from : message.from,

callResponse : 'accept',

sdpOffer : sdp

};

sendMessage(response);

}, function(error){

setCallState(NO_CALL);

});

} else {

var
response = {

id : 'incomingCallResponse',

from : message.from,

callResponse : 'reject',

message : 'user declined'

};

sendMessage(response);

stop();

}

}

function call() {

if(document.getElementById('peer').value
== ''){

window.alert("You
must specify the peer name");

return;

}

setCallState(PROCESSING_CALL);

showSpinner(videoInput, videoOutput);

kurentoUtils.WebRtcPeer.startSendRecv(videoInput, videoOutput, function(offerSdp, wp) {

webRtcPeer = wp;

console.log('Invoking SDP offer callback
function');

var message = {

id : 'call',

from : document.getElementById('name').value,

to : document.getElementById('peer').value,

sdpOffer : offerSdp

};

sendMessage(message);

}, function(error){

console.log(error);

setCallState(NO_CALL);

});

}

5.3.5 依赖库

This Java Spring application is implementad using Maven.

The relevant part of the pom.xml is where Kurento dependencies are declared.

As the following snippet shows, we need two dependencies: the Kurento Client Java dependency

(kurento-client) and the JavaScript Kurento utility library (kurento-utils) for the client-side:

<dependencies>

<dependency>

<groupId>org.kurento</groupId>

<artifactId>kurento-client</artifactId>

<version>[5.0.0,6.0.0)</version>

</dependency>

<dependency>

<groupId>org.kurento</groupId>

<artifactId>kurento-utils-js</artifactId>

<version>[5.0.0,6.0.0)</version>

</dependency>

</dependencies>

Kurento framework uses Semantic Versioning for releases.

Notice that range [5.0.0,6.0.0) downloads the latest version of Kurento artefacts

from Maven Central in version 5 (i.e. 5.x.x). Major versions are released when incompatible changes are made.

Note: We are in active development. You can find the latest version of Kurento Java Client at Maven Central.

Kurento Java Client has a minimum requirement of Java 7.

To configure the application to use Java 7, we have to include the following properties in the properties section:

<maven.compiler.target>1.7</maven.compiler.target>

<maven.compiler.source>1.7</maven.compiler.source>