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>