Rosetta/rosetta-wss
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Реализация динамического запроса TURN при подключении

Browse Source
This commit is contained in:
RoyceDa
2026-03-17 14:55:53 +02:00
parent ddcb54821f
commit 12a3e6266b
6 changed files with 81 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/im/rosetta/packet/Packet28IceServers.java
View File

@@ -19,7 +19,8 @@ public class Packet28IceServers extends Packet {
String url = stream.readString();
String username = stream.readString();
String credential = stream.readString();
RTCIceServer iceServer = new RTCIceServer(url, username, credential);
String transport = stream.readString();
RTCIceServer iceServer = new RTCIceServer(url, username, credential, transport);
iceServers.add(iceServer);
}
}
@@ -33,6 +34,7 @@ public class Packet28IceServers extends Packet {
stream.writeString(iceServer.getUrl());
stream.writeString(iceServer.getUsername());
stream.writeString(iceServer.getCredential());
stream.writeString(iceServer.getTransport());
}
return stream;
}

40
src/main/java/im/rosetta/service/services/ForwardUnitService.java
View File

@@ -28,13 +28,11 @@ public class ForwardUnitService {
private Logger logger;
private Set<SFU> sfuConnections = ConcurrentHashMap.newKeySet();
private ClientManager clientManager;
private Set<RTCIceServer> turnServers = new HashSet<>();
private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
public ForwardUnitService(Logger logger, ClientManager clientManager) {
this.logger = logger;
this.clientManager = clientManager;
this.readAllTurnServers();
this.sfuConnectionsSheduler();
}
@@ -65,37 +63,6 @@ public class ForwardUnitService {
}, 10, 10, TimeUnit.SECONDS);
}
/**
* Читаем все TURN сервера из переменной окружения и сохраняем их для дальнейшего
* использования при организации звонков через SFU серверы.
*/
private void readAllTurnServers() {
String turnServersEnv = System.getenv("TURN_SERVERS");
if(turnServersEnv == null || turnServersEnv.isEmpty()) {
this.logger.info(Color.YELLOW + "No TURN servers configured, skipping TURN servers boot");
return;
}
String[] turnServers = turnServersEnv.split(",");
for(String turnServer : turnServers) {
String[] parts = turnServer.split("@");
if(parts.length != 2) {
this.logger.error(Color.RED + "Invalid TURN server configuration: " + turnServer);
continue;
}
String address = parts[0];
String credentialsPart = parts[1];
String[] credentialsParts = credentialsPart.split(":");
if(credentialsParts.length != 2) {
this.logger.error(Color.RED + "Invalid TURN server credentials configuration: " + credentialsPart);
continue;
}
String username = credentialsParts[0];
String credential = credentialsParts[1];
RTCIceServer iceServer = new RTCIceServer(address, username, credential);
this.turnServers.add(iceServer);
}
}
/**
* Инициализирует соединения к SFU серверам для звонков.
* Ожидается, что адреса SFU серверов и секретные ключи для них будут переданы через переменную окружения SFU_SERVERS в формате "address1@secretKey1,address2@secretKey2,...".
@@ -239,6 +206,13 @@ public class ForwardUnitService {
* @return список серверов для RTC
*/
public Set<RTCIceServer> getTurnServers() {
Set<RTCIceServer> turnServers = new HashSet<>();
for(SFU sfu : this.sfuConnections) {
RTCIceServer turnServer = sfu.getTurnServer();
if(turnServer != null) {
turnServers.add(turnServer);
}
}
return turnServers;
}

48
src/main/java/io/g365sfu/SFU.java
View File

@@ -18,6 +18,7 @@ import io.g365sfu.net.Outgoing;
import io.g365sfu.net.SfuSock;
import io.g365sfu.util.StrUtils;
import io.g365sfu.webrtc.ICECandidate;
import io.g365sfu.webrtc.RTCIceServer;
import io.g365sfu.webrtc.SDPAnswer;
import io.g365sfu.webrtc.SDPOffer;
@@ -62,6 +63,13 @@ public class SFU {
*/
private Consumer<DisconnectedPeer> onPeerDisconnected;
/**
* TURN сервер предоставляемый SFU (если включен), который может быть использован
* для обмена кандидатами между участниками звонка через NAT и брандмауэры.
* Если SFU сервер не предоставляет TURN сервер, то это поле будет равно null.
*/
private RTCIceServer turnServer;
/**
* Конструктор для создания объекта SFU, который будет использоваться для установления соединения с SFU сервером.
* @param serverAddress адрес SFU сервера в формате "host:port", например "sfu.example.com:8080"
@@ -96,6 +104,10 @@ public class SFU {
if(!estabilished) {
throw new SFUHandshakeException("Failed to establish handshake with SFU server at " + this.serverAddress);
}
/**
* Спрашиваем про TURN
*/
this.askTurnServer();
}
private void onMessage(ByteBuffer message) {
@@ -215,6 +227,25 @@ public class SFU {
this.onPeerDisconnected.accept(disconnectedPeer);
}
}
if(packetId == Incoming.TURN_SERVER) {
int urlLength = message.getInt();
byte[] urlBytes = new byte[urlLength];
message.get(urlBytes);
String url = new String(urlBytes).trim();
int usernameLength = message.getInt();
byte[] usernameBytes = new byte[usernameLength];
message.get(usernameBytes);
String username = new String(usernameBytes).trim();
int credentialLength = message.getInt();
byte[] credentialBytes = new byte[credentialLength];
message.get(credentialBytes);
String credential = new String(credentialBytes).trim();
int transportLength = message.getInt();
byte[] transportBytes = new byte[transportLength];
message.get(transportBytes);
String transport = new String(transportBytes).trim();
this.turnServer = new RTCIceServer(url, username, credential, transport);
}
}
/**
@@ -274,6 +305,16 @@ public class SFU {
return room;
}
/**
* После успешного установления соединения и обменом handshake нужно узнать, поддерживает ли наш SFU встроенный TURN
*/
public void askTurnServer() {
ByteBuffer buffer = ByteBuffer.allocate(1);
buffer.put(Outgoing.ASK_TURN);
buffer.flip();
this.socket.send(buffer);
}
/**
* Получить все комнаты на сервере
* @return комнаты на этом сервере
@@ -355,4 +396,11 @@ public class SFU {
public void setDeleteRoomConsumer(Consumer<String> onDeleteRoom) {
this.onDeleteRoom = onDeleteRoom;
}
/**
* Возвращает TURN сервер на этом SFU
*/
public RTCIceServer getTurnServer() {
return turnServer;
}
}

6
src/main/java/io/g365sfu/net/Incoming.java
View File

@@ -52,4 +52,10 @@ public class Incoming {
*/
public static final byte PEER_DISCONNECTED = (byte) 0x11;
/**
* Вызывается когда сервер SFU отправляет TURN сервер (если он поддерживается), который может быть использован
* для обмена кандидатами между участниками звонка через NAT.
*/
public static final byte TURN_SERVER = (byte) 0x19;
}

6
src/main/java/io/g365sfu/net/Outgoing.java
View File

@@ -44,4 +44,10 @@ public class Outgoing {
*/
public static final byte ROOM_CREATE = (byte) 0x02;
/**
* Вызывается когда бекенд хочет спросить есть ли TURN сервер предоставляемый SFU, сервер ничего не ответит если
* TURN сервер не поддерживается. По умолчанию в G365SFU .env TURN сервер включен.
*/
public static final byte ASK_TURN = (byte) 0x19;
}

12
src/main/java/io/g365sfu/webrtc/RTCIceServer.java
View File

@@ -10,11 +10,13 @@ public class RTCIceServer {
private String url;
private String username;
private String credential;
private String transport;
public RTCIceServer(String url, String username, String credential) {
public RTCIceServer(String url, String username, String credential, String transport) {
this.url = url;
this.username = username;
this.credential = credential;
this.transport = transport;
}
/**
@@ -41,4 +43,12 @@ public class RTCIceServer {
return credential;
}
/**
* Транспортный протокол, используемый для связи с сервером ICE (например, "udp" или "tcp").
* @return строка, содержащая транспортный протокол, используемый для связи с сервером ICE (например, "udp" или "tcp").
*/
public String getTransport() {
return transport;
}
}