Kurnto Java中定义的一些接口、实现与关系
如下图，Openvidu Server作为webrtc中的信号服务器，在分别与客户端浏览器、KMS通信时都使用了websocket.
与客户端通信时，接收来自客户端的加入会议、发布视频，订阅视频等请求，并且转发连接需要的SDP offer 和 ICE candidate。
与服务器通信时，创建pipleLine，webrtc mediaElement，并编排媒体流，转发ICE canddiate等。
image.png
这些请求和响应通过websocket来完成，通信内容通过JsonRPC方式编码。所需的功能主要由kurento提供的几个类库来完成：kurento-client、kurento-jsonrpc-server, kurento-jsonrpc-client.
kurento-jsonrpc-client：定义了符合jsonrpc通讯的一些接口和实现。
kurento-jsonrpc-server: 定义了充当websocket server端时需要的一些管理类。websocket实际是使用spring的websocket库来完成，而该包在此基础上完成jsonrpc的编码和解码逻辑。
kurento-client： 定义了作为websocket客户端与KMS通信的接口与实现。
以下是相关的一些接口和类：
image.png
-
Message：
是jsonrpc消息的最上层接口，包含一个sessionid属性。它有两个实现，request代表接收到的jssonrpc 请求， response代表应该返回的响应。这一个接口、两个实现对应的是kurento的定义的通信协议(可参考kurento通信协议)。 -
Transaction:
Transaction 代表的是客户端和服务器之间的一次通信。这个接口最主要的功能就是定义了sendResponse(...)接口。AbstractTransaction是虚类，它定义了internalSendResponse(Response)接口，并用这个接口实现了transaction中定义的各种sendResponse接口。可以这么理解，使用Transaction作为消息相应媒介，最终都需要将各种参数转化为Response，并通过internalSendResponse的实现发送出去。 -
TransactionImp:
kurento jsonrpc项目中唯一的一个transaction实现（继承abstractTransation），但它并没有告诉我们消息具体是怎么发出去的；它定义了一个关键接口 RespondSender，同时拥有一个RespondSender 的属性， 通过这个属性，TransactionImp完成了internalSendResponse的消息发送。 所以，创建一个Transaction的时候，需要给它具体的RespondSender实现。有了具体的实现，其它类就可以利用Transaction 发送相应消息了。 -
RespondSender:
定义了两个接口方法:
public interface ResponseSender {
void sendResponse(Message message) throws IOException;
void sendPingResponse(Message message) throws IOException;
}
这个接口一个用于发送普通相应，一个用于发送心跳。在kurento项目中有两个地方会实现这个接口，一个作为websocket服务器端与浏览器通信；一个作为websocket 客户端与kurento通信。
- ProtocolManager：
ProtocolManager是一个重要的bean，会在openvidu初始化的时后创建，需要注意的是它被定义为@prototype类型。
@Bean
@Scope("prototype")
public ProtocolManager protocolManager(JsonRpcHandler<?> key) {
return new ProtocolManager(key);
}
ProtocolManager有一个重要方法processMessage，这个方法会将接收来的消息跟据类型不同进行处理。连接类型的消息protocolManager自己就处理了;而业务类型的消息，它会调用handlerManager.handleRequest(session, request, responseSender)来处理。
- JsonRpcWebSocketHandler:
它是spring websocket实现框架中，TextWebSocketHandler具体的实现类. 在spring websocket框架中，所有websocket接收到的消息都会被TextWebSocketHandler定义的handleTextMessage方法处理。这个方法签名中，wsSession代表的时一次websocket会话，可以直接用于发送消息。
public void handleTextMessage(final WebSocketSession wsSession, TextMessage message)
JsonRpcWebSocketHandler实现了这个接口，在这个接口中创建了一个具体的ResponseSender 实现，并把它传给了ProtocolManager
protocolManager.processMessage(messageJson, factory, new ResponseSender() {
@Override
public void sendResponse(Message message) throws IOException {
String jsonMessage = message.toString();
log.debug("{} Res<- {}", label, jsonMessage);
sendJsonMessage(jsonMessage);
}
@Override
public void sendPingResponse(Message message) throws IOException {
String jsonMessage = message.toString();
log.trace("{} Res<- {}", label, jsonMessage);
sendJsonMessage(jsonMessage);
}
private void sendJsonMessage(String jsonMessage) throws IOException {
synchronized (wsSession) {
if (wsSession.isOpen()) {
***wsSession.sendMessage(new TextMessage(jsonMessage));***
} else {
log.error("Trying to send a message to a closed session");
}
}
}
}, wsSession.getId());
这行代码中， 创建了一个匿名的ResponseSender对象，从这个对象的实现可以看到，所有的响应消息最终是通过WebSocketSession 发送出去了。
- JsonRpcHandlerManager:
它是JsonRpcHandler的管理类，在这个类的handleRequest方法中，定义了消息如何被处理。
public void handleRequest(Session session, Request<JsonElement> request, ResponseSender rs) {
try {
if (handler == null) {
log.warn("JsonRpcClient has received a request from server but"
+ " there is no JsonRpcHandler configured to manage this" + " request");
return;
}
Class<?> paramsType = getParamsType(handler.getHandlerType());
Request<?> nonGenRequest;
try {
nonGenRequest = MessageUtils.convertRequest(request, paramsType);
} catch (ClassCastException e) {
String message = "The handler " + handler.getClass()
+ " is trying to process the request. But request params '" + request.getParams()
+ "' cannot be converted to " + paramsType.getCanonicalName()
+ ". The type to convert params is specified in the"
+ " handler as the supertype generic parameter";
// TODO Maybe use the pattern handleUncaughtException
log.error(message, e);
if (request.getId() != null) {
rs.sendResponse(new Response<>(null, new ResponseError(0, message)));
}
return;
}
JsonRpcHandler nonGenHandler = handler;
TransactionImpl tx = new TransactionImpl(session, request, rs);
nonGenHandler.handleRequest(tx, nonGenRequest);
//如果不是异步，且responded是false(可能表示不需要respond),则返回一个空response.
if (!tx.isAsync() && request.getId() != null) {
log.debug("Request {} is processed asynchronously", request);
boolean notResponded = tx.setRespondedIfNot();
if (notResponded) {
// Empty response
rs.sendResponse(new Response<>(request.getId(), ""));
}
}
} catch (Exception e) {
// TODO Maybe use the pattern handleUncaughtException
log.error("Exception while processing request {}", request, e);
ResponseError error = ResponseError.newFromException(e);
try {
rs.sendResponse(new Response<>(request.getId(), error));
} catch (IOException e1) {
log.error("Exception sending error to client", e1);
}
}
}
上面的这段代买最重要的是：
JsonRpcHandler nonGenHandler = handler;
TransactionImpl tx = new TransactionImpl(session, request, rs);
nonGenHandler.handleRequest(tx, nonGenRequest);
可以看到，最终消息被包裹在transaction中，被JsonRpcHandler 的具体实现类处理。但是，JsonRpcHandlerManager中并没有定义JsonRpcHandler 的实现类，只是作为接口引用。 所以最终消息是如何被处理的，是由JsonRpcHandler 的实现决定。
Openvidu项目就定义了这个rpcHandler. 他作为全局的bean来处理所有的websocket消息。
- JsonRpcClientNettyWebSocket：
JsonRpcClientNettyWebSocket是AbstractJsonRpcClientWebSocket的具体实现，AbstractJsonRpcClientWebSocket定义了configureResponseSender（)方法，会在连接server的时候初始化ResponseSender.
void configureResponseSender() {
rs = new ResponseSender() {
@Override
public void sendResponse(Message message) throws IOException {
String jsonMessage = message.toString();
log.debug("{} <-Res {}", label, jsonMessage);
sendTextMessage(jsonMessage);
}
@Override
public void sendPingResponse(Message message) throws IOException {
String jsonMessage = message.toString();
log.trace("{} <-Res {}", label, jsonMessage);
sendTextMessage(jsonMessage);
}
};
}
这个ResponseSender发消息的实现很明显是依赖sendTextMessage(String jsonMessage)接口，而这个接口的实现正式在JsonRpcClientNettyWebSocket中，它是websocket 的客户端实现，通过netty库提供的网络接口发送消息。
JsonRpcClientNettyWebSocket作为接收端，它接收到的消息，都调用receivedTextMessage(String message)接口处理，这个接口在abstract父类中已经实现，如果接收到的是请求(kms请求)，他会起子线程，调用handlerManager处理。
handlerManager.handleRequest(session, fromJsonRequest(message, JsonElement.class), rs);
可以看到无论是来自浏览器端的websocket客户端请求， 还是来自kms 的websocket服务端请求，最终都是由handlerManager来处理。区别仅仅在于Session和RS对应的场景不同。
根据以上的描述，Transaction定义了发送响应的接口，在TransactionImp中通过定义新接口RespondSender, 作为模板方式实现所有transaction定义的方法。 RespondSender的实现类是真正的发送响应消息提供这。当transaction被用于openvidu的websocket服务端（由spring @eanbleWebsocket提供）处理请求时，RespondSender的实现类实际上就封装了textWebsocketHander提供的websocketSession, websocketSession 用于发送消息； 当transaction被用于openvidu的websocket客户（由JsonRpcNettyWebsocket提供客户端能力）端请求时，RespondSender的实现类实际上封装JsonRpcNettyWebsocket对sendTextMessage的实现。
openvidu在处理客户端websocket请求和服务器端websocket请求，都用了kurentJsonrpc包。虽然由很大的抽象，但也带来了很多困扰，特别是Session 和Transaction，由于场景不同，可能接收、发送于不同的地方。这在阅读openvidu 代码时非常的费力。如果能够把于浏览器的通信和kms的通信更好的定义和隔离开，这个项目的代码阅读性和维护性就会大大提高。
网友评论