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();
   }
}

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>