diff --git a/.env b/.env index db4a9c6..c21b809 100644 --- a/.env +++ b/.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 - diff --git a/src/main/java/im/rosetta/executors/Executor26SignalPeer.java b/src/main/java/im/rosetta/executors/Executor26SignalPeer.java index 674b864..9ea8147 100644 --- a/src/main/java/im/rosetta/executors/Executor26SignalPeer.java +++ b/src/main/java/im/rosetta/executors/Executor26SignalPeer.java @@ -36,6 +36,21 @@ public class Executor26SignalPeer extends PacketExecutor { 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){ /** * Создается комната для звонка diff --git a/src/main/java/im/rosetta/packet/Packet26SignalPeer.java b/src/main/java/im/rosetta/packet/Packet26SignalPeer.java index b72f672..ffb37a1 100644 --- a/src/main/java/im/rosetta/packet/Packet26SignalPeer.java +++ b/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) { diff --git a/src/main/java/im/rosetta/packet/runtime/NetworkSignalType.java b/src/main/java/im/rosetta/packet/runtime/NetworkSignalType.java index d062cd0..7ee79bd 100644 --- a/src/main/java/im/rosetta/packet/runtime/NetworkSignalType.java +++ b/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; diff --git a/src/main/java/im/rosetta/service/services/ForwardUnitService.java b/src/main/java/im/rosetta/service/services/ForwardUnitService.java index 02069d4..b07a9ab 100644 --- a/src/main/java/im/rosetta/service/services/ForwardUnitService.java +++ b/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(); diff --git a/src/main/java/io/g365sfu/SFU.java b/src/main/java/io/g365sfu/SFU.java index a2aa358..370922c 100644 --- a/src/main/java/io/g365sfu/SFU.java +++ b/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 onPeerDisconnected) { + this.onPeerDisconnected = onPeerDisconnected; + } + /** * Возвращает TURN сервер на этом SFU */ diff --git a/src/main/java/io/g365sfu/net/DisconnectReason.java b/src/main/java/io/g365sfu/net/DisconnectReason.java new file mode 100644 index 0000000..21f8ccf --- /dev/null +++ b/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); + } +} diff --git a/src/main/java/io/g365sfu/net/DisconnectedPeer.java b/src/main/java/io/g365sfu/net/DisconnectedPeer.java index d74e2fd..36d83fe 100644 --- a/src/main/java/io/g365sfu/net/DisconnectedPeer.java +++ b/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; + } }