Реализация пртокола для сервера g365sfu

This commit is contained in:
RoyceDa
2026-03-14 22:56:24 +02:00
parent b84f69da33
commit 68cdec860d
8 changed files with 561 additions and 1 deletions

View File

@@ -0,0 +1,158 @@
package io.g365sfu;
import java.nio.ByteBuffer;
import java.util.HashSet;
/**
* Это комната для звонков, она может быть как для двоих участников, так и для групповых звонков.
* Комната содержит в себе информацию о том, на каком она SFU сервере, какой у нее ID, кто в ней участвует.
*/
public class Room {
private String roomId;
private SFU sfu;
private HashSet<String> participants;
/**
* Создать комнату с заданным ID, SFU сервером и участниками
* @param roomId уникальный идентификатор комнаты, который должен быть согласован с SFU сервером и использоваться для всех операций с этой комнатой
* @param sfu SFU сервер, на котором будет создана комната и через который будет происходить обмен данными между участниками
* @param participants массив идентификаторов участников, которые будут добавлены в комнату. Идентификаторы должны быть согласованы с SFU сервером и использоваться для маршрутизации данных между участниками.
*/
public Room(String roomId, SFU sfu, HashSet<String> participants) {
this.roomId = roomId;
this.sfu = sfu;
this.participants = participants;
}
public String getRoomId() {
return roomId;
}
public SFU getSfu() {
return sfu;
}
public HashSet<String> getParticipants() {
return participants;
}
public void addParticipant(String participantId) {
this.participants.add(participantId);
}
public void removeParticipant(String participantId) {
this.participants.remove(participantId);
}
public boolean containsParticipant(String participantId) {
return this.participants.contains(participantId);
}
/**
* Отправляет SDP offer в SFU сервер для организации звонка между пользователями.
* Этот метод используется для отправки предложения о соединении от одного участника
* к другому через сервер SFU, который будет пересылать это предложение целевому участнику.
* Параметры roomId и peerId используются для идентификации комнаты и целевого участника,
* а sdpOffer содержит описание медиа-сессии, которую участник хочет установить.
* @param roomId идентификатор комнаты, в которой участник хочет организовать звонок
* @param peerId идентификатор целевого участника, которому отправляется предложение о соединении
* @param sdpOffer строка, содержащая SDP offer, которая описывает медиа-сессию, которую участник хочет установить с целевым участником.
* @internal Этот метод формирует пакет с кодом 0x04, за которым следует идентификатор комнаты, идентификатор целевого участника и строка SDP offer.
*/
public void sdpOffer(String participantId, String sdpOffer) {
/**
* 1 байт номер пакета,
* 4 байта длина ID комнаты,
* N байт ID комнаты,
* 4 байта длина ID участника,
* M байт ID участника,
* 4 байта длина SDP offer,
* K байт SDP offer
*/
ByteBuffer buffer = ByteBuffer.allocate(
1 + 4 +
this.roomId.getBytes().length + 4 +
participantId.getBytes().length + 4 +
sdpOffer.getBytes().length);
/**
* 0x03 - SDP offer
*/
buffer.put((byte)0x03);
buffer.putInt(this.roomId.getBytes().length);
buffer.put(this.roomId.getBytes());
buffer.putInt(participantId.getBytes().length);
buffer.put(participantId.getBytes());
buffer.putInt(sdpOffer.getBytes().length);
buffer.put(sdpOffer.getBytes());
buffer.flip();
this.sfu.getConnection().send(buffer);
}
/**
* Отправляет ICE-кандидата в SFU сервер для одного из участников комнаты.
* @param participantId участник комнаты
* @param iceCandidate кандидат
*/
public void iceCandidate(String participantId, String iceCandidate) {
/**
* 1 байт номер пакета,
* 4 байта длина ID комнаты,
* N байт ID комнаты,
* 4 байта длина ID участника,
* M байт ID участника,
* 4 байта длина ICE кандидата,
* K байт ICE кандидата
*/
ByteBuffer buffer = ByteBuffer.allocate(
1 + 4 +
this.roomId.getBytes().length + 4 +
participantId.getBytes().length + 4 +
iceCandidate.getBytes().length);
/**
* 0x06 - ICE кандидат
*/
buffer.put((byte)0x06);
buffer.putInt(this.roomId.getBytes().length);
buffer.put(this.roomId.getBytes());
buffer.putInt(participantId.getBytes().length);
buffer.put(participantId.getBytes());
buffer.putInt(iceCandidate.getBytes().length);
buffer.put(iceCandidate.getBytes());
buffer.flip();
this.sfu.getConnection().send(buffer);
}
/**
* Отправляет SDP answer в SFU сервер для одного из участников комнаты.
* @param participantId участник комнаты
* @param sdpAnswer SDP answer
*/
public void sdpAnswer(String participantId, String sdpAnswer) {
/**
* 1 байт номер пакета,
* 4 байта длина ID комнаты,
* N байт ID комнаты,
* 4 байта длина ID участника,
* M байт ID участника,
* 4 байта длина SDP answer,
* K байт SDP answer
*/
ByteBuffer buffer = ByteBuffer.allocate(
1 + 4 +
this.roomId.getBytes().length + 4 +
participantId.getBytes().length + 4 +
sdpAnswer.getBytes().length);
/**
* 0x07 - SDP answer
*/
buffer.put((byte)0x07);
buffer.putInt(this.roomId.getBytes().length);
buffer.put(this.roomId.getBytes());
buffer.putInt(participantId.getBytes().length);
buffer.put(participantId.getBytes());
buffer.putInt(sdpAnswer.getBytes().length);
buffer.put(sdpAnswer.getBytes());
buffer.flip();
this.sfu.getConnection().send(buffer);
}
}

View File

@@ -1,13 +1,22 @@
package io.g365sfu;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import io.g365sfu.exception.SFUException;
import io.g365sfu.exception.SFUHandshakeException;
import io.g365sfu.net.SfuSock;
import io.g365sfu.util.StrUtils;
import io.g365sfu.webrtc.ICECandidate;
import io.g365sfu.webrtc.SDPAnswer;
import io.g365sfu.webrtc.SDPOffer;
public class SFU {
@@ -16,6 +25,28 @@ public class SFU {
private String secretKey;
private SfuSock socket;
private HashMap<String, CompletableFuture<String>> pendingRoomCreations = new HashMap<>();
/**
* Комнаты которые принадлежат этому серверу SFU. Ключом является ID комнаты,
* а значением объект Room, который содержит информацию о комнате, ее участниках и связанном с ней сервере SFU.
*/
private HashMap<String, Room> rooms = new HashMap<>();
/**
* Потребитель для обработки входящих ICE-кандидатов от сервера SFU.
* Этот потребитель будет вызываться при получении сообщения от сервера SFU с кодом 0x04,
* содержащим информацию об ICE-кандидате для одного из участников комнаты.
*/
private Consumer<ICECandidate> onIceCandidate;
/**
* Потребитель для обработки входящих SDP Answer от сервера SFU.
* Этот потребитель будет вызываться при получении сообщения от сервера SFU с кодом 0x05,
* содержащим информацию об SDP Answer для одного из участников комнаты.
*/
private Consumer<SDPAnswer> onSdpAnswer;
private Consumer<SDPOffer> onSdpOffer;
/**
* Конструктор для создания объекта SFU, который будет использоваться для установления соединения с SFU сервером.
* @param serverAddress адрес SFU сервера в формате "host:port", например "sfu.example.com:8080"
@@ -38,6 +69,7 @@ public class SFU {
*/
public void connect() throws URISyntaxException, InterruptedException, SFUException, ExecutionException, TimeoutException, SFUHandshakeException {
this.socket = new SfuSock(this.serverAddress);
this.socket.setMessageConsumer(this::onMessage);
boolean connected = this.socket.connectBlocking(30, TimeUnit.SECONDS);
if(!connected){
throw new SFUException("Failed to connect to SFU server, timeout after 30 seconds: " + this.serverAddress);
@@ -51,6 +83,94 @@ public class SFU {
}
}
private void onMessage(ByteBuffer message) {
if(message.remaining() < 1) {
System.err.println("Received empty message from SFU server");
return;
}
byte packetId = message.get(0);
if(packetId == 0x02) {
/**
* Ответ на создание комнаты, который содержит ID созданной комнаты
*/
int roomIdLength = message.getInt();
byte[] roomIdBytes = new byte[roomIdLength];
message.get(roomIdBytes);
String roomId = new String(roomIdBytes).trim();
CompletableFuture<String> future = this.pendingRoomCreations.remove(roomId);
if(future != null) {
future.complete(roomId);
}
return;
}
if(packetId == 0x04) {
/**
* ICE-candidate от сервера SFU для одного из участников комнаты
*/
int roomidLength = message.getInt();
byte[] roomIdBytes = new byte[roomidLength];
message.get(roomIdBytes);
String roomId = new String(roomIdBytes).trim();
int peerIdLength = message.getInt();
byte[] peerIdBytes = new byte[peerIdLength];
message.get(peerIdBytes);
String peerId = new String(peerIdBytes).trim();
int candidateLength = message.getInt();
byte[] candidateBytes = new byte[candidateLength];
message.get(candidateBytes);
String candidate = new String(candidateBytes).trim();
ICECandidate iceCandidate = new ICECandidate(roomId, peerId, candidate);
if(this.onIceCandidate != null) {
this.onIceCandidate.accept(iceCandidate);
}
return;
}
if(packetId == 0x05) {
/**
* Ответ на Offer от сервера SFU, который содержит SDP Answer
*/
int roomidLength = message.getInt();
byte[] roomIdBytes = new byte[roomidLength];
message.get(roomIdBytes);
String roomId = new String(roomIdBytes).trim();
int peerIdLength = message.getInt();
byte[] peerIdBytes = new byte[peerIdLength];
message.get(peerIdBytes);
String peerId = new String(peerIdBytes).trim();
int sdpAnswerLength = message.getInt();
byte[] sdpAnswerBytes = new byte[sdpAnswerLength];
message.get(sdpAnswerBytes);
String sdpAnswer = new String(sdpAnswerBytes).trim();
SDPAnswer answer = new SDPAnswer(roomId, peerId, sdpAnswer);
if(this.onSdpAnswer != null) {
this.onSdpAnswer.accept(answer);
}
return;
}
if(packetId == 0x08) {
/**
* Offer от сервера SFU для одного из участников комнаты при renegotiation
*/
int roomidLength = message.getInt();
byte[] roomIdBytes = new byte[roomidLength];
message.get(roomIdBytes);
String roomId = new String(roomIdBytes).trim();
int peerIdLength = message.getInt();
byte[] peerIdBytes = new byte[peerIdLength];
message.get(peerIdBytes);
String peerId = new String(peerIdBytes).trim();
int sdpOfferLength = message.getInt();
byte[] sdpAnswerBytes = new byte[sdpOfferLength];
message.get(sdpAnswerBytes);
String sdpOffer = new String(sdpAnswerBytes).trim();
SDPOffer offer = new SDPOffer(roomId, peerId, sdpOffer);
if(this.onSdpOffer != null) {
this.onSdpOffer.accept(offer);
}
return;
}
}
/**
* Получить адрес SFU сервера, к которому установлено соединение
* @return адрес SFU сервера
@@ -74,4 +194,109 @@ public class SFU {
public boolean isOpen() {
return this.socket != null && this.socket.isOpen();
}
/**
* Создает комнату на сервере SFU для организации звонков между пользователями. Комната автоматически удаляется
* при выходе последнего участника из нее. Внутри комнаты пользователи могут обмениваться аудио и видео потоками, а сервер SFU
* будет эффективно их пересылать между участниками, минимизируя задержки и оптимизируя использование пропускной способности.
* @throws TimeoutException
* @throws ExecutionException
* @throws InterruptedException
* @internal Этот метод формирует пакет с кодом 0x02,
* за которым следует случайно сгенерированный идентификатор комнаты,
* и отправляет его на сервер SFU.
*/
public Room createRoom() throws InterruptedException, ExecutionException, TimeoutException {
String roomId = StrUtils.randomString(64);
/**
* 1 байт номер пакета, 4 байта длина ID комнаты, N байт ID комнаты
*/
ByteBuffer buffer = ByteBuffer.allocate(1 + 4 + roomId.getBytes().length);
CompletableFuture<String> future = new CompletableFuture<>();
this.pendingRoomCreations.put(roomId, future);
/**
* 0x02 - создание комнаты
*/
buffer.put((byte)0x02);
buffer.putInt(roomId.getBytes().length);
buffer.put(roomId.getBytes());
buffer.flip();
this.socket.send(buffer);
String createdRoomId = future.get(30, TimeUnit.SECONDS);
Room room = new Room(createdRoomId, this, new HashSet<>());
this.rooms.put(createdRoomId, room);
return room;
}
/**
* Получить все комнаты на сервере
* @return комнаты на этом сервере
*/
public HashSet<Room> getRooms() {
return new HashSet<>(this.rooms.values());
}
/**
* Получить комнату по ее идентификатору
* @param roomId идентификатор комнаты
* @return объект Room, представляющий комнату с данным идентификатором, или null, если комната не найдена
*/
public Room getRoom(String roomId) {
return this.rooms.get(roomId);
}
/**
* Получить комнату, в которой участвует пользователь с данным идентификатором
* @param participantId идентификатор пользователя, который является участником комнаты
* @return объект Room, представляющий комнату, в которой участвует пользователь с данным идентификатором, или null, если такой комнаты не найдено
*/
public Room getRoomByParticipantId(String participantId) {
for(Room room : this.rooms.values()) {
if(room.containsParticipant(participantId)) {
return room;
}
}
return null;
}
/**
* Получает количество комнат на этом сервере
* @return возвращает количество комнат на сервере
*/
public int getRoomsCount() {
return this.rooms.size();
}
/**
* Устанавливает потребителя для обработки входящих ICE-кандидатов от сервера SFU.
* @param onIceCandidate потребитель, который будет вызываться при получении сообщения от сервера SFU с кодом 0x04,
* содержащим информацию об ICE-кандидате для одного из участников комнаты.
* Параметром будет объект ICECandidate, который содержит информацию о комнате, участнике и самом
* кандидате, необходимую для правильной маршрутизации данных между участниками звонка через сервер SFU.
*/
public void setIceConsumer(Consumer<ICECandidate> onIceCandidate) {
this.onIceCandidate = onIceCandidate;
}
/**
* Устанавливает потребителя для обработки входящих SDP Answer от сервера SFU.
* @param onSdpAnswer потребитель, который будет вызываться при получении сообщения от сервера SFU с кодом 0x05,
* содержащим информацию об SDP Answer для одного из участников комнаты.
* Параметром будет объект SDPAnswer, который содержит информацию о комнате, участнике и самом SDP Answer,
* необходимую для установления медиа-сессии между участником и сервером SFU.
*/
public void setAnswerConsumer(Consumer<SDPAnswer> onSdpAnswer) {
this.onSdpAnswer = onSdpAnswer;
}
/**
* Устанавливает потребителя для обработки входящих SDP Offer от сервера SFU при renegotiation.
* @param onSdpOffer потребитель, который будет вызываться при получении сообщения от сервера SFU с кодом 0x08,
* содержащим информацию об SDP Offer для одного из участников комнаты при renegotiation.
* Параметром будет объект SDPOffer, который содержит информацию о комнате, участнике и самом SDP Offer,
* необходимую для установления медиа-сессии между участником и сервером SFU.
*/
public void setOfferConsumer(Consumer<SDPOffer> onSdpOffer) {
this.onSdpOffer = onSdpOffer;
}
}

View File

@@ -4,6 +4,7 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
@@ -11,6 +12,7 @@ import org.java_websocket.handshake.ServerHandshake;
public class SfuSock extends WebSocketClient {
private CompletableFuture<Boolean> handshakeFuture = new CompletableFuture<>();
private Consumer<ByteBuffer> onMessage;
public SfuSock(String serverAddress) throws URISyntaxException {
super(new URI("ws://" + serverAddress));
@@ -42,6 +44,13 @@ public class SfuSock extends WebSocketClient {
this.handshakeFuture.complete(false);
return;
}
/**
* Если это не сообщение рукопожатия, то мы передаем его в установленного потребителя
*/
if(this.onMessage != null) {
this.onMessage.accept(bytes);
}
}
@Override
@@ -73,11 +82,17 @@ public class SfuSock extends WebSocketClient {
* Сервер SFU должен ответить одним байтом 0x01 для успешного рукопожатия или 0xFF для отклонения рукопожатия.
*/
public CompletableFuture<Boolean> handshakeExchange(String secretKey) {
ByteBuffer buffer = ByteBuffer.allocate(secretKey.length() + 1);
/**
* 1 байт номер пакета, 4 байта длина секретного ключа, N байт секретный ключ
*/
ByteBuffer buffer = ByteBuffer.allocate(
1 + 4 + secretKey.getBytes().length
);
/**
* 0x01 - код рукопожатия в соотвествии с протоколом g365sfu, за которым следует секретный ключ в виде строки байтов
*/
buffer.put((byte)0x01);
buffer.putInt(secretKey.getBytes().length);
buffer.put(secretKey.getBytes());
buffer.flip();
/**
@@ -87,4 +102,12 @@ public class SfuSock extends WebSocketClient {
return this.handshakeFuture;
}
/**
* Устанавливает потребителя для обработки входящих текстовых сообщений от сервера SFU.
* @param onMessage потребитель, который будет вызван с текстом сообщения при получении текстового сообщения от сервера SFU.
* @internal Этот метод позволяет установить обработчик для текстовых сообщений от сервера SFU
*/
public void setMessageConsumer(Consumer<ByteBuffer> onMessage) {
this.onMessage = onMessage;
}
}

View File

@@ -0,0 +1,15 @@
package io.g365sfu.util;
public class StrUtils {
public static String randomString(int length) {
String chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
StringBuilder sb = new StringBuilder();
for (int i = 0; i < length; i++) {
int index = (int) (Math.random() * chars.length());
sb.append(chars.charAt(index));
}
return sb.toString();
}
}

View File

@@ -0,0 +1,34 @@
package io.g365sfu.webrtc;
/**
* Этот класс представляет собой ICE-кандидат,
* который используется для обмена информацией о сетевых маршрутах между участниками звонка через сервер SFU.
*
* Содержит информацию о комнате, участнике и самом кандидате, которая необходима для правильной маршрутизации данных между участниками звонка через сервер SFU.
*/
public class ICECandidate {
private String roomId;
private String participantId;
private String candidate;
public ICECandidate(String roomId, String participantId, String candidate) {
this.roomId = roomId;
this.participantId = participantId;
this.candidate = candidate;
}
public String getRoomId() {
return roomId;
}
public String getParticipantId() {
return participantId;
}
public String getCandidate() {
return candidate;
}
}

View File

@@ -0,0 +1,44 @@
package io.g365sfu.webrtc;
/**
* Представляет собой объект RTCIceServer который содержит информацию о сервере ICE,
* например TURN
*/
public class RTCIceServer {
private String url;
private String username;
private String credential;
public RTCIceServer(String url, String username, String credential) {
this.url = url;
this.username = username;
this.credential = credential;
}
/**
* URL сервера ICE, который используется для обмена кандидатами между участниками звонка через сервер SFU.
* @return строка, содержащая URL сервера ICE, который используется для обмена кандидатами между участниками звонка через сервер SFU.
*/
public String getUrl() {
return url;
}
/**
* Имя пользователя для аутентификации на сервере ICE.
* @return строка, содержащая имя пользователя для аутентификации на сервере ICE.
*/
public String getUsername() {
return username;
}
/**
* Учетные данные для аутентификации на сервере ICE.
* @return строка, содержащая учетные данные для аутентификации на сервере ICE.
*/
public String getCredential() {
return credential;
}
}

View File

@@ -0,0 +1,31 @@
package io.g365sfu.webrtc;
/**
* Приходит с сервера SFU в ответ на отправленный SDP Offer от участника комнаты.
* Содержит информацию о комнате, участнике и самом SDP Answer, который необходим для установления медиа-сессии
* между участником и сервером SFU
*/
public class SDPAnswer {
private String roomId;
private String participantId;
private String sdp;
public SDPAnswer(String roomId, String participantId, String sdp) {
this.roomId = roomId;
this.participantId = participantId;
this.sdp = sdp;
}
public String getRoomId() {
return roomId;
}
public String getParticipantId() {
return participantId;
}
public String getSdp() {
return sdp;
}
}

View File

@@ -0,0 +1,30 @@
package io.g365sfu.webrtc;
/**
* Приходит от SFU сервера для конкретного участника комнаты. Обычно приходит при renegotiation,
* когда участник комнаты отправляет новый SDP Offer на сервер SFU, а сервер SFU отвечает ему новым SDP Answer,
* который содержит обновленную информацию о медиа-сессии для этого участника.
*/
public class SDPOffer {
private String roomId;
private String participantId;
private String sdp;
public SDPOffer(String roomId, String participantId, String sdp) {
this.roomId = roomId;
this.participantId = participantId;
this.sdp = sdp;
}
public String getRoomId() {
return roomId;
}
public String getParticipantId() {
return participantId;
}
public String getSdp() {
return sdp;
}
}