red5源码分析—客户端处理connect命令并发送createStream命令

在《red5源码分析—5》中提到过，当客户端发送connect命令后，服务器经过处理会将其connect命令返回，不同的是服务器返回的结果包含了一些连接后需要发送给客户端的信息，包括服务器版本、模式等等。当返回的信息经过服务器的发送过滤器RTMPMinaProtocolEncoder时，会调用其中的RTMPProtocolEncoder的encodeCommand函数，下面来看其中的一段代码，

    protected void encodeCommand(IoBuffer out, ICommand command) {
        Output output = new org.red5.io.amf.Output(out);
        final IServiceCall call = command.getCall();
        final boolean isPending = (call.getStatus() == Call.STATUS_PENDING);
        if (!isPending) {
            Serializer.serialize(output, call.isSuccess() ? "_result" : "_error");
        } else {
            final String action = (call.getServiceName() == null) ? call.getServiceMethodName() : call.getServiceName() + ‘.‘ + call.getServiceMethodName();
            Serializer.serialize(output, action);
        }
        ...
    }

这里可以简单看出，如果经过处理后的call的状态不是STATUS_PENDING，并且处理成功，则返回”_result”，否则返回”_error”。是否处理成功由Call的isSuccess方法判断，

    public boolean isSuccess() {
        return (status == STATUS_SUCCESS_RESULT) || (status == STATUS_SUCCESS_NULL) || (status == STATUS_SUCCESS_VOID);
    }

因此，只要Call的状态是STATUS_SUCCESS_RESULT、STATUS_SUCCESS_NULL或者STATUS_SUCCESS_VOID就可以判定为处理成功了。如果call的状态是STATUS_PENDING，则根据情况返回方法名，或者服务名加上方法名。

根据上面的分析结果回顾服务器对connect命令的处理，如果连接成功连接上服务器的某个scope后，便会设置Call的状态为STATUS_SUCCESS_RESULT，也因此返回的方法名(serviceMethodName)为_result。因此，当服务器处理完connect命令后消息返回客户端时，

便调用BaseRTMPClientHandler的onCommand函数，如下

    protected void onCommand(RTMPConnection conn, Channel channel, Header source, ICommand command) {
        final IServiceCall call = command.getCall();
        final String methodName = call.getServiceMethodName();
        if ("_result".equals(methodName) || "_error".equals(methodName)) {
            final IPendingServiceCall pendingCall = conn.getPendingCall(command.getTransactionId());
            if (pendingCall != null) {
                if ("connect".equals(methodName)) {
                    Integer encoding = (Integer) connectionParams.get("objectEncoding");
                    if (encoding != null && encoding.intValue() == 3) {
                        conn.getState().setEncoding(IConnection.Encoding.AMF3);
                    }
                }
            }
            handlePendingCallResult(conn, (Invoke) command);
            return;
        }
        ...
    }

根据前面的分析，当服务器返回的消息到达客户端后，最后会调用handlePendingCallResult函数进行处理，handlePendingCallResult定义在BaseRTMPHandler中，

    protected void handlePendingCallResult(RTMPConnection conn, Invoke invoke) {
        final IServiceCall call = invoke.getCall();
        final IPendingServiceCall pendingCall = conn.retrievePendingCall(invoke.getTransactionId());
        if (pendingCall != null) {
            Object[] args = call.getArguments();
            if (args != null && args.length > 0) {
                pendingCall.setResult(args[0]);
            }
            Set<IPendingServiceCallback> callbacks = pendingCall.getCallbacks();
            if (!callbacks.isEmpty()) {
                HashSet<IPendingServiceCallback> tmp = new HashSet<IPendingServiceCallback>();
                tmp.addAll(callbacks);
                for (IPendingServiceCallback callback : tmp) {
                    try {
                        callback.resultReceived(pendingCall);
                    } catch (Exception e) {

                    }
                }
            }
        }
    }

首先通过retrievePendingCall根据事务ID获取pendingCall，

    public IPendingServiceCall retrievePendingCall(int invokeId) {
        return pendingCalls.remove(invokeId);
    }

该pendingCall是在TCP建立连接后的调用BaseRTMPClientHandler的connectionOpened中根据事务ID注册的(《red5源码分析—3》)。获得PendingCall后，就设置结果，这里注意一点，也就是服务器最后设置的结果results经过RTMP的编解码(amf)处理后到达这里会变为参数params(具体的可以查看编解码器RTMPProtocolEncoder和RTMPProtocolDecoder的对应关系)。最后通过resultReceived调用之前注册的回调函数，根据《red5源码分析—1》，该注册的回调函数如下，

    public void resultReceived(IPendingServiceCall call) {
        Object result = call.getResult();
        if (result instanceof ObjectMap) {
            if ("connect".equals(call.getServiceMethodName())) {
                createStream(this);
            }
        } else {
            if ("createStream".equals(call.getServiceMethodName())) {
                if (result instanceof Integer) {
                    Integer streamIdInt = (Integer) result;
                    // int streamId = streamIdInt.intValue();
                    // publish(streamId, "testgio2", "live", this);
                    invoke("getRoomsInfo", this);
                } else {
                    disconnect();
                }
            } else if ("getRoomsInfo".equals(call.getServiceMethodName())) {
                ArrayList<String> list = (ArrayList<String>) result;
                for (int i = 0; i < list.size(); i++) {
                    System.out.println(list.get(i));
                }
            }
        }
    }

因此，客户端再收到connect命令的结果后立即调用createStream来创建一个流，

    public void createStream(IPendingServiceCallback callback) {
        IPendingServiceCallback wrapper = new CreateStreamCallBack(callback);
        invoke("createStream", null, wrapper);
    }

CreateStreamCallBack和处理回调函数有关系，回头再分析。下面就是发送一个”createStream”的命令给客户端了。下一章看看服务器是如何处理”createStream”命令的。