说明：
Webrtc p2p 建立的是端到端的连接，没有客户端、服务端的区分。在nas设备上，为了简化、易于理解，姑且把 nas 设备称之为服务端，app称之为客户端。
对于webrtc p2p 来说，通常任意一端，都可以主动发起连接，在 nas 环境下，我们简化为，由客户端主动发起连接，服务端被动等待连接。步骤如下：
1. nas设备/App向信令服务注册
private void connectSignalServer() {
    try {
        mUserId =  mFakeUserId.getText().toString().trim();
        DeviceMsgOuterClass.DeviceMsg deviceMsg = DeviceMsgOuterClass.DeviceMsg.newBuilder()
                .setUserId(mUserId)
                .setMsgType(DeviceMsgOuterClass.DevMsgType.UP_REGISTER_USER)
                .build();
        mSignalServer.sendToDeviceServer(deviceMsg.toByteArray());
    } catch (Throwable e) {
        Log.e(TAG, "error connectSignalServer", e);
    }
}
说明：先连接这个  "http://" + SERVER_ADDR_PORT + "/ws/devices";  服务
public void onMessage(WebSocket webSocket, ByteString bytes) {
    try {
        DeviceMsgOuterClass.DeviceMsg msg = DeviceMsgOuterClass.DeviceMsg.parseFrom(bytes.toByteArray());
        DeviceMsgOuterClass.DevMsgType type = msg.getMsgType();
        if (type == DeviceMsgOuterClass.DevMsgType.DOWN_CONNECT_SUCCESS) {
            byte[] data = msg.getData().toByteArray();
            DeviceMsgOuterClass.RegisteredInfo info = DeviceMsgOuterClass.RegisteredInfo.parseFrom(data);
            if (info != null) {
                mDeviceId = info.getGeneratedDeviceId();
                if (!TextUtils.isEmpty(mUserId)) {
                    connectToSignalServer();
                }
                if (mResultCallback != null) {
                    mResultCallback.onDeviceId(mDeviceId);
                }
            }
        } else if (type == DeviceMsgOuterClass.DevMsgType.DOWN_DEVICE_LIST) {
            byte[] data = msg.getData().toByteArray();
            DeviceMsgOuterClass.DeviceListAck list = DeviceMsgOuterClass.DeviceListAck.parseFrom(data);
            if (list != null) {
                ArrayList<String> deviceIds = new ArrayList<>();
                deviceIds.addAll(list.getDeviceIdsList());
                if (mResultCallback != null) {
                    mResultCallback.onDeviceList(deviceIds);
                }
            }
        }
        if (mResultCallback != null) {
            mResultCallback.onResult("DEVICE SERVER: " + msg);
        }
    } catch (Throwable e) {
        Log.e(TAG, "error onMessage", e);
    }
}
在上面的服务连接成功之后，会返回一个deviceId，拿这个deviceId，再去连接 "http://" + SERVER_ADDR_PORT + "/ws/signaling"; 这个真正的信令服务。
信令服务连接上，即表示注册成功。连接服务所需信息，参考：
private HashMap<String, String> buildDeviceParams() {
    HashMap<String, String> result = new HashMap<>();
    result.put("token", TEST_TOKEN);
    result.put("type", TREAT_AS_NAS_DEVICE ? "Nas" : "Android");
    if (!TextUtils.isEmpty(mUuid)) {
        result.put("uuid", mUuid);
    }
    if (!TextUtils.isEmpty(mDeviceId)) {
        result.put("deviceId", mDeviceId);
    }
    if (!TextUtils.isEmpty(mUserId)) {
        result.put("userId", mUserId);
    }
    return result;
}

2. 获取用户的在线设备列表
private void GetDeviceList() {
    try {
        mUserId = mFakeUserId.getText().toString().trim();
        DeviceMsgOuterClass.DeviceMsg deviceMsg = DeviceMsgOuterClass.DeviceMsg.newBuilder()
                .setUserId(mUserId)
                .setMsgType(DeviceMsgOuterClass.DevMsgType.UP_REQUEST_DEVICE_LIST)
                .build();
        mSignalServer.sendToDeviceServer(deviceMsg.toByteArray());
        mGetDeviceList.setEnabled(false);
    } catch (Throwable e) {
        Log.e(TAG, "error GetDeviceList", e);
    }
}
向信令服务发送消息，获取当前用户的在线设备列表（app获取到的是在线的 nas 设备列表，nas端获取到的是在线的 app 列表）

3. app端主动发起连接（nas设备也支持主动发起连接）
private void connectDevice(String deviceId) {
    if (TextUtils.isEmpty(deviceId)) {
        Log.e(TAG, "connectDevice: invalid deviceId");
        return;
    }
    if (mAsNasDevice.isChecked()) {
        // TODO
    } else {
        try {
            SignalMsgOuterClass.SignalMsg requectConnect = SignalMsgOuterClass.SignalMsg.newBuilder()
                    .setMsgType(SignalMsgOuterClass.SigMsgType.UP_CONNECT_DEVICE)
                    .setRemoteDeviceId(deviceId)
                    .build();
            mSignalServer.sendToSignalServer(requectConnect.toByteArray());
            Log.i(TAG, "connectDevice: " + requectConnect);
        } catch (Throwable e) {
            Log.e(TAG, "error connectDevice", e);
        }
    }
}
从获取到的设备列表里，选择要连接的目标设备，向信令服务发送连接请求

4. nas设备收到连接请求，开始准备 p2p 连接（准备offer）
public void onMessage(@NonNull WebSocket webSocket, @NonNull ByteString bytes) {
    try {
        SignalMsgOuterClass.SignalMsg signalMsg = SignalMsgOuterClass.SignalMsg.parseFrom(bytes.toByteArray());
        SignalMsgOuterClass.SigMsgType type = signalMsg.getMsgType();
        String remoteDeviceId = signalMsg.getRemoteDeviceId();
        int rtcOfferId = signalMsg.getRtcOfferId();
        if (type == SignalMsgOuterClass.SigMsgType.DOWN_REMOTE_SDP) {
            String remoteSdp = signalMsg.getSdp();
            if (mResultCallback != null) {
                mResultCallback.onRemoteSdp(rtcOfferId, remoteDeviceId, remoteSdp);
            }
            if (!TextUtils.isEmpty(remoteDeviceId) && !TextUtils.isEmpty(remoteSdp)) {
                SignalMsgOuterClass.SignalMsg reply = SignalMsgOuterClass.SignalMsg.newBuilder()
                        .setMsgType(SignalMsgOuterClass.SigMsgType.UP_ON_RECEIVE_SDP)
                        .setRemoteDeviceId(remoteDeviceId)
                        .setSdp(remoteSdp)
                        .setRtcOfferId(rtcOfferId)
                        .build();
                webSocket.send(ByteString.of(reply.toByteArray()));
                Log.i(TAG, "SignalWebsocket: onMessage, reply=" + reply);
            }
        } else if (type == SignalMsgOuterClass.SigMsgType.DOWN_REMOTE_ANSWER) {
            String remoteAnswer = signalMsg.getAnswer();
            if (mResultCallback != null) {
                mResultCallback.onRemoteAnswer(remoteDeviceId, remoteAnswer);
            }
        } else if (type == SignalMsgOuterClass.SigMsgType.DOWN_LOCAL_SDP_USED) {
            if (mResultCallback != null) {
                mResultCallback.onLocalOfferUsed(rtcOfferId, remoteDeviceId);
            }
        } else if (type == SignalMsgOuterClass.SigMsgType.DOWN_REQUIRE_CONNECT) {
            if (mResultCallback != null) {
                mResultCallback.onRequestConnect(remoteDeviceId);
            }
        } else if (type == SignalMsgOuterClass.SigMsgType.DOWN_REMOTE_CANDIDATE) {
            SignalMsgOuterClass.Candidate candidate = signalMsg.getCandidate();
            if (candidate != null) {
                String sdpMid = candidate.getSdpMid();
                int sdpMLineIndex = candidate.getSdpMLineIndex();
                String sdp = candidate.getCandidate();
                if (mResultCallback != null) {
                    mResultCallback.onRemoteIceCandidate(remoteDeviceId, sdpMid, sdpMLineIndex, sdp);
                }
            }
        }

        if (mResultCallback != null) {
            mResultCallback.onResult("SIGNAL SERVER: " + signalMsg);
        }
    } catch (Throwable e) {
        Log.e(TAG, "error onMessage", e);
    }
}
Nas 设备，在信令服务的 websocket 的 onmessage 回调里，收到 DOWN_REQUIRE_CONNECT 表示收到了客户端的webrtc请求，开始准备webrtc连接
private void startNewConnection(String remoteDeviceId) {
    WebrtcConnection connection = mLocalOfferConnections.get(remoteDeviceId);
    if (connection ==  null) {
        connection = new WebrtcConnection(this, mUseTurn.isChecked(), true, remoteDeviceId);
        connection.setWebrtcEventCallback(new WebrtcEventCallback());
        connection.setDataChannelObserver(new DataChannelObserver(connection));
        mLocalOfferConnections.put(remoteDeviceId, connection);
    }
    connection.startConnect(false);
}
public void startConnect(boolean reconnect) {
    try {
        initializePeerConnectionFactory();
        createPeerConnection();
    } catch (Throwable e) {
        Log.e(TAG, "Error connect", e);
        mPeerConnectionFactory = null;
    }
}
在 createPeerConnection 里，依次：
- 配置 RTCConfiguration （配置需要连接的turn、stun服务信息）
- mPeerConnectionFactory.createPeerConnection
- mPeerConnection.createDataChannel 创建两条 datachannel
- mPeerConnection.createOffer 创建 offer
  - 在该 offer 的回调 SdpObserver 的 onCreateSuccess 里，mPeerConnection.setLocalDescription，同时在 setLocalDescription 的 onSetSuccess 回调里（不同的平台接口名可能不太一样），将最终获取到的 session description，通知给信令服务
mPeerConnection.createOffer(new SdpObserver() {
    @Override
    public void onCreateSuccess(SessionDescription sessionDescription) {
        Log.i(TAG, "SdpObserver.onCreateSuccess, description=" + sessionDescription);
        if (sessionDescription != null) {
            Log.i(TAG, "    type=" + sessionDescription.type);
            Log.i(TAG, "    des=" + sessionDescription.description);
        }
        mPeerConnection.setLocalDescription(new SdpObserver() {
            @Override
            public void onCreateSuccess(SessionDescription sessionDescription) {
                Log.i(TAG, "AAAA onCreateSuccess=" + sessionDescription);
                if (sessionDescription != null) {
                    Log.i(TAG, "AAAA    type=" + sessionDescription.type);
                    Log.i(TAG, "AAAA    des=" + sessionDescription.description);
                }
            }

            @Override
            public void onSetSuccess() {
                Log.i(TAG, "AAAA onSetSuccess");
                if (sessionDescription != null && mWebrtcEventCallback != null) {
                    mWebrtcEventCallback.onLocalOffer(WebrtcConnection.this, sessionDescription.description);
                }
            }
public void onLocalOffer(WebrtcConnection connection, String description) {
    runOnUiThread(() -> { mRtcInfo.setText("localOffer=" + description); });
    try {
        SignalMsgOuterClass.SignalMsg.Builder builder = SignalMsgOuterClass.SignalMsg.newBuilder()
                .setMsgType(SignalMsgOuterClass.SigMsgType.UP_PUBLISH_SDP)
                .setRtcOfferId(connection.getConnectionId())
                .setSdp(description);
        if (!TextUtils.isEmpty(connection.getRemoteDeviceId())) {
            builder.setRemoteDeviceId(connection.getRemoteDeviceId());
        }
        SignalMsgOuterClass.SignalMsg publishSdp = builder.build();
        mSignalServer.sendToSignalServer(publishSdp.toByteArray());
        Log.i(TAG, "onLocalOffer: " + publishSdp);
    } catch (Throwable e) {
        Log.e(TAG, "error onLocalOffer", e);
    }
}

5. app端收到offer信息
public void onMessage(@NonNull WebSocket webSocket, @NonNull ByteString bytes) {
    try {
        SignalMsgOuterClass.SignalMsg signalMsg = SignalMsgOuterClass.SignalMsg.parseFrom(bytes.toByteArray());
        SignalMsgOuterClass.SigMsgType type = signalMsg.getMsgType();
        String remoteDeviceId = signalMsg.getRemoteDeviceId();
        int rtcOfferId = signalMsg.getRtcOfferId();
        if (type == SignalMsgOuterClass.SigMsgType.DOWN_REMOTE_SDP) {
            String remoteSdp = signalMsg.getSdp();
            if (mResultCallback != null) {
                mResultCallback.onRemoteSdp(rtcOfferId, remoteDeviceId, remoteSdp);
            }
            if (!TextUtils.isEmpty(remoteDeviceId) && !TextUtils.isEmpty(remoteSdp)) {
                SignalMsgOuterClass.SignalMsg reply = SignalMsgOuterClass.SignalMsg.newBuilder()
                        .setMsgType(SignalMsgOuterClass.SigMsgType.UP_ON_RECEIVE_SDP)
                        .setRemoteDeviceId(remoteDeviceId)
                        .setSdp(remoteSdp)
                        .setRtcOfferId(rtcOfferId)
                        .build();
                webSocket.send(ByteString.of(reply.toByteArray()));
                Log.i(TAG, "SignalWebsocket: onMessage, reply=" + reply);
            }
        }
收到nas端的offer信息后，分两步：
- 拿收到的offer信息创建 webrtc answer，并将 answer 回传给 信令服务
- 向信令发通知，收到 offer：UP_ON_RECEIVE_SDP
创建webrtc answer：
public void onRemoteSdp(int rtcOfferId, String remoteDeviceId, String remoteSdp) {
    WebrtcConnection connection = mRemoteSdpConnections.get(remoteDeviceId);
    if (connection == null) {
        connection = new WebrtcConnection(MainActivity.this, mUseTurn.isChecked(), false, remoteDeviceId);
        connection.setWebrtcEventCallback(new WebrtcEventCallback());
        connection.setDataChannelObserver(new DataChannelObserver(connection));
        mRemoteSdpConnections.put(remoteDeviceId, connection);
        connection.startConnect(true);
    }
    connection.setRemoteSdp(remoteSdp);
}
    public void setRemoteSdp(String remoteSdp) {
        if (mStartLocalOffer) {
            // already has local offer
            return;
        }
        if (mPeerConnection != null) {
            SessionDescription sdp = new SessionDescription(SessionDescription.Type.OFFER, remoteSdp);
            mPeerConnection.setRemoteDescription(new SdpObserver() {
                @Override
                public void onCreateSuccess(SessionDescription sessionDescription) {
                    Log.i(TAG, "RemoteOFFER.onSetSuccess, desc=" + sessionDescription);
                }

                @Override
                public void onSetSuccess() {
                    Log.i(TAG, "RemoteOFFER.onSetSuccess, sdp=" + remoteSdp);
                }

                @Override
                public void onCreateFailure(String s) {
                    Log.i(TAG, "RemoteOFFER.onCreateFailure=" + s);
                }

                @Override
                public void onSetFailure(String s) {
                    Log.i(TAG, "RemoteOFFER.onSetFailure=" + s);
                }
            }, sdp);

            MediaConstraints mediaConstraints = new MediaConstraints();
//            mediaConstraints.mandatory.add(new MediaConstraints.KeyValuePair("IceTransportType", "all"));

            mPeerConnection.createAnswer(new SdpObserver() {

                @Override
                public void onCreateSuccess(SessionDescription sessionDescription) {
                    Log.i(TAG, "LocalAnswer.onCreateSuccess, desc=" + sessionDescription);
                    if (sessionDescription != null) {
                        Log.i(TAG, "    type=" + sessionDescription.type);
                        Log.i(TAG, "    des=" + sessionDescription.description);
                    }
                    mPeerConnection.setLocalDescription(new SdpObserver() {
                        @Override
                        public void onCreateSuccess(SessionDescription sessionDescription) {
                            Log.i(TAG, "LocalAnswerSET.onCreateSuccess, desc=" + sessionDescription);
                            if (sessionDescription != null) {
                                Log.i(TAG, "SET    type=" + sessionDescription.type);
                                Log.i(TAG, "SET    des=" + sessionDescription.description);
                            }
                        }

                        @Override
                        public void onSetSuccess() {
                            Log.i(TAG, "LocalAnswerSET.onSetSuccess, desc=" + WebrtcConnection.this.mPeerConnection.getLocalDescription().description);
                            if (mWebrtcEventCallback != null) {
                                mWebrtcEventCallback.onLocalAnswer(WebrtcConnection.this, sessionDescription.description);
                            }
                            mCanSendIce = true;
                            handleCachedIceCandidates();
                        }

                        @Override
                        public void onCreateFailure(String s) {
                            Log.i(TAG, "LocalAnswerSET.onCreateFailure=" + s);
                        }

                        @Override
                        public void onSetFailure(String s) {
                            Log.i(TAG, "LocalAnswerSET.onSetFailure=" + s);
                        }
                    }, sessionDescription);
                }

                @Override
                public void onSetSuccess() {
                    Log.i(TAG, "LocalAnswer.onSetSuccess");
                }

                @Override
                public void onCreateFailure(String s) {
                    Log.i(TAG, "LocalAnswer.onCreateFailure=" + s);
                }

                @Override
                public void onSetFailure(String s) {
                    Log.i(TAG, "LocalAnswer.onSetFailure=" + s);
                }
            }, mediaConstraints);
        }
    }
跟创建offer一样，分如下几步：
- new SessionDescription(SessionDescription.Type.OFFER, remoteSdp);
- mPeerConnection.setRemoteDescription
- mPeerConnection.createAnswer，并在回调的onCreateSuccess 里，setLocalDescription，且再次在回调的onSetSuccess里，获取到真正的 answer，并将 answer 回传给 信令服务
public void onLocalAnswer(WebrtcConnection connection, String description) {
    runOnUiThread(() -> { mRtcInfo.setText("localAnswer=" + description); });
    try {
        SignalMsgOuterClass.SignalMsg publishAnswer = SignalMsgOuterClass.SignalMsg.newBuilder()
                .setMsgType(SignalMsgOuterClass.SigMsgType.UP_PUBLISH_ANSWER)
                .setRemoteDeviceId(connection.getRemoteDeviceId())
                .setAnswer(description)
                .build();
        mSignalServer.sendToSignalServer(publishAnswer.toByteArray());
        Log.i(TAG, "onLocalAnswer: " + publishAnswer);
    } catch (Throwable e) {
        Log.e(TAG, "error onLocalAnswer", e);
    }
}

6. nas设备收到客户端的 answer 信息
public void onMessage(@NonNull WebSocket webSocket, @NonNull ByteString bytes) {
    try {
        SignalMsgOuterClass.SignalMsg signalMsg = SignalMsgOuterClass.SignalMsg.parseFrom(bytes.toByteArray());
        SignalMsgOuterClass.SigMsgType type = signalMsg.getMsgType();
        String remoteDeviceId = signalMsg.getRemoteDeviceId();
        int rtcOfferId = signalMsg.getRtcOfferId();
        if (type == SignalMsgOuterClass.SigMsgType.DOWN_REMOTE_SDP) {
            String remoteSdp = signalMsg.getSdp();
            if (mResultCallback != null) {
                mResultCallback.onRemoteSdp(rtcOfferId, remoteDeviceId, remoteSdp);
            }
            if (!TextUtils.isEmpty(remoteDeviceId) && !TextUtils.isEmpty(remoteSdp)) {
                SignalMsgOuterClass.SignalMsg reply = SignalMsgOuterClass.SignalMsg.newBuilder()
                        .setMsgType(SignalMsgOuterClass.SigMsgType.UP_ON_RECEIVE_SDP)
                        .setRemoteDeviceId(remoteDeviceId)
                        .setSdp(remoteSdp)
                        .setRtcOfferId(rtcOfferId)
                        .build();
                webSocket.send(ByteString.of(reply.toByteArray()));
                Log.i(TAG, "SignalWebsocket: onMessage, reply=" + reply);
            }
        } else if (type == SignalMsgOuterClass.SigMsgType.DOWN_REMOTE_ANSWER) {
            String remoteAnswer = signalMsg.getAnswer();
            if (mResultCallback != null) {
                mResultCallback.onRemoteAnswer(remoteDeviceId, remoteAnswer);
            }
        }
public void onRemoteAnswer(String remoteDeviceId, String remoteAnswer) {
    WebrtcConnection connection = mLocalOfferConnections.get(remoteDeviceId);
    if (connection == null) {
        Log.e(TAG, "onRemoteAnswer: connection not found, remoteDeviceId=" + remoteDeviceId);
        return;
    }
    connection.setRemoteAnswer(remoteAnswer);
}
public void setRemoteAnswer(String remoteAnswer) {
    if (!mStartLocalOffer) {
        // already has local answer
        return;
    }
    if (mPeerConnection != null) {
        SessionDescription sdp = new SessionDescription(SessionDescription.Type.ANSWER, remoteAnswer);
        mPeerConnection.setRemoteDescription(new SdpObserver() {
            @Override
            public void onCreateSuccess(SessionDescription sessionDescription) {
                Log.i(TAG, "RemoteAnswer.onCreateSuccess, desc=" + sessionDescription);
                if (sessionDescription != null) {
                    Log.i(TAG, "    type=" + sessionDescription.type);
                    Log.i(TAG, "    des=" + sessionDescription.description);
                }
            }

            @Override
            public void onSetSuccess() {
                Log.i(TAG, "RemoteAnswer.onSetSuccess, answer=" + remoteAnswer);
                mCanSendIce = true;
                handleCachedIceCandidates();
            }

            @Override
            public void onCreateFailure(String s) {
                Log.i(TAG, "RemoteAnswer.onCreateFailure=" + s);
            }

            @Override
            public void onSetFailure(String s) {
                Log.i(TAG, "RemoteAnswer.onSetFailure=" + s);
            }
        }, sdp);
    }
}
核心是 setRemoteDescription，注意类型别填错

7. 两端互相发送 ice candidate
在创建 webrtc peer connection 的时候，第二个参数填的 observer 的回调里，收到的 ice 信息，在 answer、offer 设置成功之后，可以互相通过信令服务对发了。
设置成功之前收到的 ice 信息，先 缓存起来，在offer/answer 设置成功之后，将缓存的 ice 信息对发。
mPeerConnection = mPeerConnectionFactory.createPeerConnection(rtcConfig, mPeerConnectionObserver);
private class MyPeerConnectionObserver implements PeerConnection.Observer {
    @Override
    public void onSignalingChange(PeerConnection.SignalingState signalingState) {
        Log.i(TAG, "PeerConnectionObserver.onSignalingChange, state=" + signalingState);
    }

    @Override
    public void onIceConnectionChange(PeerConnection.IceConnectionState iceConnectionState) {
        Log.i(TAG, "PeerConnectionObserver.onIceConnectionChange, state=" + iceConnectionState);
        if (mWebrtcEventCallback != null) {
            mWebrtcEventCallback.onIceStateChanged(WebrtcConnection.this, iceConnectionState);
        }
    }

    @Override
    public void onIceConnectionReceivingChange(boolean b) {
        Log.i(TAG, "PeerConnectionObserver.onIceConnectionReceivingChange=" + b);
    }

    @Override
    public void onIceGatheringChange(PeerConnection.IceGatheringState iceGatheringState) {
        Log.i(TAG, "PeerConnectionObserver.onIceGatheringChange, state=" + iceGatheringState);
    }

    @Override
    public void onIceCandidate(IceCandidate iceCandidate) {
        Log.i(TAG, "PeerConnectionObserver.onIceCandidate=" + iceCandidate);
        if (iceCandidate != null) {
            if (mCanSendIce) {
                if (mWebrtcEventCallback != null) {
                    mWebrtcEventCallback.onIceCandidate(WebrtcConnection.this, iceCandidate.sdpMid, iceCandidate.sdpMLineIndex, iceCandidate.sdp);
                }
            } else {
                mCachedIceCandidates.add(iceCandidate);
            }
        }
    }

    @Override
    public void onIceCandidatesRemoved(IceCandidate[] iceCandidates) {
        Log.i(TAG, "PeerConnectionObserver.onIceCandidatesRemoved");
        if (!mCanSendIce && iceCandidates != null && iceCandidates.length > 0) {
            for (IceCandidate candidate : iceCandidates) {
                mCachedIceCandidates.remove(candidate);
                Log.i(TAG, "    removed candidate=" + candidate);
            }
        }
    }

    @Override
    public void onAddStream(MediaStream mediaStream) {
        Log.i(TAG, "PeerConnectionObserver.onAddStream");
    }

    @Override
    public void onRemoveStream(MediaStream mediaStream) {
        Log.i(TAG, "PeerConnectionObserver.onRemoveStream");
    }

    @Override
    public void onDataChannel(DataChannel dataChannel) {
        Log.i(TAG, "PeerConnectionObserver.onDataChannel=" + dataChannel);
        if (dataChannel != null) {
            dataChannel.registerObserver(new DataChannel.Observer() {
                @Override
                public void onBufferedAmountChange(long l) {
                    Log.i(TAG, "DataChannel.Observer.onBufferedAmountChange=" + l);
                    if (mDataChannelCallback != null) {
                        mDataChannelCallback.onBufferedAmountChange(l);
                    }
                }

                @Override
                public void onStateChange() {
                    Log.i(TAG, "DataChannel.Observer.onStateChange=" + dataChannel.state());
                    if (mDataChannelCallback != null) {
                        mDataChannelCallback.onStateChange();
                    }
                }

                @Override
                public void onMessage(DataChannel.Buffer buffer) {
                    String msg = "DataChannel.Observer.onMessage: from dev=" + getRemoteDeviceId() + "; data=" + buffer.data.toString();
                    Log.i(TAG, msg);
                    if (mDataChannelCallback != null) {
                        mDataChannelCallback.onMessage(buffer);
                    }
                }
            });
        }
    }

    @Override
    public void onRenegotiationNeeded() {
        Log.i(TAG, "PeerConnectionObserver.onRenegotiationNeeded");
    }

    @Override
    public void onAddTrack(RtpReceiver rtpReceiver, MediaStream[] mediaStreams) {
        Log.i(TAG, "PeerConnectionObserver.onAddTrack");
    }
}

8. 两端收到对端的 ice 信息之后，设置 对发的ice 信息
public void onMessage(@NonNull WebSocket webSocket, @NonNull ByteString bytes) {
    try {
        SignalMsgOuterClass.SignalMsg signalMsg = SignalMsgOuterClass.SignalMsg.parseFrom(bytes.toByteArray());
        SignalMsgOuterClass.SigMsgType type = signalMsg.getMsgType();
        String remoteDeviceId = signalMsg.getRemoteDeviceId();
        int rtcOfferId = signalMsg.getRtcOfferId();
        if (type == SignalMsgOuterClass.SigMsgType.DOWN_REMOTE_SDP) {
            String remoteSdp = signalMsg.getSdp();
            if (mResultCallback != null) {
                mResultCallback.onRemoteSdp(rtcOfferId, remoteDeviceId, remoteSdp);
            }
            if (!TextUtils.isEmpty(remoteDeviceId) && !TextUtils.isEmpty(remoteSdp)) {
                SignalMsgOuterClass.SignalMsg reply = SignalMsgOuterClass.SignalMsg.newBuilder()
                        .setMsgType(SignalMsgOuterClass.SigMsgType.UP_ON_RECEIVE_SDP)
                        .setRemoteDeviceId(remoteDeviceId)
                        .setSdp(remoteSdp)
                        .setRtcOfferId(rtcOfferId)
                        .build();
                webSocket.send(ByteString.of(reply.toByteArray()));
                Log.i(TAG, "SignalWebsocket: onMessage, reply=" + reply);
            }
        } else if (type == SignalMsgOuterClass.SigMsgType.DOWN_REMOTE_ANSWER) {
            String remoteAnswer = signalMsg.getAnswer();
            if (mResultCallback != null) {
                mResultCallback.onRemoteAnswer(remoteDeviceId, remoteAnswer);
            }
        } else if (type == SignalMsgOuterClass.SigMsgType.DOWN_LOCAL_SDP_USED) {
            if (mResultCallback != null) {
                mResultCallback.onLocalOfferUsed(rtcOfferId, remoteDeviceId);
            }
        } else if (type == SignalMsgOuterClass.SigMsgType.DOWN_REQUIRE_CONNECT) {
            if (mResultCallback != null) {
                mResultCallback.onRequestConnect(remoteDeviceId);
            }
        } else if (type == SignalMsgOuterClass.SigMsgType.DOWN_REMOTE_CANDIDATE) {
            SignalMsgOuterClass.Candidate candidate = signalMsg.getCandidate();
            if (candidate != null) {
                String sdpMid = candidate.getSdpMid();
                int sdpMLineIndex = candidate.getSdpMLineIndex();
                String sdp = candidate.getCandidate();
                if (mResultCallback != null) {
                    mResultCallback.onRemoteIceCandidate(remoteDeviceId, sdpMid, sdpMLineIndex, sdp);
                }
            }
        }
public void onRemoteIceCandidate(String remoteDeviceId, String sdpMid, int sdpMLineIndex, String sdp) {
    WebrtcConnection connection = mRemoteSdpConnections.get(remoteDeviceId);
    if (connection == null) {
        connection = mLocalOfferConnections.get(remoteDeviceId);
    }
    if (connection == null) {
        Log.e(TAG, "onRemoteIceCandidate: connection not found, remoteDeviceId=" + remoteDeviceId);
        return;
    }
    connection.setRemoteIceCandidate(sdpMid, sdpMLineIndex, sdp);
}
public void setRemoteIceCandidate(String sdpMid, int sdpMLineIndex, String sdp) {
    if (mPeerConnection != null) {
        IceCandidate candidate = new IceCandidate(sdpMid, sdpMLineIndex, sdp);
        Log.i(TAG, "setRemoteIceCandidate:" + candidate);
        mPeerConnection.addIceCandidate(candidate);
    }
}

9. 等待 datachannel 回调，可以进行 p2p 通信
等待 ice 信息互相匹配成功之后，在 第 7 步里的 MyPeerConnectionObserver 里的 ondatachanel 回调里，收到真正的 datachannel，就可以通过datachannnel 与 对端进行通信里
public void onDataChannel(DataChannel dataChannel) {
    Log.i(TAG, "PeerConnectionObserver.onDataChannel=" + dataChannel);
    if (dataChannel != null) {
        dataChannel.registerObserver(new DataChannel.Observer() {
            @Override
            public void onBufferedAmountChange(long l) {
                Log.i(TAG, "DataChannel.Observer.onBufferedAmountChange=" + l);
                if (mDataChannelCallback != null) {
                    mDataChannelCallback.onBufferedAmountChange(l);
                }
            }

            @Override
            public void onStateChange() {
                Log.i(TAG, "DataChannel.Observer.onStateChange=" + dataChannel.state());
                if (mDataChannelCallback != null) {
                    mDataChannelCallback.onStateChange();
                }
            }

            @Override
            public void onMessage(DataChannel.Buffer buffer) {
                String msg = "DataChannel.Observer.onMessage: from dev=" + getRemoteDeviceId() + "; data=" + buffer.data.toString();
                Log.i(TAG, msg);
                if (mDataChannelCallback != null) {
                    mDataChannelCallback.onMessage(buffer);
                }
            }
        });
    }
}
在 onDataChannel 收到的 dataChannel 才能真正的用于的 p2p 通信


