Merge pull request 'Видеосообщения (кружочки)' (#19) from dev into main

Reviewed-on: #19

serviceAccount.json

@@ -1,13 +0,0 @@
-{
-  "type": "service_account",
-  "project_id": "rosetta-messanger-dev",
-  "private_key_id": "69822675ff0a49a8ce0bd147dbab3ef432963485",
-  "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC/iYhrgWfi/ecP\nIs7Y0u7MueE/leU0QZF3V6hIcs1Iq1/CSUYcszv+YXuKItfwi0pYtntPmnCtdKnv\nFgZZN/9I0/AjFPlggqEXBH223FjsAfoNTk8a0PXyF8oOG0u2pN34t20nxXFSYars\n/6yJMKOBdaJ+L51dlBeCj40UFm3zQSVpejRkspWmT5Qz7kWh2e8yj6ukPyXEfrrD\n8jmM6uQeKwzlmfjD8YbL7R2m+lE1hzkoFCo0PrqMQYIIpAMT9n9meCk/LrQMIUnA\n+zICsTffFwbFzmJ2A022po3ObL4dUYgqGrPs5DqjxwTF4mI0g2dkPvoxpUKkVzC7\nHzzOYTwRAgMBAAECggEAGIacDSa1rpdpfK2DLrjFHY+YXAPYehpX7fVaVeXmtcKa\nppkAXSYcQq9T2jUqUSGyOZgq7l9Yb0rhhFJd1L3QDBDLY9/zK3GpU/ZeF0oG1DhP\n6bVDJCwUBNXD/eNujKrMG5AimOBLlEtvpMemXa4jOa1eSxR+F4tM1AndUZr/pZFT\nIdPBrId3WemkcWH2b1R4RLVu8zqci1x+nWvL4Gt8O/gDpDzlmV2Disvpf300a5wU\ngIKcOI2H6vouOwL9ltMPTBXdpDIC64Q8zvK3973d3Fqx2Mt2mO4ZgylFpk2nmAm6\n3wknrkJDE+Z/H7XmdwpMhRmmyku3vuqsDwmy7BEGGwKBgQD03oOJ6eb9kG6JX830\nRH/Vhh3NlpVEzuHzRA8WznsX5WaeyMF9NWo4Lh9ifBxkRAQKftEC0CR5GZp3/895\nQuoMMbrrklWWkVRwCZWYQBquRXNHbTPj6ztvHaZZrw8HhemZWtOF89T5f8Kn4LcD\nFwR8kIqXeLxeTbfc2hQXZvP7vwKBgQDIPmhM2Td/Ni4Mbe3wGmOyS0RweK5qcKBF\n32N1rCr8smrD1irg+w9Qrmz2cX3udtVJDRLz4ctSmiEPs+UrzuyG6yTbdv7bh8AN\nsqHOdc4GRqGsB/jfXagVjRMSeObQk5fR03tiPLP15J+lpoEsjwUzw7L3Ioacaq62\nqSi87br8LwKBgQDu/uMJz4bJg5evcxeUSusuH5mlGE0WfIniIlJL4zoXR6qSXcUk\nDOdgb/vn5tTbM9tx1vbvNPH0VH4Ek2QPqbTANCWJWSk6LRxpwaEFmcOwxk5Or5IO\n6X/34suDCy6zHAu0xwZe3m7HGeCGc/iMBoI1heoPDyNjM5257AviD3UhBwKBgCWS\nmC17QI+NEfzhD5lSykwlFVVpP4jXUytpLBdjU7mQnLncULVgRlJkOCvRxchd4c1Q\nN7MtNeJs6zEwFxsuO3FhY8wOOunkQeQQFY5QynShAiruYANBZo2Mp/x6VQzj9MO5\nQ9h9/WJxIIeLg4dh2p8I5Ga8wrdMyTWa7frtPH2fAoGALUkccsu4Mcws55Xy1aHH\nNETI6CRQpaU4v6jNhuuY4g6jp5ScaqkpuuzHOYorpmdC3180YXzPn2pwKeYWa0BA\nTjR8z603bCpdRfpic1UvYi96jwAcNiJTAGGVwmyYopIFBdHphfGS3hWZlYfonVps\n+k87j3WGwOiUGqoKSGJi1cw=\n-----END PRIVATE KEY-----\n",
-  "client_email": "firebase-adminsdk-fbsvc@rosetta-messanger-dev.iam.gserviceaccount.com",
-  "client_id": "115421173243098464717",
-  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
-  "token_uri": "https://oauth2.googleapis.com/token",
-  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
-  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/firebase-adminsdk-fbsvc%40rosetta-messanger-dev.iam.gserviceaccount.com",
-  "universe_domain": "googleapis.com"
-}

src/main/java/im/rosetta/Boot.java

@@ -116,7 +116,7 @@ public class Boot {
         int cleanupEveryDays = System.getenv("BUFFER_CLEANUP_DAYS") != null ? 
             Integer.parseInt(System.getenv("BUFFER_CLEANUP_DAYS")) : 7;
         this.bufferCleanupService = new BufferCleanupService(cleanupEveryDays, this.logger);
-        this.forwardUnitService = new ForwardUnitService(this.logger, this.clientManager);
+        this.forwardUnitService = new ForwardUnitService(this.logger, this.clientManager, this.callManager);
     }
 
     /**

src/main/java/im/rosetta/Failures.java

@@ -27,7 +27,11 @@ public enum Failures implements BaseFailures {
     /**
      * Слишком много подписок на онлайн статусы
      */
-    TOO_MANY_ONLINE_SUBSCRIPTIONS(3010);
+    TOO_MANY_ONLINE_SUBSCRIPTIONS(3010),
+    /**
+     * Нет сессии звонка
+     */
+    NO_CALL_SESSION(3011);
 
     private final int code;
 

src/main/java/im/rosetta/calls/CallManager.java

@@ -9,6 +9,7 @@ import java.util.concurrent.TimeUnit;
 import im.rosetta.client.ClientManager;
 import im.rosetta.packet.Packet26SignalPeer;
 import im.rosetta.packet.runtime.NetworkSignalType;
+import io.g365sfu.Room;
 import io.orprotocol.ProtocolException;
 import io.orprotocol.client.Client;
 import io.orprotocol.packet.Packet;
@@ -17,13 +18,13 @@ public class CallManager {
 
     private List<CallSession> callSessions = new ArrayList<>();
     private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
-    private static final long RINGING_TIMEOUT = 10 * 1000; // 1 минута
+    private static final long RINGING_TIMEOUT = 30 * 1000;
     private ClientManager clientManager;
 
 
     public CallManager(ClientManager clientManager) {
         this.clientManager = clientManager;
-        scheduler.scheduleAtFixedRate(this::cleanupCallSessions, 0, 30, TimeUnit.SECONDS);
+        scheduler.scheduleAtFixedRate(this::cleanupCallSessions, 0, 1, TimeUnit.SECONDS);
     }
 
     public void cleanupCallSessions() {
@@ -75,6 +76,15 @@ public class CallManager {
         return null;
     }
 
+    public CallSession getCallSession(Room room) {
+        for (CallSession session : this.callSessions) {
+            if (session.getRoom() != null && session.getRoom().equals(room)) {
+                return session;
+            }
+        }
+        return null;
+    }
+
     public boolean isBusy(String publicKey) {
         for (CallSession session : this.callSessions) {
             if (session.clients.containsKey(publicKey)) {

src/main/java/im/rosetta/calls/CallSession.java

@@ -18,7 +18,9 @@ public class CallSession {
      * Клиенты в этом списке не могут принимать другие звонки, так как они уже заняты дозвоном,
      * но они еще не в звонке, так как не приняли звонок
      * 
-     * Клиенты удаляются из этого списка, когда они принимают звонок или отклоняют его, тогда они либо переходят в звонок, либо становятся свободными для других звонков
+     * Клиенты удаляются из этого списка, когда они принимают звонок или отклоняют его, 
+     * тогда они либо переходят в звонок, либо становятся свободными для других звонков
+     * pk -> время начала дозвона (timestamp в миллисекундах)
      */
     public HashMap<String, Long> ringing;
     /**
@@ -48,6 +50,16 @@ public class CallSession {
         this.clients.put(publicKey, client);
     }
 
+    /**
+     * Проверяет, может ли этот публичный ключ выполнять какие-либо действия в рамках этой сессии звонка,
+     * чтобы не допустить выполнение действий от посторонних публичных ключей, которые не участвуют в звонке
+     * @param publicKey Публичный ключ для проверки
+     * @return true, если этот публичный ключ может выполнять действия в рамках этой сессии звонка, false иначе
+     */
+    public boolean isValidSource(String publicKey) {
+        return this.ringing.containsKey(publicKey) || this.clients.containsKey(publicKey);
+    }
+
     /**
      * Получаем публичный ключ клиента по его сокету, чтобы понимать, кто отправляет сигналы в рамках звонка
      * @param client Сокет клиента, для которого нужно получить публичный ключ
@@ -70,6 +82,15 @@ public class CallSession {
         return this.joinToken;
     }
 
+    public void leaveCall(String publicKey) {
+        if(this.clients.containsKey(publicKey)) {
+            this.clients.remove(publicKey);
+        }
+        if(this.ringing.containsKey(publicKey)) {
+            this.ringing.remove(publicKey);
+        }
+    }
+
     public void addRinging(String publicKey) {
         this.ringing.put(publicKey, System.currentTimeMillis());
     }

src/main/java/im/rosetta/client/ClientManager.java

@@ -56,9 +56,10 @@ public class ClientManager {
      * Отправить пакет ВСЕМ АВТОРИЗОВАННЫМ клиентам с публичным ключом publicKey
      * @param publicKey публичный ключ получателя
      * @param packet пакет для отправки
+     * @param exclude клиент, который не должен получать этот пакет, может быть null
      * @throws ProtocolException если произошла ошибка при отправке пакета клиенту
      */
-    public void sendPacketToAuthorizedPK(String publicKey, Packet packet) throws ProtocolException {
+    public void sendPacketToAuthorizedPK(String publicKey, Packet packet, Client exclude) throws ProtocolException {
         Set<Client> clients = this.clientIndexer.getClients(ECIAuthentificate.class, "publicKey", publicKey);
         if(clients == null){
             /**
@@ -77,12 +78,55 @@ public class ClientManager {
                 continue;
             }
             /**
-             * Отправляем пакет каждому клиенту с таким публичным ключом (то есть всем его авторизованным сессиям/устройствам)
+             * Отправляем пакет каждому клиенту с таким публичным ключом (то есть всем его авторизованным сессиям/устройствам),
+             * исключая клиента exclude
              */
+            if(exclude != null && client.equals(exclude)){
+                /**
+                 * Этот клиент является исключением, он не должен получать этот пакет
+                 */
+                continue;
+            }
             client.send(packet);
         }
     }
 
+    /**
+     * Отправить пакет ВСЕМ АВТОРИЗОВАННЫМ клиентам с публичным ключом publicKey
+     * @param publicKey публичный ключ получателя
+     * @param packet пакет для отправки
+     * @throws ProtocolException если произошла ошибка при отправке пакета клиенту
+     */
+    public void sendPacketToAuthorizedPK(String publicKey, Packet packet) throws ProtocolException {
+        this.sendPacketToAuthorizedPK(publicKey, packet, null);
+    }
+
+    /**
+     * Отправить пакет всем клиентам с публичными ключами из списка publicKeys
+     * @param publicKeys список публичных ключей получателей
+     * @param packet пакет для отправки
+     * @param exclude клиент, который не должен получать этот пакет, может быть null
+     * @throws ProtocolException если произошла ошибка при отправке пакета клиенту
+     */
+    public void sendPacketToAuthorizedPK(List<String> publicKeys, Packet packet, Client exclude) throws ProtocolException {
+        for(String publicKey : publicKeys){
+            this.sendPacketToAuthorizedPK(publicKey, packet, exclude);
+        }
+    }
+
+    /**
+     * Отправить пакет всем клиентам с публичными ключами из списка publicKeys
+     * @param publicKeys список публичных ключей получателей
+     * @param packet пакет для отправки
+     * @throws ProtocolException если произошла ошибка при отправке пакета клиенту
+     */
+    public void sendPacketToAuthorizedPK(List<String> publicKeys, Packet packet) throws ProtocolException {
+        for(String publicKey : publicKeys){
+            this.sendPacketToAuthorizedPK(publicKey, packet, null);
+        }
+    }
+
+
     /**
      * Отправить пакет всем клиентам с публичными ключом как у client, кроме клиента client, который является отправителем и не должен получать этот пакет
      * @param client клиент
@@ -111,18 +155,6 @@ public class ClientManager {
         }
     }
 
-    /**
-     * Отправить пакет всем клиентам с публичными ключами из списка publicKeys
-     * @param publicKeys список публичных ключей получателей
-     * @param packet пакет для отправки
-     * @throws ProtocolException если произошла ошибка при отправке пакета клиенту
-     */
-    public void sendPacketToAuthorizedPK(List<String> publicKeys, Packet packet) throws ProtocolException {
-        for(String publicKey : publicKeys){
-            this.sendPacketToAuthorizedPK(publicKey, packet);
-        }
-    }
-
     /**
      * Получить список клиентов по публичному ключу (get PublicKey clients), могут быть неавторизованные клиенты
      * @param publicKey публичный ключ клиента

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

@@ -64,7 +64,7 @@ public class Executor26SignalPeer extends PacketExecutor<Packet26SignalPeer> {
             client.disconnect(Failures.HANDSHAKE_NOT_COMPLETED);
             return;
         }
-        if(!src.equals(eciAuthentificate.getPublicKey()) && authentificatedTypes.contains(type)) {
+        if(src != null && !src.equals(eciAuthentificate.getPublicKey()) && authentificatedTypes.contains(type)) {
             /**
              * Если src в пакете не совпадает с авторизованным PK клиента, то это может означать, что клиент пытается
              * отправить сигнал от другого пользователя, отключаем его от сервера.
@@ -124,6 +124,13 @@ public class Executor26SignalPeer extends PacketExecutor<Packet26SignalPeer> {
                 /**
                  * Сессии звонка нет
                  */
+                client.disconnect(Failures.NO_CALL_SESSION);
+                return;
+            }
+            if(!session.isValidSource(src)) {
+                /**
+                 * Клиент не состоит в сессии звонка, отключаем его от сервера, так как он отправляет некорректные данные
+                 */
                 client.disconnect(Failures.DATA_MISSMATCH);
                 return;
             }
@@ -134,6 +141,13 @@ public class Executor26SignalPeer extends PacketExecutor<Packet26SignalPeer> {
                 room.addParticipant(participant);
             }
             session.sendPacket(packet, client);
+            /**
+             * Сбрасываем вызов на всех остальных устройствах пользователя, который принимает звонок,
+             * чтобы он не смог принять или отклонить звонок с другого устройства
+             */
+            Packet26SignalPeer endCallOtherDevices = new Packet26SignalPeer();
+            endCallOtherDevices.setSignalType(NetworkSignalType.END_CALL);
+            this.clientManager.sendPacketToAuthorizedPK(src, endCallOtherDevices, client);
             return;
         }
         if(type == NetworkSignalType.KEY_EXCHANGE){
@@ -145,7 +159,7 @@ public class Executor26SignalPeer extends PacketExecutor<Packet26SignalPeer> {
                 /**
                  * Сессии звонка нет
                  */
-                client.disconnect(Failures.DATA_MISSMATCH);
+                client.disconnect(Failures.NO_CALL_SESSION);
                 return;
             }
             /**
@@ -170,12 +184,22 @@ public class Executor26SignalPeer extends PacketExecutor<Packet26SignalPeer> {
             }
             if(session == null) {
                 /**
-                 * Сессии звонка нет
+                 * Сессии звонка нет, скорее всего она была удалена при обрыве RTC Peer Connection,
+                 * при срабатывании RTCPeerConnection::close на клиенте раньше, чем клиент отправил сигнал END_CALL
                  */
-                client.disconnect(Failures.DATA_MISSMATCH);
                 return;
             }
+            /**
+             * Отправляем сигнал окончания звонка всем участникам сессии, кроме отправителя
+             */
             session.sendPacket(packet, client);
+            /**
+             * Отправляем пакет вызываемым (ringing) пользователям (которые еще не в сессии)
+             */
+            this.callManager.sendPacketToRinging(session, packet);
+            /**
+             * Удаляем сессию из активных сессий звонков
+             */
             this.callManager.removeSession(session);
             return;
         }
@@ -188,7 +212,7 @@ public class Executor26SignalPeer extends PacketExecutor<Packet26SignalPeer> {
                 /**
                  * Сессии звонка нет
                  */
-                client.disconnect(Failures.DATA_MISSMATCH);
+                client.disconnect(Failures.NO_CALL_SESSION);
                 return;
             }
             session.sendPacket(packet, null);

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

@@ -27,9 +27,7 @@ public class Executor27WebRTC extends PacketExecutor<Packet27WebRTC> {
         if(session == null) {
             /**
              * Если сессия не найдена, то мы не будем обрабатывать сигналы для звонка
-             * и просто отключим клиента от сервера.
              */
-            client.disconnect(Failures.DATA_MISSMATCH);
             return;
         }
         Room room = session.getRoom();
@@ -37,7 +35,6 @@ public class Executor27WebRTC extends PacketExecutor<Packet27WebRTC> {
             /**
              * Звонок еще не активен, а значит комнаты еще нет. Нельзя обменяться WebRTC сигналами пока комнаты еще нет.
              */
-            client.disconnect(Failures.DATA_MISSMATCH);
             return;
         }
         /**

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

@@ -8,7 +8,9 @@ public enum AttachmentType {
     MESSAGES(1),
     FILE(2),
     AVATAR(3),
-    CALL(4);
+    CALL(4),
+    VOICE(5),
+    VIDEO_CIRCLE(6);
 
 
     private final int code;

src/main/java/im/rosetta/service/dispatch/push/dispatchers/FCM.java

@@ -61,9 +61,10 @@ public class FCM extends Pusher {
                  * Тихий тип уведомления для очистки отправленных уведомлений на устройстве, 
                  * не должен отображаться пользователю, поэтому не задаем звук и ставим contentAvailable для iOS и high priority для Android
                  */
-                apnsConfig.setAps(Aps.builder().setContentAvailable(true).setSound("default").build());
+                apnsConfig.setAps(Aps.builder().setContentAvailable(true).setMutableContent(true).build());
                 androidConfig.setPriority(AndroidConfig.Priority.HIGH);
                 messageBuilder.setApnsConfig(apnsConfig.build());
+                messageBuilder.setNotification(Notification.builder().setTitle("").setBody("").build());
                 messageBuilder.setAndroidConfig(androidConfig.build());
                 break;
             case PushType.PERSONAL_MESSAGE:

src/main/java/im/rosetta/service/dispatch/push/dispatchers/VoIPApns.java

@@ -59,11 +59,13 @@ public class VoIPApns extends Pusher {
                   "aps": { "content-available": 1 },
                   "type": "CALL",
                   "callId": "%s",
-                  "from": "%s"
+                  "from": "%s",
+                  "joinToken": "%s"
                 }
                 """.formatted(
                     escape(data.getOrDefault("callId", "")),
-                    escape(data.getOrDefault("dialog", ""))
+                    escape(data.getOrDefault("dialog", "")),
+                    escape(data.getOrDefault("joinToken", ""))
                 );
 
             SimpleApnsPushNotification push = new SimpleApnsPushNotification(

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

@@ -7,6 +7,8 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import im.rosetta.calls.CallManager;
+import im.rosetta.calls.CallSession;
 import im.rosetta.client.ClientManager;
 import im.rosetta.logger.Logger;
 import im.rosetta.logger.enums.Color;
@@ -33,10 +35,12 @@ public class ForwardUnitService {
     private Set<SFU> sfuConnections = ConcurrentHashMap.newKeySet();
     private ClientManager clientManager;
     private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+    private CallManager callManager;
 
-    public ForwardUnitService(Logger logger, ClientManager clientManager) {
+    public ForwardUnitService(Logger logger, ClientManager clientManager, CallManager callManager) {
         this.logger = logger;
         this.clientManager = clientManager;
+        this.callManager = callManager;
         this.sfuConnectionsSheduler();
     }
 
@@ -135,25 +139,33 @@ public class ForwardUnitService {
     
     public void onPeerDisconnected(DisconnectedPeer disconnectedPeer) throws ProtocolException {
         Room room = disconnectedPeer.getRoom();
-        if(disconnectedPeer.getReason() != DisconnectReason.FAILED){
+        CallSession callSession = this.callManager.getCallSession(room);
+        callSession.leaveCall(disconnectedPeer.getPeerId());
+
+        if(disconnectedPeer.getReason() == DisconnectReason.FAILED){
             /**
-             * Если у нас произошло штатное отключение, а не в результате обрыва связи - то не нужно отправлять
-             * оппонентам пакеты о том, что участник отключился в результате обрыва связи.
+             * Произошло нештатное отключение клиента от сервера SFU, например, из-за сбоя сети
              */
-            return;
-        }
-        for(String peerId : room.getParticipants()) {
-            /**
-             * Уведомляем все пиры, что соединение с пиром было потеряно
-             */
-            if(room.getParticipants().size() == 1) {
-                /**
-                 * Звонок был завершен, так как в комнате остался только один участник, который не может продолжать звонок в одиночку.
-                 */
+            if(callSession.shouldRemove()){
                 Packet26SignalPeer packet = new Packet26SignalPeer();
                 packet.setSignalType(NetworkSignalType.END_CALL_BECAUSE_PEER_DISCONNECTED);
-                this.clientManager.sendPacketToAuthorizedPK(peerId, packet);
+                callSession.sendPacket(packet, null);
+                this.callManager.removeSession(callSession);
             }
+            return;
+        }
+        if(disconnectedPeer.getReason() == DisconnectReason.CLOSED){
+            /**
+             * Клиент намеренно покинул звонок, например, отключился от SFU сервера, так как завершил звонок или вышел из комнаты
+             * (например если клиент отрабатывает выход из звонка по кнопке END не правильно)
+             */
+            if(callSession.shouldRemove()){
+                Packet26SignalPeer packet = new Packet26SignalPeer();
+                packet.setSignalType(NetworkSignalType.END_CALL);
+                callSession.sendPacket(packet, null);
+                this.callManager.removeSession(callSession);
+            }
+            return;
         }
     }
 

src/main/java/io/g365sfu/Room.java

@@ -157,4 +157,12 @@ public class Room {
         buffer.flip();
         this.sfu.getConnection().send(buffer);
     }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) return true;
+        if (obj == null || getClass() != obj.getClass()) return false;
+        Room room = (Room) obj;
+        return roomId.equals(room.roomId);
+    }
 }