Правильная обработка DisconnectReason при сборсах и обрывах соединения

.env

@@ -21,7 +21,3 @@ BUFFER_CLEANUP_DAYS=7
 #SFU Сервера
 SFU_SERVERS=127.0.0.1:1001@SFU_TEST_SECRET
 
-#TURN Сервера (должны поддерживать TCP и UDP протоколы)
-# Формат: host:port@username:password через запятую если их несколько, без пробелов
-TURN_SERVERS=10.211.55.2:3478@user:pass
-

src/main/java/im/rosetta/executors/Executor26SignalPeer.java

@@ -36,6 +36,21 @@ public class Executor26SignalPeer extends PacketExecutor<Packet26SignalPeer> {
             return;
         }
         NetworkSignalType type = packet.getSignalType();
+        if(type == NetworkSignalType.CALL) {
+            /**
+             * Инициируется звонок от src к dst, проверяем, что dst не занят другим звонком, если занят, то отправляем сигнал END_CALL_BECAUSE_BUSY обратно src
+             */
+            Room room = this.fus.getRoomByParticipantId(packet.getDst());
+            if(room != null) {
+                /**
+                 * Получатель сигнала уже находится в другой комнате, значит он занят другим звонком, отправляем сигнал END_CALL_BECAUSE_BUSY обратно src
+                 */
+                Packet26SignalPeer responsePacket = new Packet26SignalPeer();
+                responsePacket.setSignalType(NetworkSignalType.END_CALL_BECAUSE_BUSY);
+                this.clientManager.sendPacketToAuthorizedPK(packet.getSrc(), responsePacket);
+                return;
+            }
+        }
         if(type == NetworkSignalType.CREATE_ROOM){
             /**
              * Создается комната для звонка

src/main/java/im/rosetta/packet/Packet26SignalPeer.java

@@ -38,6 +38,9 @@ public class Packet26SignalPeer extends Packet {
     @Override
     public void read(Stream stream) {
         this.signalType = NetworkSignalType.fromCode(stream.readInt8());
+        if(this.signalType == NetworkSignalType.END_CALL_BECAUSE_BUSY || this.signalType == NetworkSignalType.END_CALL_BECAUSE_PEER_DISCONNECTED) {
+            return;
+        }
         this.src = stream.readString();
         this.dst = stream.readString();
         if (signalType == NetworkSignalType.KEY_EXCHANGE) {
@@ -53,6 +56,9 @@ public class Packet26SignalPeer extends Packet {
         Stream stream = new Stream();
         stream.writeInt16(this.packetId);
         stream.writeInt8(this.signalType.getCode());
+        if(this.signalType == NetworkSignalType.END_CALL_BECAUSE_BUSY || this.signalType == NetworkSignalType.END_CALL_BECAUSE_PEER_DISCONNECTED) {
+            return stream;
+        }
         stream.writeString(this.src);
         stream.writeString(this.dst);
         if (signalType == NetworkSignalType.KEY_EXCHANGE) {

src/main/java/im/rosetta/packet/runtime/NetworkSignalType.java

@@ -23,7 +23,15 @@ public enum NetworkSignalType {
     /**
      * Создание комнаты
      */
-    CREATE_ROOM(4);
+    CREATE_ROOM(4),
+    /**
+     * Обрыв связи с пиром
+     */
+    END_CALL_BECAUSE_PEER_DISCONNECTED(5),
+    /**
+     * Не удалось дозвониться - пользователь занят другим звонком
+     */
+    END_CALL_BECAUSE_BUSY(6);
 
     private final int code;
 

src/main/java/im/rosetta/service/services/ForwardUnitService.java

@@ -10,10 +10,14 @@ import java.util.concurrent.TimeUnit;
 import im.rosetta.client.ClientManager;
 import im.rosetta.logger.Logger;
 import im.rosetta.logger.enums.Color;
+import im.rosetta.packet.Packet26SignalPeer;
 import im.rosetta.packet.Packet27WebRTC;
+import im.rosetta.packet.runtime.NetworkSignalType;
 import im.rosetta.packet.runtime.NetworkWebRTCType;
 import io.g365sfu.Room;
 import io.g365sfu.SFU;
+import io.g365sfu.net.DisconnectReason;
+import io.g365sfu.net.DisconnectedPeer;
 import io.g365sfu.webrtc.ICECandidate;
 import io.g365sfu.webrtc.RTCIceServer;
 import io.g365sfu.webrtc.SDPAnswer;
@@ -113,6 +117,14 @@ public class ForwardUnitService {
                         e.printStackTrace();
                     }
                 });
+                connection.setPeerDisconnectedConsumer(arg0 -> {
+                    try{
+                        onPeerDisconnected(arg0);
+                    }catch(ProtocolException e){
+                        this.logger.error(Color.RED + "Failed to handle peer disconnection from SFU server: " + address + ", error: " + e.getMessage());
+                        e.printStackTrace();
+                    }
+                });
                 this.sfuConnections.add(connection);
                 this.logger.info(Color.GREEN + "Successfully connected to SFU server: " + address);
             } catch (Exception e) {
@@ -121,6 +133,35 @@ public class ForwardUnitService {
         }
     }
     
+    public void onPeerDisconnected(DisconnectedPeer disconnectedPeer) throws ProtocolException {
+        Room room = disconnectedPeer.getRoom();
+        if(disconnectedPeer.getReason() != DisconnectReason.FAILED){
+            /**
+             * Если у нас произошло штатное отключение, а не в результате обрыва связи - то не нужно отправлять
+             * оппонентам пакеты о том, что участник отключился в результате обрыва связи.
+             */
+            return;
+        }
+        for(String peerId : room.getParticipants()) {
+            /**
+             * Уведомляем все пиры, что соединение с пиром было потеряно
+             */
+            if(room.getParticipants().size() == 1) {
+                /**
+                 * Звонок был завершен, так как в комнате остался только один участник, который не может продолжать звонок в одиночку.
+                 */
+                Packet26SignalPeer packet = new Packet26SignalPeer();
+                packet.setSignalType(NetworkSignalType.END_CALL_BECAUSE_PEER_DISCONNECTED);
+                this.clientManager.sendPacketToAuthorizedPK(peerId, packet);
+            }
+        }
+    }
+
+    /**
+     * Выполняется когда сервер SFU отправляет SDP answer для одного из участников комнаты.
+     * @param sdpAnswer объект SDPAnswer, который содержит информацию о комнате, участнике и самом answer,
+     * @throws ProtocolException 
+     */
     public void onSdpAnswer(SDPAnswer sdpAnswer) throws ProtocolException {
         String participantId = sdpAnswer.getParticipantId();
         Packet27WebRTC packet = new Packet27WebRTC();

src/main/java/io/g365sfu/SFU.java

@@ -13,6 +13,7 @@ import java.util.function.Consumer;
 
 import io.g365sfu.exception.SFUException;
 import io.g365sfu.exception.SFUHandshakeException;
+import io.g365sfu.net.DisconnectReason;
 import io.g365sfu.net.DisconnectedPeer;
 import io.g365sfu.net.Incoming;
 import io.g365sfu.net.Outgoing;
@@ -215,13 +216,14 @@ public class SFU {
             message.get(peerIdBytes);
             String peerId = new String(peerIdBytes).trim();
             Room room = this.rooms.get(roomId);
-            DisconnectedPeer disconnectedPeer = new DisconnectedPeer(peerId, roomId, room);
+            DisconnectReason reason = io.g365sfu.net.DisconnectReason.fromCode(message.getInt());
             if(room != null) {
                 /**
                  * Если такая комната существует то удаляем оттуда участника
                  */
                 room.removeParticipant(peerId);
             }
+            DisconnectedPeer disconnectedPeer = new DisconnectedPeer(peerId, roomId, room, reason);
             if(this.onPeerDisconnected != null) {
                 this.onPeerDisconnected.accept(disconnectedPeer);
             }
@@ -396,6 +398,14 @@ public class SFU {
         this.onDeleteRoom = onDeleteRoom;
     }
 
+    /**
+     * Устанавливает потребителя для обработки сообщений от сервера SFU о том, что участник отключился от комнаты.
+     * @param onPeerDisconnected потребитель, который будет вызываться при получении сообщения от сервера SFU с кодом 0x11,
+     */
+    public void setPeerDisconnectedConsumer(Consumer<DisconnectedPeer> onPeerDisconnected) {
+        this.onPeerDisconnected = onPeerDisconnected;
+    }
+
     /**
      * Возвращает TURN сервер на этом SFU
      */

src/main/java/io/g365sfu/net/DisconnectReason.java

@@ -0,0 +1,25 @@
+package io.g365sfu.net;
+
+public enum DisconnectReason {
+    FAILED(0),
+    CLOSED(1);
+
+    private final int code;
+
+    DisconnectReason(int code) {
+        this.code = code;
+    }
+
+    public int getCode() {
+        return code;
+    }
+
+    public static DisconnectReason fromCode(int code) {
+        for (DisconnectReason reason : DisconnectReason.values()) {
+            if (reason.code == code) {
+                return reason;
+            }
+        }
+        throw new IllegalArgumentException("Unknown DisconnectReason code: " + code);
+    }
+}

src/main/java/io/g365sfu/net/DisconnectedPeer.java

@@ -7,11 +7,13 @@ public class DisconnectedPeer {
     private String peerId;
     private Room room;
     private String roomId;
+    private DisconnectReason reason;
 
-    public DisconnectedPeer(String peerId, String roomId, Room room) {
+    public DisconnectedPeer(String peerId, String roomId, Room room, DisconnectReason reason) {
         this.peerId = peerId;
         this.room = room;
         this.roomId = roomId;
+        this.reason = reason;
     }
 
     public String getPeerId() {
@@ -25,4 +27,8 @@ public class DisconnectedPeer {
     public String getRoomId() {
         return roomId;
     }
+
+    public DisconnectReason getReason() {
+        return reason;
+    }
 }