mirror of
https://github.com/Swiftgram/Telegram-iOS.git
synced 2025-06-16 05:55:20 +00:00
982 lines
32 KiB
Swift
982 lines
32 KiB
Swift
import Foundation
|
|
import SwiftSignalKit
|
|
import CoreMedia
|
|
|
|
private struct PayloadDescription: Codable {
|
|
var id: UInt32
|
|
var timestamp: Int32
|
|
}
|
|
|
|
private struct JoinPayload: Codable {
|
|
var id: UInt32
|
|
var string: String
|
|
}
|
|
|
|
private struct JoinResponsePayload: Codable {
|
|
var id: UInt32
|
|
var string: String
|
|
}
|
|
|
|
private struct KeepaliveInfo: Codable {
|
|
var id: UInt32
|
|
var timestamp: Int32
|
|
}
|
|
|
|
private let checkInterval: Double = 0.2
|
|
private let keepaliveTimeout: Double = 2.0
|
|
|
|
private func payloadDescriptionPath(basePath: String) -> String {
|
|
return basePath + "/currentPayloadDescription.json"
|
|
}
|
|
|
|
private func joinPayloadPath(basePath: String) -> String {
|
|
return basePath + "/joinPayload.json"
|
|
}
|
|
|
|
private func joinResponsePayloadPath(basePath: String) -> String {
|
|
return basePath + "/joinResponsePayload.json"
|
|
}
|
|
|
|
private func keepaliveInfoPath(basePath: String) -> String {
|
|
return basePath + "/keepaliveInfo.json"
|
|
}
|
|
|
|
private func broadcastAppSocketPath(basePath: String) -> String {
|
|
return basePath + "/0"
|
|
}
|
|
|
|
public final class IpcGroupCallAppContext {
|
|
private let basePath: String
|
|
private let currentId: UInt32
|
|
|
|
private let joinPayloadPromise = Promise<String>()
|
|
public var joinPayload: Signal<String, NoError> {
|
|
return self.joinPayloadPromise.get()
|
|
}
|
|
private var joinPayloadCheckTimer: SwiftSignalKit.Timer?
|
|
|
|
private let isActivePromise = ValuePromise<Bool>(false, ignoreRepeated: true)
|
|
public var isActive: Signal<Bool, NoError> {
|
|
return self.isActivePromise.get()
|
|
}
|
|
private var keepaliveCheckTimer: SwiftSignalKit.Timer?
|
|
|
|
public init(basePath: String) {
|
|
self.basePath = basePath
|
|
self.currentId = UInt32.random(in: 0 ..< UInt32.max)
|
|
|
|
let _ = try? FileManager.default.createDirectory(atPath: basePath, withIntermediateDirectories: true, attributes: nil)
|
|
|
|
self.sendRequest()
|
|
}
|
|
|
|
deinit {
|
|
self.joinPayloadCheckTimer?.invalidate()
|
|
self.keepaliveCheckTimer?.invalidate()
|
|
}
|
|
|
|
private func sendRequest() {
|
|
let timestamp = Int32(Date().timeIntervalSince1970)
|
|
let payloadDescription = PayloadDescription(
|
|
id: self.currentId,
|
|
timestamp: timestamp
|
|
)
|
|
guard let payloadDescriptionData = try? JSONEncoder().encode(payloadDescription) else {
|
|
preconditionFailure()
|
|
}
|
|
guard let _ = try? payloadDescriptionData.write(to: URL(fileURLWithPath: payloadDescriptionPath(basePath: self.basePath)), options: .atomic) else {
|
|
preconditionFailure()
|
|
}
|
|
|
|
self.receiveJoinPayload()
|
|
}
|
|
|
|
private func receiveJoinPayload() {
|
|
let joinPayloadCheckTimer = SwiftSignalKit.Timer(timeout: checkInterval, repeat: true, completion: { [weak self] in
|
|
self?.checkJoinPayload()
|
|
}, queue: .mainQueue())
|
|
self.joinPayloadCheckTimer = joinPayloadCheckTimer
|
|
joinPayloadCheckTimer.start()
|
|
}
|
|
|
|
private func checkJoinPayload() {
|
|
let filePath = joinPayloadPath(basePath: self.basePath)
|
|
guard let joinPayloadData = try? Data(contentsOf: URL(fileURLWithPath: filePath)) else {
|
|
return
|
|
}
|
|
|
|
self.joinPayloadCheckTimer?.invalidate()
|
|
let _ = try? FileManager.default.removeItem(atPath: filePath)
|
|
|
|
guard let joinPayload = try? JSONDecoder().decode(JoinPayload.self, from: joinPayloadData) else {
|
|
return
|
|
}
|
|
|
|
if joinPayload.id != self.currentId {
|
|
return
|
|
}
|
|
|
|
self.joinPayloadPromise.set(.single(joinPayload.string))
|
|
}
|
|
|
|
public func setJoinResponsePayload(_ joinResponsePayload: String) {
|
|
let inputJoinResponsePayload = JoinResponsePayload(
|
|
id: self.currentId,
|
|
string: joinResponsePayload
|
|
)
|
|
guard let inputJoinResponsePayloadData = try? JSONEncoder().encode(inputJoinResponsePayload) else {
|
|
preconditionFailure()
|
|
}
|
|
guard let _ = try? inputJoinResponsePayloadData.write(to: URL(fileURLWithPath: joinResponsePayloadPath(basePath: self.basePath)), options: .atomic) else {
|
|
preconditionFailure()
|
|
}
|
|
|
|
self.beginCheckingKeepaliveInfo()
|
|
}
|
|
|
|
private func beginCheckingKeepaliveInfo() {
|
|
let filePath = keepaliveInfoPath(basePath: self.basePath)
|
|
guard let keepaliveInfoData = try? Data(contentsOf: URL(fileURLWithPath: filePath)) else {
|
|
return
|
|
}
|
|
guard let keepaliveInfo = try? JSONDecoder().decode(KeepaliveInfo.self, from: keepaliveInfoData) else {
|
|
return
|
|
}
|
|
if keepaliveInfo.id != self.currentId {
|
|
self.isActivePromise.set(false)
|
|
return
|
|
}
|
|
let timestamp = Int32(Date().timeIntervalSince1970)
|
|
if keepaliveInfo.timestamp < timestamp - Int32(keepaliveTimeout) {
|
|
self.isActivePromise.set(false)
|
|
return
|
|
}
|
|
|
|
self.isActivePromise.set(true)
|
|
}
|
|
}
|
|
|
|
public final class IpcGroupCallBroadcastContext {
|
|
public enum Request {
|
|
case request
|
|
case failed
|
|
}
|
|
|
|
private let basePath: String
|
|
|
|
private var currentId: UInt32?
|
|
|
|
private var requestCheckTimer: SwiftSignalKit.Timer?
|
|
private let requestPromise = Promise<Request>()
|
|
public var request: Signal<Request, NoError> {
|
|
return self.requestPromise.get()
|
|
}
|
|
|
|
private var joinResponsePayloadCheckTimer: SwiftSignalKit.Timer?
|
|
private let joinResponsePayloadPromise = Promise<String>()
|
|
public var joinResponsePayload: Signal<String, NoError> {
|
|
return self.joinResponsePayloadPromise.get()
|
|
}
|
|
|
|
private var keepaliveTimer: SwiftSignalKit.Timer?
|
|
|
|
public init(basePath: String) {
|
|
self.basePath = basePath
|
|
|
|
let _ = try? FileManager.default.createDirectory(atPath: basePath, withIntermediateDirectories: true, attributes: nil)
|
|
|
|
self.receiveRequest()
|
|
}
|
|
|
|
deinit {
|
|
self.requestCheckTimer?.invalidate()
|
|
self.joinResponsePayloadCheckTimer?.invalidate()
|
|
self.keepaliveTimer?.invalidate()
|
|
self.endActiveIndication()
|
|
}
|
|
|
|
private func receiveRequest() {
|
|
let requestCheckTimer = SwiftSignalKit.Timer(timeout: checkInterval, repeat: true, completion: { [weak self] in
|
|
self?.checkRequest()
|
|
}, queue: .mainQueue())
|
|
self.requestCheckTimer = requestCheckTimer
|
|
requestCheckTimer.start()
|
|
}
|
|
|
|
private func checkRequest() {
|
|
let filePath = payloadDescriptionPath(basePath: self.basePath)
|
|
guard let payloadDescriptionData = try? Data(contentsOf: URL(fileURLWithPath: filePath)) else {
|
|
return
|
|
}
|
|
|
|
let _ = try? FileManager.default.removeItem(atPath: filePath)
|
|
|
|
guard let payloadDescription = try? JSONDecoder().decode(PayloadDescription.self, from: payloadDescriptionData) else {
|
|
self.requestCheckTimer?.invalidate()
|
|
self.requestPromise.set(.single(.failed))
|
|
return
|
|
}
|
|
let timestamp = Int32(Date().timeIntervalSince1970)
|
|
if payloadDescription.timestamp < timestamp - 1 * 60 {
|
|
self.requestPromise.set(.single(.failed))
|
|
return
|
|
}
|
|
|
|
self.requestCheckTimer?.invalidate()
|
|
|
|
self.currentId = payloadDescription.id
|
|
self.requestPromise.set(.single(.request))
|
|
}
|
|
|
|
public func setJoinPayload(_ joinPayload: String) {
|
|
guard let currentId = self.currentId else {
|
|
preconditionFailure()
|
|
}
|
|
let inputPayload = JoinPayload(
|
|
id: currentId,
|
|
string: joinPayload
|
|
)
|
|
guard let inputPayloadData = try? JSONEncoder().encode(inputPayload) else {
|
|
preconditionFailure()
|
|
}
|
|
guard let _ = try? inputPayloadData.write(to: URL(fileURLWithPath: joinPayloadPath(basePath: self.basePath)), options: .atomic) else {
|
|
preconditionFailure()
|
|
}
|
|
|
|
self.receiveJoinResponsePayload()
|
|
}
|
|
|
|
private func receiveJoinResponsePayload() {
|
|
let joinResponsePayloadCheckTimer = SwiftSignalKit.Timer(timeout: checkInterval, repeat: true, completion: { [weak self] in
|
|
self?.checkJoinResponsePayload()
|
|
}, queue: .mainQueue())
|
|
self.joinResponsePayloadCheckTimer = joinResponsePayloadCheckTimer
|
|
joinResponsePayloadCheckTimer.start()
|
|
}
|
|
|
|
private func checkJoinResponsePayload() {
|
|
let filePath = joinResponsePayloadPath(basePath: self.basePath)
|
|
guard let joinResponsePayloadData = try? Data(contentsOf: URL(fileURLWithPath: filePath)) else {
|
|
return
|
|
}
|
|
|
|
self.joinResponsePayloadCheckTimer?.invalidate()
|
|
let _ = try? FileManager.default.removeItem(atPath: filePath)
|
|
|
|
guard let joinResponsePayload = try? JSONDecoder().decode(JoinResponsePayload.self, from: joinResponsePayloadData) else {
|
|
return
|
|
}
|
|
if joinResponsePayload.id != self.currentId {
|
|
return
|
|
}
|
|
|
|
self.joinResponsePayloadPromise.set(.single(joinResponsePayload.string))
|
|
}
|
|
|
|
public func beginActiveIndication() {
|
|
if self.keepaliveTimer != nil {
|
|
return
|
|
}
|
|
|
|
self.writeKeepaliveInfo()
|
|
|
|
let keepaliveTimer = SwiftSignalKit.Timer(timeout: 1.0, repeat: true, completion: { [weak self] in
|
|
self?.writeKeepaliveInfo()
|
|
}, queue: .mainQueue())
|
|
self.keepaliveTimer = keepaliveTimer
|
|
keepaliveTimer.start()
|
|
}
|
|
|
|
private func writeKeepaliveInfo() {
|
|
guard let currentId = self.currentId else {
|
|
preconditionFailure()
|
|
}
|
|
let keepaliveInfo = KeepaliveInfo(
|
|
id: currentId,
|
|
timestamp: Int32(Date().timeIntervalSince1970)
|
|
)
|
|
guard let keepaliveInfoData = try? JSONEncoder().encode(keepaliveInfo) else {
|
|
preconditionFailure()
|
|
}
|
|
guard let _ = try? keepaliveInfoData.write(to: URL(fileURLWithPath: keepaliveInfoPath(basePath: self.basePath)), options: .atomic) else {
|
|
preconditionFailure()
|
|
}
|
|
}
|
|
|
|
private func endActiveIndication() {
|
|
let _ = try? FileManager.default.removeItem(atPath: keepaliveInfoPath(basePath: self.basePath))
|
|
}
|
|
}
|
|
|
|
private final class FdReadConnection {
|
|
private final class PendingData {
|
|
var data: Data
|
|
var offset: Int = 0
|
|
|
|
init(count: Int) {
|
|
self.data = Data(bytesNoCopy: malloc(count)!, count: count, deallocator: .free)
|
|
}
|
|
}
|
|
|
|
private let queue: Queue
|
|
let fd: Int32
|
|
private let didRead: ((Data) -> Void)?
|
|
private let channel: DispatchSourceRead
|
|
|
|
private var currendData: PendingData?
|
|
|
|
init(queue: Queue, fd: Int32, didRead: ((Data) -> Void)?) {
|
|
assert(queue.isCurrent())
|
|
self.queue = queue
|
|
self.fd = fd
|
|
self.didRead = didRead
|
|
|
|
self.channel = DispatchSource.makeReadSource(fileDescriptor: fd, queue: queue.queue)
|
|
self.channel.setEventHandler(handler: { [weak self] in
|
|
guard let strongSelf = self else {
|
|
return
|
|
}
|
|
|
|
while true {
|
|
if let currendData = strongSelf.currendData {
|
|
let offset = currendData.offset
|
|
let count = currendData.data.count - offset
|
|
let bytesRead = currendData.data.withUnsafeMutableBytes { bytes -> Int in
|
|
return Darwin.read(fd, bytes.baseAddress!.advanced(by: offset), min(8129, count))
|
|
}
|
|
if bytesRead <= 0 {
|
|
break
|
|
} else {
|
|
currendData.offset += bytesRead
|
|
if currendData.offset == currendData.data.count {
|
|
strongSelf.currendData = nil
|
|
strongSelf.didRead?(currendData.data)
|
|
}
|
|
}
|
|
} else {
|
|
var length: Int32 = 0
|
|
let bytesRead = read(fd, &length, 4)
|
|
if bytesRead < 0 {
|
|
break
|
|
} else {
|
|
assert(bytesRead == 4)
|
|
assert(length > 0 && length <= 30 * 1024 * 1024)
|
|
strongSelf.currendData = PendingData(count: Int(length))
|
|
}
|
|
}
|
|
}
|
|
})
|
|
self.channel.resume()
|
|
}
|
|
|
|
deinit {
|
|
assert(self.queue.isCurrent())
|
|
self.channel.cancel()
|
|
}
|
|
}
|
|
|
|
private final class FdWriteConnection {
|
|
private final class PendingData {
|
|
let data: Data
|
|
var didWriteHeader: Bool = false
|
|
var offset: Int = 0
|
|
|
|
init(data: Data) {
|
|
self.data = data
|
|
}
|
|
}
|
|
|
|
private let queue: Queue
|
|
let fd: Int32
|
|
private let channel: DispatchSourceWrite
|
|
private var isResumed = false
|
|
|
|
private let bufferSize: Int
|
|
private let buffer: UnsafeMutableRawPointer
|
|
|
|
private var currentData: PendingData?
|
|
private var nextData: Data?
|
|
|
|
init(queue: Queue, fd: Int32) {
|
|
assert(queue.isCurrent())
|
|
self.queue = queue
|
|
self.fd = fd
|
|
|
|
self.bufferSize = 8192
|
|
self.buffer = malloc(self.bufferSize)
|
|
|
|
self.channel = DispatchSource.makeWriteSource(fileDescriptor: fd, queue: queue.queue)
|
|
self.channel.setEventHandler(handler: { [weak self] in
|
|
guard let strongSelf = self else {
|
|
return
|
|
}
|
|
|
|
while true {
|
|
if let currentData = strongSelf.currentData {
|
|
if !currentData.didWriteHeader {
|
|
var length: Int32 = Int32(currentData.data.count)
|
|
let writtenBytes = Darwin.write(fd, &length, 4)
|
|
if writtenBytes > 0 {
|
|
assert(writtenBytes == 4)
|
|
currentData.didWriteHeader = true
|
|
} else {
|
|
strongSelf.channel.suspend()
|
|
strongSelf.isResumed = false
|
|
break
|
|
}
|
|
} else {
|
|
let offset = currentData.offset
|
|
let count = currentData.data.count - offset
|
|
let writtenBytes = currentData.data.withUnsafeBytes { bytes -> Int in
|
|
return Darwin.write(fd, bytes.baseAddress!.advanced(by: offset), min(count, strongSelf.bufferSize))
|
|
}
|
|
if writtenBytes > 0 {
|
|
currentData.offset += writtenBytes
|
|
if currentData.offset == currentData.data.count {
|
|
strongSelf.currentData = nil
|
|
|
|
if let nextData = strongSelf.nextData {
|
|
strongSelf.nextData = nil
|
|
strongSelf.currentData = PendingData(data: nextData)
|
|
} else {
|
|
strongSelf.channel.suspend()
|
|
strongSelf.isResumed = false
|
|
break
|
|
}
|
|
}
|
|
} else {
|
|
strongSelf.channel.suspend()
|
|
strongSelf.isResumed = false
|
|
break
|
|
}
|
|
}
|
|
} else {
|
|
strongSelf.channel.suspend()
|
|
strongSelf.isResumed = false
|
|
break
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
deinit {
|
|
assert(self.queue.isCurrent())
|
|
|
|
if !self.isResumed {
|
|
self.channel.resume()
|
|
}
|
|
self.channel.cancel()
|
|
|
|
free(self.buffer)
|
|
}
|
|
|
|
func replaceData(data: Data) {
|
|
if self.currentData == nil {
|
|
self.currentData = PendingData(data: data)
|
|
} else {
|
|
self.nextData = data
|
|
}
|
|
|
|
if !self.isResumed {
|
|
self.isResumed = true
|
|
self.channel.resume()
|
|
}
|
|
}
|
|
}
|
|
|
|
private final class NamedPipeReaderImpl {
|
|
private let queue: Queue
|
|
private var connection: FdReadConnection?
|
|
|
|
init(queue: Queue, path: String, didRead: @escaping (Data) -> Void) {
|
|
self.queue = queue
|
|
|
|
unlink(path)
|
|
mkfifo(path, 0o666)
|
|
let fd = open(path, O_RDONLY | O_NONBLOCK, S_IRUSR | S_IWUSR)
|
|
if fd != -1 {
|
|
self.connection = FdReadConnection(queue: self.queue, fd: fd, didRead: { data in
|
|
didRead(data)
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
private final class NamedPipeReader {
|
|
private let queue = Queue()
|
|
let impl: QueueLocalObject<NamedPipeReaderImpl>
|
|
|
|
init(path: String, didRead: @escaping (Data) -> Void) {
|
|
let queue = self.queue
|
|
self.impl = QueueLocalObject(queue: queue, generate: {
|
|
return NamedPipeReaderImpl(queue: queue, path: path, didRead: didRead)
|
|
})
|
|
}
|
|
}
|
|
|
|
private final class NamedPipeWriterImpl {
|
|
private let queue: Queue
|
|
private var connection: FdWriteConnection?
|
|
|
|
init(queue: Queue, path: String) {
|
|
self.queue = queue
|
|
|
|
let fd = open(path, O_WRONLY | O_NONBLOCK, S_IRUSR | S_IWUSR)
|
|
if fd != -1 {
|
|
self.connection = FdWriteConnection(queue: self.queue, fd: fd)
|
|
}
|
|
}
|
|
|
|
func replaceData(data: Data) {
|
|
guard let connection = self.connection else {
|
|
return
|
|
}
|
|
connection.replaceData(data: data)
|
|
}
|
|
}
|
|
|
|
private final class NamedPipeWriter {
|
|
private let queue = Queue()
|
|
private let impl: QueueLocalObject<NamedPipeWriterImpl>
|
|
|
|
init(path: String) {
|
|
let queue = self.queue
|
|
self.impl = QueueLocalObject(queue: queue, generate: {
|
|
return NamedPipeWriterImpl(queue: queue, path: path)
|
|
})
|
|
}
|
|
|
|
func replaceData(data: Data) {
|
|
self.impl.with { impl in
|
|
impl.replaceData(data: data)
|
|
}
|
|
}
|
|
}
|
|
|
|
private final class MappedFile {
|
|
let path: String
|
|
private var handle: Int32
|
|
private var currentSize: Int
|
|
private(set) var memory: UnsafeMutableRawPointer
|
|
|
|
init?(path: String, createIfNotExists: Bool) {
|
|
self.path = path
|
|
|
|
var flags: Int32 = O_RDWR | O_APPEND
|
|
if createIfNotExists {
|
|
flags |= O_CREAT
|
|
}
|
|
self.handle = open(path, flags, S_IRUSR | S_IWUSR)
|
|
|
|
if self.handle < 0 {
|
|
return nil
|
|
}
|
|
|
|
var value = stat()
|
|
stat(path, &value)
|
|
self.currentSize = Int(value.st_size)
|
|
|
|
self.memory = mmap(nil, self.currentSize, PROT_READ | PROT_WRITE, MAP_SHARED, self.handle, 0)
|
|
}
|
|
|
|
deinit {
|
|
munmap(self.memory, self.currentSize)
|
|
close(self.handle)
|
|
}
|
|
|
|
var size: Int {
|
|
get {
|
|
return self.currentSize
|
|
} set(value) {
|
|
if value != self.currentSize {
|
|
munmap(self.memory, self.currentSize)
|
|
ftruncate(self.handle, off_t(value))
|
|
self.currentSize = value
|
|
self.memory = mmap(nil, self.currentSize, PROT_READ | PROT_WRITE, MAP_SHARED, self.handle, 0)
|
|
}
|
|
}
|
|
}
|
|
|
|
func synchronize() {
|
|
msync(self.memory, self.currentSize, MS_ASYNC)
|
|
}
|
|
|
|
func write(at range: Range<Int>, from data: UnsafeRawPointer) {
|
|
memcpy(self.memory.advanced(by: range.lowerBound), data, range.count)
|
|
}
|
|
|
|
func read(at range: Range<Int>, to data: UnsafeMutableRawPointer) {
|
|
memcpy(data, self.memory.advanced(by: range.lowerBound), range.count)
|
|
}
|
|
|
|
func clear() {
|
|
memset(self.memory, 0, self.currentSize)
|
|
}
|
|
}
|
|
|
|
public final class IpcGroupCallBufferAppContext {
|
|
private let basePath: String
|
|
private let server: NamedPipeReader
|
|
|
|
private let id: UInt32
|
|
|
|
private let isActivePromise = ValuePromise<Bool>(false, ignoreRepeated: true)
|
|
public var isActive: Signal<Bool, NoError> {
|
|
return self.isActivePromise.get()
|
|
}
|
|
private var isActiveCheckTimer: SwiftSignalKit.Timer?
|
|
|
|
private let framesPipe = ValuePipe<CVPixelBuffer>()
|
|
public var frames: Signal<CVPixelBuffer, NoError> {
|
|
return self.framesPipe.signal()
|
|
}
|
|
|
|
private var framePollTimer: SwiftSignalKit.Timer?
|
|
private var mappedFile: MappedFile?
|
|
|
|
private var callActiveInfoTimer: SwiftSignalKit.Timer?
|
|
|
|
public init(basePath: String) {
|
|
self.basePath = basePath
|
|
let _ = try? FileManager.default.createDirectory(atPath: basePath, withIntermediateDirectories: true, attributes: nil)
|
|
|
|
self.id = UInt32.random(in: 0 ..< UInt32.max)
|
|
|
|
let framesPipe = self.framesPipe
|
|
self.server = NamedPipeReader(path: broadcastAppSocketPath(basePath: basePath), didRead: { data in
|
|
//framesPipe.putNext(data)
|
|
})
|
|
|
|
let dataPath = broadcastAppSocketPath(basePath: basePath) + "-data-\(self.id)"
|
|
|
|
if let mappedFile = MappedFile(path: dataPath, createIfNotExists: true) {
|
|
self.mappedFile = mappedFile
|
|
if mappedFile.size < 10 * 1024 * 1024 {
|
|
mappedFile.size = 10 * 1024 * 1024
|
|
}
|
|
}
|
|
|
|
let framePollTimer = SwiftSignalKit.Timer(timeout: 1.0 / 30.0, repeat: true, completion: { [weak self] in
|
|
guard let strongSelf = self, let mappedFile = strongSelf.mappedFile else {
|
|
return
|
|
}
|
|
|
|
let data = Data(bytesNoCopy: mappedFile.memory, count: mappedFile.size, deallocator: .none)
|
|
if let frame = deserializePixelBuffer(data: data) {
|
|
strongSelf.framesPipe.putNext(frame)
|
|
}
|
|
}, queue: .mainQueue())
|
|
self.framePollTimer = framePollTimer
|
|
framePollTimer.start()
|
|
|
|
self.updateCallIsActive()
|
|
|
|
let callActiveInfoTimer = SwiftSignalKit.Timer(timeout: 1.0, repeat: true, completion: { [weak self] in
|
|
self?.updateCallIsActive()
|
|
}, queue: .mainQueue())
|
|
self.callActiveInfoTimer = callActiveInfoTimer
|
|
callActiveInfoTimer.start()
|
|
|
|
let isActiveCheckTimer = SwiftSignalKit.Timer(timeout: 1.0, repeat: true, completion: { [weak self] in
|
|
self?.updateKeepaliveInfo()
|
|
}, queue: .mainQueue())
|
|
self.isActiveCheckTimer = isActiveCheckTimer
|
|
isActiveCheckTimer.start()
|
|
}
|
|
|
|
deinit {
|
|
self.framePollTimer?.invalidate()
|
|
self.callActiveInfoTimer?.invalidate()
|
|
self.isActiveCheckTimer?.invalidate()
|
|
if let mappedFile = self.mappedFile {
|
|
self.mappedFile = nil
|
|
let _ = try? FileManager.default.removeItem(atPath: mappedFile.path)
|
|
}
|
|
}
|
|
|
|
private func updateCallIsActive() {
|
|
let timestamp = Int32(Date().timeIntervalSince1970)
|
|
let payloadDescription = PayloadDescription(
|
|
id: self.id,
|
|
timestamp: timestamp
|
|
)
|
|
guard let payloadDescriptionData = try? JSONEncoder().encode(payloadDescription) else {
|
|
return
|
|
}
|
|
guard let _ = try? payloadDescriptionData.write(to: URL(fileURLWithPath: payloadDescriptionPath(basePath: self.basePath)), options: .atomic) else {
|
|
return
|
|
}
|
|
}
|
|
|
|
private func updateKeepaliveInfo() {
|
|
let filePath = keepaliveInfoPath(basePath: self.basePath)
|
|
guard let keepaliveInfoData = try? Data(contentsOf: URL(fileURLWithPath: filePath)) else {
|
|
return
|
|
}
|
|
guard let keepaliveInfo = try? JSONDecoder().decode(KeepaliveInfo.self, from: keepaliveInfoData) else {
|
|
return
|
|
}
|
|
if keepaliveInfo.id != self.id {
|
|
self.isActivePromise.set(false)
|
|
return
|
|
}
|
|
let timestamp = Int32(Date().timeIntervalSince1970)
|
|
if keepaliveInfo.timestamp < timestamp - Int32(keepaliveTimeout) {
|
|
self.isActivePromise.set(false)
|
|
return
|
|
}
|
|
|
|
self.isActivePromise.set(true)
|
|
}
|
|
}
|
|
|
|
public final class IpcGroupCallBufferBroadcastContext {
|
|
public enum Status {
|
|
case finished
|
|
}
|
|
|
|
private let basePath: String
|
|
private let client: NamedPipeWriter
|
|
private var timer: SwiftSignalKit.Timer?
|
|
|
|
private let statusPromise = Promise<Status>()
|
|
public var status: Signal<Status, NoError> {
|
|
return self.statusPromise.get()
|
|
}
|
|
|
|
private var mappedFile: MappedFile?
|
|
private var currentId: UInt32?
|
|
|
|
private var callActiveInfoTimer: SwiftSignalKit.Timer?
|
|
|
|
private var keepaliveInfoTimer: SwiftSignalKit.Timer?
|
|
|
|
public init(basePath: String) {
|
|
self.basePath = basePath
|
|
let _ = try? FileManager.default.createDirectory(atPath: basePath, withIntermediateDirectories: true, attributes: nil)
|
|
|
|
self.client = NamedPipeWriter(path: broadcastAppSocketPath(basePath: basePath))
|
|
|
|
let callActiveInfoTimer = SwiftSignalKit.Timer(timeout: 1.0, repeat: true, completion: { [weak self] in
|
|
self?.updateCallIsActive()
|
|
}, queue: .mainQueue())
|
|
self.callActiveInfoTimer = callActiveInfoTimer
|
|
callActiveInfoTimer.start()
|
|
}
|
|
|
|
deinit {
|
|
self.endActiveIndication()
|
|
|
|
self.callActiveInfoTimer?.invalidate()
|
|
self.keepaliveInfoTimer?.invalidate()
|
|
}
|
|
|
|
private func updateCallIsActive() {
|
|
let filePath = payloadDescriptionPath(basePath: self.basePath)
|
|
guard let payloadDescriptionData = try? Data(contentsOf: URL(fileURLWithPath: filePath)) else {
|
|
self.statusPromise.set(.single(.finished))
|
|
return
|
|
}
|
|
|
|
guard let payloadDescription = try? JSONDecoder().decode(PayloadDescription.self, from: payloadDescriptionData) else {
|
|
self.statusPromise.set(.single(.finished))
|
|
return
|
|
}
|
|
let timestamp = Int32(Date().timeIntervalSince1970)
|
|
if payloadDescription.timestamp < timestamp - 4 {
|
|
self.statusPromise.set(.single(.finished))
|
|
return
|
|
}
|
|
|
|
if let currentId = self.currentId {
|
|
if currentId != payloadDescription.id {
|
|
self.statusPromise.set(.single(.finished))
|
|
}
|
|
} else {
|
|
self.currentId = payloadDescription.id
|
|
|
|
let dataPath = broadcastAppSocketPath(basePath: basePath) + "-data-\(payloadDescription.id)"
|
|
|
|
if let mappedFile = MappedFile(path: dataPath, createIfNotExists: false) {
|
|
self.mappedFile = mappedFile
|
|
if mappedFile.size < 10 * 1024 * 1024 {
|
|
mappedFile.size = 10 * 1024 * 1024
|
|
}
|
|
}
|
|
|
|
self.writeKeepaliveInfo()
|
|
|
|
let keepaliveInfoTimer = SwiftSignalKit.Timer(timeout: 1.0, repeat: true, completion: { [weak self] in
|
|
self?.writeKeepaliveInfo()
|
|
}, queue: .mainQueue())
|
|
self.keepaliveInfoTimer = keepaliveInfoTimer
|
|
keepaliveInfoTimer.start()
|
|
}
|
|
}
|
|
|
|
public func setCurrentFrame(data: Data) {
|
|
//let _ = try? data.write(to: URL(fileURLWithPath: dataPath), options: [])
|
|
|
|
if let mappedFile = self.mappedFile, mappedFile.size >= data.count {
|
|
let _ = data.withUnsafeBytes { bytes in
|
|
memcpy(mappedFile.memory, bytes.baseAddress!, data.count)
|
|
}
|
|
}
|
|
|
|
//self.client.replaceData(data: data)
|
|
}
|
|
|
|
private func writeKeepaliveInfo() {
|
|
guard let currentId = self.currentId else {
|
|
preconditionFailure()
|
|
}
|
|
let keepaliveInfo = KeepaliveInfo(
|
|
id: currentId,
|
|
timestamp: Int32(Date().timeIntervalSince1970)
|
|
)
|
|
guard let keepaliveInfoData = try? JSONEncoder().encode(keepaliveInfo) else {
|
|
preconditionFailure()
|
|
}
|
|
guard let _ = try? keepaliveInfoData.write(to: URL(fileURLWithPath: keepaliveInfoPath(basePath: self.basePath)), options: .atomic) else {
|
|
preconditionFailure()
|
|
}
|
|
}
|
|
|
|
private func endActiveIndication() {
|
|
let _ = try? FileManager.default.removeItem(atPath: keepaliveInfoPath(basePath: self.basePath))
|
|
}
|
|
}
|
|
|
|
public func serializePixelBuffer(buffer: CVPixelBuffer) -> Data? {
|
|
let pixelFormat = CVPixelBufferGetPixelFormatType(buffer)
|
|
switch pixelFormat {
|
|
case kCVPixelFormatType_420YpCbCr8BiPlanarFullRange, kCVPixelFormatType_420YpCbCr8BiPlanarVideoRange:
|
|
let status = CVPixelBufferLockBaseAddress(buffer, .readOnly)
|
|
if status != kCVReturnSuccess {
|
|
return nil
|
|
}
|
|
defer {
|
|
CVPixelBufferUnlockBaseAddress(buffer, .readOnly)
|
|
}
|
|
|
|
let width = CVPixelBufferGetWidth(buffer)
|
|
let height = CVPixelBufferGetHeight(buffer)
|
|
|
|
guard let yPlane = CVPixelBufferGetBaseAddressOfPlane(buffer, 0) else {
|
|
return nil
|
|
}
|
|
let yStride = CVPixelBufferGetBytesPerRowOfPlane(buffer, 0)
|
|
let yPlaneSize = yStride * height
|
|
|
|
guard let uvPlane = CVPixelBufferGetBaseAddressOfPlane(buffer, 1) else {
|
|
return nil
|
|
}
|
|
let uvStride = CVPixelBufferGetBytesPerRowOfPlane(buffer, 1)
|
|
let uvPlaneSize = uvStride * (height / 2)
|
|
|
|
let headerSize: Int = 4 + 4 + 4 + 4 + 4
|
|
|
|
let dataSize = headerSize + yPlaneSize + uvPlaneSize
|
|
let resultBytes = malloc(dataSize)!
|
|
|
|
var pixelFormatValue = pixelFormat
|
|
memcpy(resultBytes.advanced(by: 0), &pixelFormatValue, 4)
|
|
var widthValue = Int32(width)
|
|
memcpy(resultBytes.advanced(by: 4), &widthValue, 4)
|
|
var heightValue = Int32(height)
|
|
memcpy(resultBytes.advanced(by: 4 + 4), &heightValue, 4)
|
|
var yStrideValue = Int32(yStride)
|
|
memcpy(resultBytes.advanced(by: 4 + 4 + 4), &yStrideValue, 4)
|
|
var uvStrideValue = Int32(uvStride)
|
|
memcpy(resultBytes.advanced(by: 4 + 4 + 4 + 4), &uvStrideValue, 4)
|
|
|
|
memcpy(resultBytes.advanced(by: headerSize), yPlane, yPlaneSize)
|
|
memcpy(resultBytes.advanced(by: headerSize + yPlaneSize), uvPlane, uvPlaneSize)
|
|
|
|
return Data(bytesNoCopy: resultBytes, count: dataSize, deallocator: .free)
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
public func deserializePixelBuffer(data: Data) -> CVPixelBuffer? {
|
|
if data.count < 4 + 4 + 4 + 4 {
|
|
return nil
|
|
}
|
|
let count = data.count
|
|
return data.withUnsafeBytes { bytes -> CVPixelBuffer? in
|
|
let dataBytes = bytes.baseAddress!
|
|
|
|
var pixelFormat: UInt32 = 0
|
|
memcpy(&pixelFormat, dataBytes.advanced(by: 0), 4)
|
|
|
|
switch pixelFormat {
|
|
case kCVPixelFormatType_420YpCbCr8BiPlanarFullRange, kCVPixelFormatType_420YpCbCr8BiPlanarVideoRange:
|
|
break
|
|
default:
|
|
return nil
|
|
}
|
|
|
|
var width: Int32 = 0
|
|
memcpy(&width, dataBytes.advanced(by: 4), 4)
|
|
var height: Int32 = 0
|
|
memcpy(&height, dataBytes.advanced(by: 4 + 4), 4)
|
|
var yStride: Int32 = 0
|
|
memcpy(&yStride, dataBytes.advanced(by: 4 + 4 + 4), 4)
|
|
var uvStride: Int32 = 0
|
|
memcpy(&uvStride, dataBytes.advanced(by: 4 + 4 + 4 + 4), 4)
|
|
|
|
if width < 0 || width > 8192 {
|
|
return nil
|
|
}
|
|
if height < 0 || height > 8192 {
|
|
return nil
|
|
}
|
|
|
|
let headerSize: Int = 4 + 4 + 4 + 4 + 4
|
|
|
|
let yPlaneSize = Int(yStride * height)
|
|
let uvPlaneSize = Int(uvStride * height / 2)
|
|
let dataSize = headerSize + yPlaneSize + uvPlaneSize
|
|
|
|
if dataSize > count {
|
|
return nil
|
|
}
|
|
|
|
var buffer: CVPixelBuffer? = nil
|
|
CVPixelBufferCreate(nil, Int(width), Int(height), pixelFormat, nil, &buffer)
|
|
if let buffer = buffer {
|
|
let status = CVPixelBufferLockBaseAddress(buffer, [])
|
|
if status != kCVReturnSuccess {
|
|
return nil
|
|
}
|
|
defer {
|
|
CVPixelBufferUnlockBaseAddress(buffer, [])
|
|
}
|
|
|
|
guard let destYPlane = CVPixelBufferGetBaseAddressOfPlane(buffer, 0) else {
|
|
return nil
|
|
}
|
|
let destYStride = CVPixelBufferGetBytesPerRowOfPlane(buffer, 0)
|
|
if destYStride != Int(yStride) {
|
|
return nil
|
|
}
|
|
|
|
guard let destUvPlane = CVPixelBufferGetBaseAddressOfPlane(buffer, 1) else {
|
|
return nil
|
|
}
|
|
let destUvStride = CVPixelBufferGetBytesPerRowOfPlane(buffer, 1)
|
|
if destUvStride != Int(uvStride) {
|
|
return nil
|
|
}
|
|
|
|
memcpy(destYPlane, dataBytes.advanced(by: headerSize), yPlaneSize)
|
|
memcpy(destUvPlane, dataBytes.advanced(by: headerSize + yPlaneSize), uvPlaneSize)
|
|
|
|
return buffer
|
|
} else {
|
|
return nil
|
|
}
|
|
}
|
|
}
|