Update calls

submodules/TelegramApi/Sources/Api0.swift

@@ -549,7 +549,7 @@ fileprivate let parsers: [Int32 : (BufferReader) -> Any?] = {
     dict[1471006352] = { return Api.PhoneCallDiscardReason.parse_phoneCallDiscardReasonHangup($0) }
     dict[-2048646399] = { return Api.PhoneCallDiscardReason.parse_phoneCallDiscardReasonMissed($0) }
     dict[-58224696] = { return Api.PhoneCallProtocol.parse_phoneCallProtocol($0) }
-    dict[-1655957568] = { return Api.PhoneConnection.parse_phoneConnection($0) }
+    dict[-1665063993] = { return Api.PhoneConnection.parse_phoneConnection($0) }
     dict[1667228533] = { return Api.PhoneConnection.parse_phoneConnectionWebrtc($0) }
     dict[-82216347] = { return Api.Photo.parse_photo($0) }
     dict[590459437] = { return Api.Photo.parse_photoEmpty($0) }

submodules/TelegramApi/Sources/Api14.swift

@@ -1130,15 +1130,16 @@ public extension Api {
 }
 public extension Api {
     enum PhoneConnection: TypeConstructorDescription {
-        case phoneConnection(id: Int64, ip: String, ipv6: String, port: Int32, peerTag: Buffer)
+        case phoneConnection(flags: Int32, id: Int64, ip: String, ipv6: String, port: Int32, peerTag: Buffer)
         case phoneConnectionWebrtc(flags: Int32, id: Int64, ip: String, ipv6: String, port: Int32, username: String, password: String)
     
     public func serialize(_ buffer: Buffer, _ boxed: Swift.Bool) {
     switch self {
-                case .phoneConnection(let id, let ip, let ipv6, let port, let peerTag):
+                case .phoneConnection(let flags, let id, let ip, let ipv6, let port, let peerTag):
                     if boxed {
-                        buffer.appendInt32(-1655957568)
+                        buffer.appendInt32(-1665063993)
                     }
+                    serializeInt32(flags, buffer: buffer, boxed: false)
                     serializeInt64(id, buffer: buffer, boxed: false)
                     serializeString(ip, buffer: buffer, boxed: false)
                     serializeString(ipv6, buffer: buffer, boxed: false)
@@ -1162,31 +1163,34 @@ public extension Api {
     
     public func descriptionFields() -> (String, [(String, Any)]) {
         switch self {
-                case .phoneConnection(let id, let ip, let ipv6, let port, let peerTag):
-                return ("phoneConnection", [("id", String(describing: id)), ("ip", String(describing: ip)), ("ipv6", String(describing: ipv6)), ("port", String(describing: port)), ("peerTag", String(describing: peerTag))])
+                case .phoneConnection(let flags, let id, let ip, let ipv6, let port, let peerTag):
+                return ("phoneConnection", [("flags", String(describing: flags)), ("id", String(describing: id)), ("ip", String(describing: ip)), ("ipv6", String(describing: ipv6)), ("port", String(describing: port)), ("peerTag", String(describing: peerTag))])
                 case .phoneConnectionWebrtc(let flags, let id, let ip, let ipv6, let port, let username, let password):
                 return ("phoneConnectionWebrtc", [("flags", String(describing: flags)), ("id", String(describing: id)), ("ip", String(describing: ip)), ("ipv6", String(describing: ipv6)), ("port", String(describing: port)), ("username", String(describing: username)), ("password", String(describing: password))])
     }
     }
     
         public static func parse_phoneConnection(_ reader: BufferReader) -> PhoneConnection? {
-            var _1: Int64?
-            _1 = reader.readInt64()
-            var _2: String?
-            _2 = parseString(reader)
+            var _1: Int32?
+            _1 = reader.readInt32()
+            var _2: Int64?
+            _2 = reader.readInt64()
             var _3: String?
             _3 = parseString(reader)
-            var _4: Int32?
-            _4 = reader.readInt32()
-            var _5: Buffer?
-            _5 = parseBytes(reader)
+            var _4: String?
+            _4 = parseString(reader)
+            var _5: Int32?
+            _5 = reader.readInt32()
+            var _6: Buffer?
+            _6 = parseBytes(reader)
             let _c1 = _1 != nil
             let _c2 = _2 != nil
             let _c3 = _3 != nil
             let _c4 = _4 != nil
             let _c5 = _5 != nil
-            if _c1 && _c2 && _c3 && _c4 && _c5 {
-                return Api.PhoneConnection.phoneConnection(id: _1!, ip: _2!, ipv6: _3!, port: _4!, peerTag: _5!)
+            let _c6 = _6 != nil
+            if _c1 && _c2 && _c3 && _c4 && _c5 && _c6 {
+                return Api.PhoneConnection.phoneConnection(flags: _1!, id: _2!, ip: _3!, ipv6: _4!, port: _5!, peerTag: _6!)
             }
             else {
                 return nil

submodules/TelegramCore/Sources/State/CallSessionManager.swift

@@ -217,6 +217,7 @@ public enum CallSessionConnection: Equatable {
         public let id: Int64
         public let ip: String
         public let ipv6: String
+        public let isTcp: Bool
         public let port: Int32
         public let peerTag: Data
         
@@ -224,12 +225,14 @@ public enum CallSessionConnection: Equatable {
             id: Int64,
             ip: String,
             ipv6: String,
+            isTcp: Bool,
             port: Int32,
             peerTag: Data
         ) {
             self.id = id
             self.ip = ip
             self.ipv6 = ipv6
+            self.isTcp = isTcp
             self.port = port
             self.peerTag = peerTag
         }
@@ -272,8 +275,9 @@ public enum CallSessionConnection: Equatable {
 
 private func parseConnection(_ apiConnection: Api.PhoneConnection) -> CallSessionConnection {
     switch apiConnection {
-    case let .phoneConnection(id, ip, ipv6, port, peerTag):
-        return .reflector(CallSessionConnection.Reflector(id: id, ip: ip, ipv6: ipv6, port: port, peerTag: peerTag.makeData()))
+    case let .phoneConnection(flags, id, ip, ipv6, port, peerTag):
+        let isTcp = (flags & (1 << 0)) != 0
+        return .reflector(CallSessionConnection.Reflector(id: id, ip: ip, ipv6: ipv6, isTcp: isTcp, port: port, peerTag: peerTag.makeData()))
     case let .phoneConnectionWebrtc(flags, id, ip, ipv6, port, username, password):
         return .webRtcReflector(CallSessionConnection.WebRtcReflector(
             id: id,

submodules/TelegramVoip/Sources/OngoingCallContext.swift

@@ -3,10 +3,67 @@ import UIKit
 import SwiftSignalKit
 import TelegramCore
 import TelegramUIPreferences
+import Network
 
 import TgVoip
 import TgVoipWebrtc
 
+private let debugUseLegacyVersionForReflectors: Bool = {
+    #if DEBUG && false
+    return true
+    #else
+    return false
+    #endif
+}()
+
+private struct PeerTag: Hashable, CustomStringConvertible {
+    var bytes: [UInt8] = Array<UInt8>(repeating: 0, count: 16)
+    
+    var canonical: PeerTag {
+        var updatedBytes = self.bytes
+        updatedBytes[0] &= ~1
+        return PeerTag(bytes: updatedBytes)
+    }
+    
+    var flipped: PeerTag {
+        var updatedBytes = self.bytes
+        updatedBytes[0] ^= 1
+        return PeerTag(bytes: updatedBytes)
+    }
+    
+    var description: String {
+        var result = ""
+        
+        for byte in bytes {
+            result.append(String(byte, radix: 16))
+        }
+        
+        return result
+    }
+}
+
+private extension PeerTag {
+    init(data: Data) {
+        precondition(data.count >= 16)
+        data.withUnsafeBytes { buffer -> Void in
+            memcpy(&self.bytes, buffer.baseAddress!.assumingMemoryBound(to: UInt8.self), 16)
+        }
+    }
+    
+    var data: Data {
+        var bytes = self.bytes
+        var resultData = Data(repeating: 0, count: 16)
+        resultData.withUnsafeMutableBytes { buffer -> Void in
+            memcpy(buffer.baseAddress!.assumingMemoryBound(to: UInt8.self), &bytes, 16)
+        }
+        return resultData
+    }
+}
+
+private func flippedPeerTag(_ data: Data) -> Data {
+    return PeerTag(data: data).flipped.data
+}
+
 private func callConnectionDescription(_ connection: CallSessionConnection) -> OngoingCallConnectionDescription? {
     switch connection {
     case let .reflector(reflector):
@@ -16,28 +73,27 @@ private func callConnectionDescription(_ connection: CallSessionConnection) -> O
     }
 }
 
-private func callConnectionDescriptionsWebrtc(_ connection: CallSessionConnection) -> [OngoingCallConnectionDescriptionWebrtc] {
+private func callConnectionDescriptionsWebrtc(_ connection: CallSessionConnection, idMapping: [Int64: UInt8]) -> [OngoingCallConnectionDescriptionWebrtc] {
     switch connection {
     case let .reflector(reflector):
-        #if DEBUG
+        guard let id = idMapping[reflector.id] else {
+            return []
+        }
         var result: [OngoingCallConnectionDescriptionWebrtc] = []
         if !reflector.ip.isEmpty {
-            result.append(OngoingCallConnectionDescriptionWebrtc(connectionId: reflector.id, hasStun: false, hasTurn: true, ip: reflector.ip, port: reflector.port, username: "reflector", password: hexString(reflector.peerTag)))
+            result.append(OngoingCallConnectionDescriptionWebrtc(reflectorId: id, hasStun: false, hasTurn: true, hasTcp: reflector.isTcp, ip: reflector.ip, port: reflector.port, username: "reflector", password: hexString(reflector.peerTag)))
         }
         if !reflector.ipv6.isEmpty {
-            result.append(OngoingCallConnectionDescriptionWebrtc(connectionId: reflector.id, hasStun: false, hasTurn: true, ip: reflector.ipv6, port: reflector.port, username: "reflector", password: hexString(reflector.peerTag)))
+            result.append(OngoingCallConnectionDescriptionWebrtc(reflectorId: id, hasStun: false, hasTurn: true, hasTcp: reflector.isTcp, ip: reflector.ipv6, port: reflector.port, username: "reflector", password: hexString(reflector.peerTag)))
         }
         return result
-        #else
-        return []
-        #endif
     case let .webRtcReflector(reflector):
         var result: [OngoingCallConnectionDescriptionWebrtc] = []
         if !reflector.ip.isEmpty {
-            result.append(OngoingCallConnectionDescriptionWebrtc(connectionId: reflector.id, hasStun: reflector.hasStun, hasTurn: reflector.hasTurn, ip: reflector.ip, port: reflector.port, username: reflector.username, password: reflector.password))
+            result.append(OngoingCallConnectionDescriptionWebrtc(reflectorId: 0, hasStun: reflector.hasStun, hasTurn: reflector.hasTurn, hasTcp: false, ip: reflector.ip, port: reflector.port, username: reflector.username, password: reflector.password))
         }
         if !reflector.ipv6.isEmpty {
-            result.append(OngoingCallConnectionDescriptionWebrtc(connectionId: reflector.id, hasStun: reflector.hasStun, hasTurn: reflector.hasTurn, ip: reflector.ipv6, port: reflector.port, username: reflector.username, password: reflector.password))
+            result.append(OngoingCallConnectionDescriptionWebrtc(reflectorId: 0, hasStun: reflector.hasStun, hasTurn: reflector.hasTurn, hasTcp: false, ip: reflector.ipv6, port: reflector.port, username: reflector.username, password: reflector.password))
         }
         return result
     }
@@ -645,6 +701,7 @@ public final class OngoingCallContext {
         }
     }
     
+    public let callId: CallId
     public let internalId: CallSessionInternalId
     
     private let queue = Queue()
@@ -683,25 +740,31 @@ public final class OngoingCallContext {
     private let tempLogFile: EngineTempBoxFile
     private let tempStatsLogFile: EngineTempBoxFile
     
+    private var signalingConnectionManager: QueueLocalObject<CallSignalingConnectionManager>?
+    
     public static func versions(includeExperimental: Bool, includeReference: Bool) -> [(version: String, supportsVideo: Bool)] {
-        var result: [(version: String, supportsVideo: Bool)] = [(OngoingCallThreadLocalContext.version(), false)]
-        if includeExperimental {
+        #if DEBUG
+        if "".isEmpty {
+            return [("5.0.0", true)]
+        }
+        #endif
+        
+        if debugUseLegacyVersionForReflectors {
+            return [(OngoingCallThreadLocalContext.version(), true)]
+        } else {
+            var result: [(version: String, supportsVideo: Bool)] = [(OngoingCallThreadLocalContext.version(), false)]
             result.append(contentsOf: OngoingCallThreadLocalContextWebrtc.versions(withIncludeReference: includeReference).map { version -> (version: String, supportsVideo: Bool) in
                 return (version, true)
             })
+            return result
         }
-        return result
     }
 
-    public init(account: Account, callSessionManager: CallSessionManager, internalId: CallSessionInternalId, proxyServer: ProxyServerSettings?, initialNetworkType: NetworkType, updatedNetworkType: Signal<NetworkType, NoError>, serializedData: String?, dataSaving: VoiceCallDataSaving, derivedState: VoipDerivedState, key: Data, isOutgoing: Bool, video: OngoingCallVideoCapturer?, connections: CallSessionConnectionSet, maxLayer: Int32, version: String, allowP2P: Bool, enableTCP: Bool, enableStunMarking: Bool, audioSessionActive: Signal<Bool, NoError>, logName: String, preferredVideoCodec: String?) {
+    public init(account: Account, callSessionManager: CallSessionManager, callId: CallId, internalId: CallSessionInternalId, proxyServer: ProxyServerSettings?, initialNetworkType: NetworkType, updatedNetworkType: Signal<NetworkType, NoError>, serializedData: String?, dataSaving: VoiceCallDataSaving, derivedState: VoipDerivedState, key: Data, isOutgoing: Bool, video: OngoingCallVideoCapturer?, connections: CallSessionConnectionSet, maxLayer: Int32, version: String, allowP2P: Bool, enableTCP: Bool, enableStunMarking: Bool, audioSessionActive: Signal<Bool, NoError>, logName: String, preferredVideoCodec: String?) {
         let _ = setupLogs
         OngoingCallThreadLocalContext.applyServerConfig(serializedData)
         
-        #if DEBUG
-        let version = "4.1.2"
-        let allowP2P = false
-        #endif
-        
+        self.callId = callId
         self.internalId = internalId
         self.account = account
         self.callSessionManager = callSessionManager
@@ -722,7 +785,18 @@ public final class OngoingCallContext {
         |> take(1)
         |> deliverOn(queue)).start(next: { [weak self] _ in
             if let strongSelf = self {
-                if OngoingCallThreadLocalContextWebrtc.versions(withIncludeReference: true).contains(version) {
+                var useModernImplementation = true
+                var version = version
+                var allowP2P = allowP2P
+                if debugUseLegacyVersionForReflectors {
+                    useModernImplementation = true
+                    version = "5.0.0"
+                    allowP2P = false
+                } else {
+                    useModernImplementation = version != OngoingCallThreadLocalContext.version()
+                }
+                
+                if useModernImplementation {
                     var voipProxyServer: VoipProxyServerWebrtc?
                     if let proxyServer = proxyServer {
                         switch proxyServer.connection {
@@ -734,30 +808,89 @@ public final class OngoingCallContext {
                     }
                     
                     let unfilteredConnections = [connections.primary] + connections.alternatives
+                    
+                    var reflectorIdList: [Int64] = []
+                    for connection in unfilteredConnections {
+                        switch connection {
+                        case let .reflector(reflector):
+                            reflectorIdList.append(reflector.id)
+                        case .webRtcReflector:
+                            break
+                        }
+                    }
+                    
+                    reflectorIdList.sort()
+                    
+                    var reflectorIdMapping: [Int64: UInt8] = [:]
+                    for i in 0 ..< reflectorIdList.count {
+                        reflectorIdMapping[reflectorIdList[i]] = UInt8(i + 1)
+                    }
+                    
+                    var signalingReflector: OngoingCallConnectionDescriptionWebrtc?
+                    
                     var processedConnections: [CallSessionConnection] = []
                     var filteredConnections: [OngoingCallConnectionDescriptionWebrtc] = []
-                    for connection in unfilteredConnections {
+                    connectionsLoop: for connection in unfilteredConnections {
                         if processedConnections.contains(connection) {
                             continue
                         }
                         processedConnections.append(connection)
-                        filteredConnections.append(contentsOf: callConnectionDescriptionsWebrtc(connection))
+                        
+                        switch connection {
+                        case let .reflector(reflector):
+                            if reflector.isTcp {
+                                if signalingReflector == nil {
+                                    signalingReflector = OngoingCallConnectionDescriptionWebrtc(reflectorId: 0, hasStun: false, hasTurn: true, hasTcp: true, ip: reflector.ip, port: reflector.port, username: "reflector", password: hexString(reflector.peerTag))
+                                }
+                                
+                                continue connectionsLoop
+                            }
+                        case .webRtcReflector:
+                            break
+                        }
+                        
+                        var webrtcConnections: [OngoingCallConnectionDescriptionWebrtc] = []
+                        for connection in callConnectionDescriptionsWebrtc(connection, idMapping: reflectorIdMapping) {
+                            webrtcConnections.append(connection)
+                        }
+                        
+                        filteredConnections.append(contentsOf: webrtcConnections)
                     }
                     
-                    /*#if DEBUG
-                    filteredConnections.removeAll()
-                    filteredConnections.append(OngoingCallConnectionDescriptionWebrtc(
-                        connectionId: 1,
-                        hasStun: true,
-                        hasTurn: true, ip: "178.62.7.192",
-                        port: 1400,
-                        username: "user",
-                        password: "user")
-                    )
-                    #endif*/
+                    if let signalingReflector = signalingReflector {
+                        if #available(iOS 12.0, *) {
+                            let peerTag = dataWithHexString(signalingReflector.password)
+                            
+                            strongSelf.signalingConnectionManager = QueueLocalObject(queue: queue, generate: {
+                                return CallSignalingConnectionManager(queue: queue, peerTag: peerTag, servers: [signalingReflector], dataReceived: { data in
+                                    guard let strongSelf = self else {
+                                        return
+                                    }
+                                    strongSelf.withContext { context in
+                                        if let context = context as? OngoingCallThreadLocalContextWebrtc {
+                                            context.addSignaling(data)
+                                        }
+                                    }
+                                })
+                            })
+                        }
+                    }
                     
                     let context = OngoingCallThreadLocalContextWebrtc(version: version, queue: OngoingCallThreadLocalContextQueueImpl(queue: queue), proxy: voipProxyServer, networkType: ongoingNetworkTypeForTypeWebrtc(initialNetworkType), dataSaving: ongoingDataSavingForTypeWebrtc(dataSaving), derivedState: derivedState.data, key: key, isOutgoing: isOutgoing, connections: filteredConnections, maxLayer: maxLayer, allowP2P: allowP2P, allowTCP: enableTCP, enableStunMarking: enableStunMarking, logPath: tempLogPath, statsLogPath: tempStatsLogPath, sendSignalingData: { [weak callSessionManager] data in
-                        callSessionManager?.sendSignalingData(internalId: internalId, data: data)
+                        queue.async {
+                            guard let strongSelf = self else {
+                                return
+                            }
+                            if let signalingConnectionManager = strongSelf.signalingConnectionManager {
+                                signalingConnectionManager.with { impl in
+                                    impl.send(payloadData: data)
+                                }
+                            }
+                            
+                            if let callSessionManager = callSessionManager {
+                                callSessionManager.sendSignalingData(internalId: internalId, data: data)
+                            }
+                        }
                     }, videoCapturer: video?.impl, preferredVideoCodec: preferredVideoCodec, audioInputDeviceId: "")
                     
                     strongSelf.contextRef = Unmanaged.passRetained(OngoingCallThreadLocalContextHolder(context))
@@ -867,6 +1000,10 @@ public final class OngoingCallContext {
                         }
                     }
                 })
+                
+                strongSelf.signalingConnectionManager?.with { impl in
+                    impl.start()
+                }
             }
         }))
     }
@@ -908,7 +1045,8 @@ public final class OngoingCallContext {
         }
     }
     
-    public func stop(callId: CallId? = nil, sendDebugLogs: Bool = false, debugLogValue: Promise<String?>) {
+    public func stop(sendDebugLogs: Bool = false, debugLogValue: Promise<String?>) {
+        let callId = self.callId
         let account = self.account
         let logPath = self.logPath
         var statsLogPath = ""
@@ -941,7 +1079,7 @@ public final class OngoingCallContext {
                     let _ = try? FileManager.default.moveItem(atPath: tempStatsLogPath, toPath: statsLogPath)
                 }
                 
-                if let callId = callId, !statsLogPath.isEmpty, let data = try? Data(contentsOf: URL(fileURLWithPath: statsLogPath)), let dataString = String(data: data, encoding: .utf8) {
+                if !statsLogPath.isEmpty, let data = try? Data(contentsOf: URL(fileURLWithPath: statsLogPath)), let dataString = String(data: data, encoding: .utf8) {
                     debugLogValue.set(.single(dataString))
                     let engine = TelegramEngine(account: self.account)
                     let _ = engine.calls.saveCallDebugLog(callId: callId, log: dataString).start(next: { result in
@@ -1061,3 +1199,257 @@ public final class OngoingCallContext {
         }
     }
 }
+
+private protocol CallSignalingConnection {
+    func start()
+    func stop()
+    func send(payloadData: Data)
+}
+
+@available(iOS 12.0, *)
+private final class CallSignalingConnectionImpl: CallSignalingConnection {
+    private let queue: Queue
+    private let host: NWEndpoint.Host
+    private let port: NWEndpoint.Port
+    private let peerTag: Data
+    private let dataReceived: (Data) -> Void
+    private let isClosed: () -> Void
+    private let connection: NWConnection
+    
+    private var isConnected: Bool = false
+    
+    private var pingTimer: SwiftSignalKit.Timer?
+    
+    private var queuedPayloads: [Data] = []
+    
+    init(queue: Queue, host: String, port: UInt16, peerTag: Data, dataReceived: @escaping (Data) -> Void, isClosed: @escaping () -> Void) {
+        self.queue = queue
+        self.host = NWEndpoint.Host(host)
+        self.port = NWEndpoint.Port(rawValue: port)!
+        self.peerTag = peerTag
+        self.dataReceived = dataReceived
+        self.isClosed = isClosed
+        self.connection = NWConnection(host: self.host, port: self.port, using: .tcp)
+        
+        self.connection.stateUpdateHandler = { [weak self] state in
+            queue.async {
+                self?.stateUpdated(state: state)
+            }
+        }
+    }
+    
+    private func stateUpdated(state: NWConnection.State) {
+        switch state {
+        case .ready:
+            Logger.shared.log("CallSignaling", "Connection state is ready")
+            
+            var headerData = Data(count: 4)
+            headerData.withUnsafeMutableBytes { bytes in
+                bytes.baseAddress!.assumingMemoryBound(to: UInt32.self).pointee = 0xeeeeeeee
+            }
+            self.connection.send(content: headerData, completion: .contentProcessed({ error in
+                if let error = error {
+                    Logger.shared.log("CallSignaling", "Connection send header error: \(error)")
+                }
+            }))
+            
+            self.beginPingTimer()
+            
+            self.sendPacket(payload: Data())
+        case let .failed(error):
+            Logger.shared.log("CallSignaling", "Connection error: \(error)")
+            self.onIsClosed()
+        default:
+            break
+        }
+    }
+    
+    func start() {
+        self.connection.start(queue: self.queue.queue)
+        self.receivePacketHeader()
+    }
+    
+    private func beginPingTimer() {
+        self.pingTimer = SwiftSignalKit.Timer(timeout: self.isConnected ? 2.0 : 0.15, repeat: false, completion: { [weak self] in
+            guard let strongSelf = self else {
+                return
+            }
+            strongSelf.sendPacket(payload: Data())
+            
+            strongSelf.beginPingTimer()
+        }, queue: self.queue)
+        self.pingTimer?.start()
+    }
+    
+    private func receivePacketHeader() {
+        self.connection.receive(minimumIncompleteLength: 4, maximumLength: 4, completion: { [weak self] data, _, _, error in
+            guard let strongSelf = self else {
+                return
+            }
+            if let data = data, data.count == 4 {
+                let payloadSize = data.withUnsafeBytes { bytes -> UInt32 in
+                    return bytes.baseAddress!.assumingMemoryBound(to: UInt32.self).pointee
+                }
+                if payloadSize < 2 * 1024 * 1024 {
+                    strongSelf.receivePacketPayload(size: Int(payloadSize))
+                } else {
+                    Logger.shared.log("CallSignaling", "Connection received invalid packet size: \(payloadSize)")
+                }
+            } else {
+                Logger.shared.log("CallSignaling", "Connection receive packet header error: \(String(describing: error))")
+                strongSelf.onIsClosed()
+            }
+        })
+    }
+    
+    private func receivePacketPayload(size: Int) {
+        self.connection.receive(minimumIncompleteLength: size, maximumLength: size, completion: { [weak self] data, _, _, error in
+            guard let strongSelf = self else {
+                return
+            }
+            if let data = data, data.count == size {
+                Logger.shared.log("CallSignaling", "Connection receive packet payload: \(data.count) bytes")
+                
+                if data.count < 16 + 4 {
+                    Logger.shared.log("CallSignaling", "Connection invalid payload size: \(data.count)")
+                    strongSelf.onIsClosed()
+                } else {
+                    let readPeerTag = data.subdata(in: 0 ..< 16)
+                    if readPeerTag != strongSelf.peerTag {
+                        Logger.shared.log("CallSignaling", "Peer tag mismatch: \(hexString(readPeerTag))")
+                        strongSelf.onIsClosed()
+                    } else {
+                        let actualPayloadSize = data.withUnsafeBytes { bytes -> UInt32 in
+                            var result: UInt32 = 0
+                            memcpy(&result, bytes.baseAddress!.assumingMemoryBound(to: UInt8.self).advanced(by: 16), 4)
+                            return result
+                        }
+                        
+                        if Int(actualPayloadSize) > data.count - 16 - 4 {
+                            Logger.shared.log("CallSignaling", "Connection invalid actual payload size: \(actualPayloadSize)")
+                            strongSelf.onIsClosed()
+                        } else {
+                            if !strongSelf.isConnected {
+                                strongSelf.isConnected = true
+                                
+                                for payload in strongSelf.queuedPayloads {
+                                    strongSelf.sendPacket(payload: payload)
+                                }
+                                strongSelf.queuedPayloads.removeAll()
+                            }
+                            
+                            if actualPayloadSize != 0 {
+                                strongSelf.dataReceived(data.subdata(in: (16 + 4) ..< (16 + 4 + Int(actualPayloadSize))))
+                            } else {
+                                //strongSelf.sendPacket(payload: Data())
+                            }
+                            strongSelf.receivePacketHeader()
+                        }
+                    }
+                }
+            } else {
+                Logger.shared.log("CallSignaling", "Connection receive packet payload error: \(String(describing: error))")
+                strongSelf.onIsClosed()
+            }
+        })
+    }
+    
+    func stop() {
+        self.connection.stateUpdateHandler = nil
+        self.connection.cancel()
+    }
+    
+    private func onIsClosed() {
+        self.connection.stateUpdateHandler = nil
+        self.connection.cancel()
+        
+        self.isClosed()
+    }
+    
+    func send(payloadData: Data) {
+        if self.isConnected {
+            self.sendPacket(payload: payloadData)
+        } else {
+            self.queuedPayloads.append(payloadData)
+        }
+    }
+    
+    private func sendPacket(payload: Data) {
+        var payloadSize = UInt32(payload.count)
+        let cleanSize = 16 + 4 + payloadSize
+        let paddingSize = ((cleanSize + 3) & ~(4 - 1)) - cleanSize
+        var totalSize = cleanSize + paddingSize
+        
+        var sendBuffer = Data(count: 4 + Int(totalSize))
+        sendBuffer.withUnsafeMutableBytes { bytes in
+            let baseAddress = bytes.baseAddress!.assumingMemoryBound(to: UInt8.self)
+            
+            memcpy(baseAddress, &totalSize, 4)
+            
+            self.peerTag.withUnsafeBytes { peerTagBytes -> Void in
+                memcpy(baseAddress.advanced(by: 4), peerTagBytes.baseAddress!.assumingMemoryBound(to: UInt8.self), 16)
+            }
+            
+            memcpy(baseAddress.advanced(by: 4 + 16), &payloadSize, 4)
+            
+            payload.withUnsafeBytes { payloadBytes -> Void in
+                memcpy(baseAddress.advanced(by: 4 + 16 + 4), payloadBytes.baseAddress!.assumingMemoryBound(to: UInt8.self), payloadBytes.count)
+            }
+        }
+        
+        Logger.shared.log("CallSignaling", "Send packet payload: \(totalSize) bytes")
+        
+        self.connection.send(content: sendBuffer, isComplete: true, completion: .contentProcessed({ error in
+            if let error = error {
+                Logger.shared.log("CallSignaling", "Connection send payload error: \(error)")
+            }
+        }))
+    }
+}
+
+private final class CallSignalingConnectionManager {
+    private let queue: Queue
+    
+    private var nextConnectionId: Int = 0
+    private var connections: [Int: CallSignalingConnection] = [:]
+    
+    init(queue: Queue, peerTag: Data, servers: [OngoingCallConnectionDescriptionWebrtc], dataReceived: @escaping (Data) -> Void) {
+        self.queue = queue
+        
+        for server in servers {
+            if server.hasTcp {
+                let id = self.nextConnectionId
+                self.nextConnectionId += 1
+                if #available(iOS 12.0, *) {
+                    let connection = CallSignalingConnectionImpl(queue: queue, host: server.ip, port: UInt16(server.port), peerTag: peerTag, dataReceived: { data in
+                        dataReceived(data)
+                    }, isClosed: { [weak self] in
+                        guard let strongSelf = self else {
+                            return
+                        }
+                        let _ = strongSelf
+                    })
+                    connections[id] = connection
+                }
+            }
+        }
+    }
+    
+    func start() {
+        for (_, connection) in self.connections {
+            connection.start()
+        }
+    }
+    
+    func stop() {
+        for (_, connection) in self.connections {
+            connection.stop()
+        }
+    }
+    
+    func send(payloadData: Data) {
+        for (_, connection) in self.connections {
+            connection.send(payloadData: payloadData)
+        }
+    }
+}

submodules/TgVoipWebrtc/tgcalls

@@ -1 +1 @@
-Subproject commit 3ce2c38805ea5af7a05b8fe93a26becaf9e76bd3
+Subproject commit c741da4568b0971ed06d9ccdc7a94db566bb84a0