From 5ae4425e4b7916d3d0e956e2ee6df800c856b098 Mon Sep 17 00:00:00 2001 From: shogo4405 Date: Tue, 9 Apr 2024 21:31:41 +0900 Subject: [PATCH] Improve handling of RTMP timestamps. --- Examples/iOSSwiftUI/Model/ViewModel.swift | 2 +- HaishinKit.xcodeproj/project.pbxproj | 16 +-- Sources/RTMP/RTMPMuxer.swift | 113 ++++++++++------------ Sources/RTMP/RTMPStream.swift | 83 ++++++---------- Sources/RTMP/RTMPTimestamp.swift | 82 ++++++++++++++++ Tests/RTMP/RTMPTimestampTests.swift | 11 +++ 6 files changed, 186 insertions(+), 121 deletions(-) create mode 100644 Sources/RTMP/RTMPTimestamp.swift create mode 100644 Tests/RTMP/RTMPTimestampTests.swift diff --git a/Examples/iOSSwiftUI/Model/ViewModel.swift b/Examples/iOSSwiftUI/Model/ViewModel.swift index fb33f3ffe..d7fc84f8a 100644 --- a/Examples/iOSSwiftUI/Model/ViewModel.swift +++ b/Examples/iOSSwiftUI/Model/ViewModel.swift @@ -170,7 +170,7 @@ final class ViewModel: ObservableObject { func rotateCamera() { let position: AVCaptureDevice.Position = currentPosition == .back ? .front : .back - rtmpStream.attachCamera(AVCaptureDevice.default(.builtInWideAngleCamera, for: .video, position: position)) { unit, error in + rtmpStream.attachCamera(AVCaptureDevice.default(.builtInWideAngleCamera, for: .video, position: position)) { _, error in logger.error(error) } currentPosition = position diff --git a/HaishinKit.xcodeproj/project.pbxproj b/HaishinKit.xcodeproj/project.pbxproj index a8e064fda..8d2ea42b4 100644 --- a/HaishinKit.xcodeproj/project.pbxproj +++ b/HaishinKit.xcodeproj/project.pbxproj @@ -221,6 +221,8 @@ BC959F1229717EDB0067BA97 /* PreferenceViewController.swift in Sources */ = {isa = PBXBuildFile; fileRef = BC959F1129717EDB0067BA97 /* PreferenceViewController.swift */; }; BC9CFA9323BDE8B700917EEF /* IOStreamView.swift in Sources */ = {isa = PBXBuildFile; fileRef = BC9CFA9223BDE8B700917EEF /* IOStreamView.swift */; }; BC9F9C7826F8C16600B01ED0 /* Choreographer.swift in Sources */ = {isa = PBXBuildFile; fileRef = BC9F9C7726F8C16600B01ED0 /* Choreographer.swift */; }; + BCA3A5252BC4ED220083BBB1 /* RTMPTimestamp.swift in Sources */ = {isa = PBXBuildFile; fileRef = BCA3A5242BC4ED220083BBB1 /* RTMPTimestamp.swift */; }; + BCA3A5272BC507880083BBB1 /* RTMPTimestampTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = BCA3A5262BC507880083BBB1 /* RTMPTimestampTests.swift */; }; BCA7C24F2A91AA0500882D85 /* IOStreamRecorderTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = BCA7C24E2A91AA0500882D85 /* IOStreamRecorderTests.swift */; }; BCAD0C18263ED67F00ADFB80 /* SampleVideo_360x240_5mb@m4v.m3u8 in Resources */ = {isa = PBXBuildFile; fileRef = BCAD0C16263ED67F00ADFB80 /* SampleVideo_360x240_5mb@m4v.m3u8 */; }; BCAD0C19263ED67F00ADFB80 /* SampleVideo_360x240_5mb@m4v in Resources */ = {isa = PBXBuildFile; fileRef = BCAD0C17263ED67F00ADFB80 /* SampleVideo_360x240_5mb@m4v */; }; @@ -636,6 +638,8 @@ BC959F1129717EDB0067BA97 /* PreferenceViewController.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = PreferenceViewController.swift; sourceTree = ""; }; BC9CFA9223BDE8B700917EEF /* IOStreamView.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = IOStreamView.swift; sourceTree = ""; }; BC9F9C7726F8C16600B01ED0 /* Choreographer.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Choreographer.swift; sourceTree = ""; }; + BCA3A5242BC4ED220083BBB1 /* RTMPTimestamp.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = RTMPTimestamp.swift; sourceTree = ""; }; + BCA3A5262BC507880083BBB1 /* RTMPTimestampTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = RTMPTimestampTests.swift; sourceTree = ""; }; BCA7C24E2A91AA0500882D85 /* IOStreamRecorderTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = IOStreamRecorderTests.swift; sourceTree = ""; }; BCAD0C16263ED67F00ADFB80 /* SampleVideo_360x240_5mb@m4v.m3u8 */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = text; path = "SampleVideo_360x240_5mb@m4v.m3u8"; sourceTree = ""; }; BCAD0C17263ED67F00ADFB80 /* SampleVideo_360x240_5mb@m4v */ = {isa = PBXFileReference; lastKnownFileType = folder; path = "SampleVideo_360x240_5mb@m4v"; sourceTree = ""; }; @@ -822,6 +826,7 @@ 290686021DFDB7A6008EB7ED /* RTMPConnectionTests.swift */, 2976077E20A89FBB00DCF24F /* RTMPMessageTests.swift */, 035AFA032263868E009DD0BB /* RTMPStreamTests.swift */, + BCA3A5262BC507880083BBB1 /* RTMPTimestampTests.swift */, ); path = RTMP; sourceTree = ""; @@ -1006,7 +1011,6 @@ 295018191FFA196800358E10 /* Codec */, 291C2AD21CE9FF48006F042B /* Core */, BC03945D2AA8AFDD006EDE38 /* Extension */, - BC1DC5022A02893600E928ED /* FLV */, BC0BF4F329866FB700D72CB4 /* IO */, 291C2ACF1CE9FF2B006F042B /* ISO */, 291C2ACE1CE9FF25006F042B /* RTMP */, @@ -1107,6 +1111,7 @@ 29DF20652312A436004057C3 /* RTMPSocketCompatible.swift */, 29B876AA1CD70B2800FC07DA /* RTMPStream.swift */, BC558267240BB40E00011AC0 /* RTMPStreamInfo.swift */, + BCA3A5242BC4ED220083BBB1 /* RTMPTimestamp.swift */, 294852551D84BFAD002DE492 /* RTMPTSocket.swift */, ); path = RTMP; @@ -1193,13 +1198,6 @@ path = IO; sourceTree = ""; }; - BC1DC5022A02893600E928ED /* FLV */ = { - isa = PBXGroup; - children = ( - ); - path = FLV; - sourceTree = ""; - }; BC3004FA296C3FC400119932 /* Extension */ = { isa = PBXGroup; children = ( @@ -1821,6 +1819,7 @@ BC570B4828E9ACC10098A12C /* IOUnit.swift in Sources */, 2976A4861D4903C300B53EF2 /* DeviceUtil.swift in Sources */, B34239852B9FD3E30068C3FB /* AudioNode.swift in Sources */, + BCA3A5252BC4ED220083BBB1 /* RTMPTimestamp.swift in Sources */, BC7C56BB299E595000C41A9B /* VideoCodecSettings.swift in Sources */, 29B876881CD70AE800FC07DA /* TSPacket.swift in Sources */, BC22EEEE2AAF50F200E3406D /* Codec.swift in Sources */, @@ -1885,6 +1884,7 @@ BC03945F2AA8AFF5006EDE38 /* ExpressibleByIntegerLiteral+ExtensionTests.swift in Sources */, 290EA8AA1DFB61E700053022 /* CRC32Tests.swift in Sources */, 035AFA042263868E009DD0BB /* RTMPStreamTests.swift in Sources */, + BCA3A5272BC507880083BBB1 /* RTMPTimestampTests.swift in Sources */, 290686031DFDB7A7008EB7ED /* RTMPConnectionTests.swift in Sources */, BCC9E9092636FF7400948774 /* DataBufferTests.swift in Sources */, ); diff --git a/Sources/RTMP/RTMPMuxer.swift b/Sources/RTMP/RTMPMuxer.swift index c4a76c8e4..7eb8c4498 100644 --- a/Sources/RTMP/RTMPMuxer.swift +++ b/Sources/RTMP/RTMPMuxer.swift @@ -13,7 +13,11 @@ final class RTMPMuxer { } var buffer = Data([RTMPMuxer.aac, FLVAACPacketType.seq.rawValue]) buffer.append(contentsOf: AudioSpecificConfig(formatDescription: audioFormat.formatDescription).bytes) - stream?.outputAudio(buffer, withTimestamp: 0) + stream?.doOutput( + .zero, + chunkStreamId: FLVTagType.audio.streamId, + message: RTMPAudioMessage(streamId: 0, timestamp: 0, payload: buffer) + ) case .playing: if let audioFormat { audioBuffer = AVAudioCompressedBuffer(format: audioFormat, packetCapacity: 1, maximumPacketSize: 1024 * Int(audioFormat.channelCount)) @@ -30,24 +34,29 @@ final class RTMPMuxer { didSet { switch stream?.readyState { case .publishing: - guard let videoFormat else { - return - } - switch CMFormatDescriptionGetMediaSubType(videoFormat) { - case kCMVideoCodecType_H264: + switch videoFormat?.mediaSubType { + case .h264?: guard let avcC = AVCDecoderConfigurationRecord.getData(videoFormat) else { return } var buffer = Data([FLVFrameType.key.rawValue << 4 | FLVVideoCodec.avc.rawValue, FLVAVCPacketType.seq.rawValue, 0, 0, 0]) buffer.append(avcC) - stream?.outputVideo(buffer, withTimestamp: 0) - case kCMVideoCodecType_HEVC: + stream?.doOutput( + .zero, + chunkStreamId: FLVTagType.video.streamId, + message: RTMPVideoMessage(streamId: 0, timestamp: 0, payload: buffer) + ) + case .hevc?: guard let hvcC = HEVCDecoderConfigurationRecord.getData(videoFormat) else { return } var buffer = Data([0b10000000 | FLVFrameType.key.rawValue << 4 | FLVVideoPacketType.sequenceStart.rawValue, 0x68, 0x76, 0x63, 0x31]) buffer.append(hvcC) - stream?.outputVideo(buffer, withTimestamp: 0) + stream?.doOutput( + .zero, + chunkStreamId: FLVTagType.video.streamId, + message: RTMPVideoMessage(streamId: 0, timestamp: 0, payload: buffer) + ) default: break } @@ -60,10 +69,9 @@ final class RTMPMuxer { } var isRunning: Atomic = .init(false) - private var videoTimeStamp: CMTime = .zero private var audioBuffer: AVAudioCompressedBuffer? - private var audioTimeStamp: AVAudioTime = .init(hostTime: 0) - private let compositiionTimeOffset: CMTime = .init(value: 3, timescale: 30) + private var audioTimestamp: RTMPTimestamp = .init() + private var videoTimestamp: RTMPTimestamp = .init() private weak var stream: RTMPStream? init(_ stream: RTMPStream) { @@ -74,11 +82,10 @@ final class RTMPMuxer { let payload = message.payload let codec = message.codec stream?.info.byteCount.mutate { $0 += Int64(payload.count) } - + audioTimestamp.update(message, chunkType: type) guard let stream, message.codec.isSupported else { return } - switch payload[1] { case FLVAACPacketType.seq.rawValue: let config = AudioSpecificConfig(bytes: [UInt8](payload[codec.headerSize.. Int32 { - guard sampleBuffer.decodeTimeStamp.isValid, sampleBuffer.decodeTimeStamp != sampleBuffer.presentationTimeStamp else { - return 0 - } - return Int32((sampleBuffer.presentationTimeStamp - videoTimeStamp + compositiionTimeOffset).seconds * 1000) } } @@ -210,10 +203,10 @@ extension RTMPMuxer: Running { guard !isRunning.value else { return } - audioTimeStamp = .init(hostTime: 0) - videoTimeStamp = .zero audioFormat = nil videoFormat = nil + audioTimestamp.clear() + videoTimestamp.clear() isRunning.mutate { $0 = true } } diff --git a/Sources/RTMP/RTMPStream.swift b/Sources/RTMP/RTMPStream.swift index b8ff8e678..8d3806b17 100644 --- a/Sources/RTMP/RTMPStream.swift +++ b/Sources/RTMP/RTMPStream.swift @@ -234,17 +234,13 @@ open class RTMPStream: IOStream { public var fcPublishName: String? var id: UInt32 = RTMPStream.defaultID - var audioTimestamp: Double = 0.0 - var videoTimestamp: Double = 0.0 + var frameCount: UInt16 = 0 private(set) lazy var muxer = { return RTMPMuxer(self) }() private var messages: [RTMPCommandMessage] = [] private var startedAt = Date() - private var frameCount: UInt16 = 0 private var dispatcher: (any EventDispatcherConvertible)! - private var audioWasSent = false - private var videoWasSent = false private var pausedStatus = PausedStatus(hasAudio: false, hasVideo: false) private var howToPublish: RTMPStream.HowToPublish = .live private var dataTimestamps: [String: Date] = .init() @@ -370,12 +366,21 @@ open class RTMPStream: IOStream { } /// Sends a message on a published stream to all subscribing clients. - /// @param handlerName - /// @param arguemnts - /// @param isResetTimestamp A workaround option for sending timestamps as 0 in some services. + /// + /// ``` + /// // To add a metadata to a live stream sent to an RTMP Service. + /// stream.send("@setDataFrame", "onMetaData", metaData) + /// // To clear a metadata that has already been set in the stream. + /// stream.send("@clearDataFrame", "onMetaData"); + /// ``` + /// + /// - Parameters: + /// - handlerName: The message to send. + /// - arguemnts: Optional arguments. + /// - isResetTimestamp: A workaround option for sending timestamps as 0 in some services. public func send(handlerName: String, arguments: Any?..., isResetTimestamp: Bool = false) { lockQueue.async { - guard let connection = self.connection, self.readyState == .publishing(muxer: self.muxer) else { + guard self.readyState == .publishing(muxer: self.muxer) else { return } if isResetTimestamp { @@ -383,19 +388,18 @@ open class RTMPStream: IOStream { } let dataWasSent = self.dataTimestamps[handlerName] == nil ? false : true let timestmap: UInt32 = dataWasSent ? UInt32((self.dataTimestamps[handlerName]?.timeIntervalSinceNow ?? 0) * -1000) : UInt32(self.startedAt.timeIntervalSinceNow * -1000) - let chunk = RTMPChunk( - type: dataWasSent ? RTMPChunkType.one : RTMPChunkType.zero, - streamId: RTMPChunk.StreamID.data.rawValue, + self.doOutput( + dataWasSent ? RTMPChunkType.one : RTMPChunkType.zero, + chunkStreamId: RTMPChunk.StreamID.data.rawValue, message: RTMPDataMessage( streamId: self.id, objectEncoding: self.objectEncoding, timestamp: timestmap, handlerName: handlerName, arguments: arguments - )) - let length = connection.socket?.doOutput(chunk: chunk) ?? 0 + ) + ) self.dataTimestamps[handlerName] = .init() - self.info.byteCount.mutate { $0 += Int64(length) } } } @@ -455,13 +459,9 @@ open class RTMPStream: IOStream { messages.removeAll() case .play: startedAt = .init() - videoTimestamp = 0 - audioTimestamp = 0 case .publish: bitrateStrategy.setUp() startedAt = .init() - videoWasSent = false - audioWasSent = false dataTimestamps.removeAll() case .publishing: let metadata = makeMetaData() @@ -500,44 +500,23 @@ open class RTMPStream: IOStream { ))) } - func on(timer: Timer) { - currentFPS = frameCount - frameCount = 0 - info.on(timer: timer) - } - - func outputAudio(_ buffer: Data, withTimestamp: Double) { - guard let connection, readyState == .publishing(muxer: muxer) else { + func doOutput(_ type: RTMPChunkType, chunkStreamId: UInt16, message: RTMPMessage) { + guard let socket = connection?.socket else { return } - let type: FLVTagType = .audio - let length = connection.socket?.doOutput(chunk: RTMPChunk( - type: audioWasSent ? .one : .zero, - streamId: type.streamId, - message: RTMPAudioMessage(streamId: id, timestamp: UInt32(audioTimestamp), payload: buffer) - )) ?? 0 - audioWasSent = true + message.streamId = id + let length = socket.doOutput(chunk: .init( + type: type, + streamId: chunkStreamId, + message: message + )) info.byteCount.mutate { $0 += Int64(length) } - audioTimestamp = withTimestamp + (audioTimestamp - floor(audioTimestamp)) } - func outputVideo(_ buffer: Data, withTimestamp: Double) { - guard let connection, readyState == .publishing(muxer: muxer) else { - return - } - let type: FLVTagType = .video - let length = connection.socket?.doOutput(chunk: RTMPChunk( - type: videoWasSent ? .one : .zero, - streamId: type.streamId, - message: RTMPVideoMessage(streamId: id, timestamp: UInt32(videoTimestamp), payload: buffer) - )) ?? 0 - if !videoWasSent { - logger.debug("first video frame was sent") - } - videoWasSent = true - info.byteCount.mutate { $0 += Int64(length) } - videoTimestamp = withTimestamp + (videoTimestamp - floor(videoTimestamp)) - frameCount += 1 + func on(timer: Timer) { + currentFPS = frameCount + frameCount = 0 + info.on(timer: timer) } @objc diff --git a/Sources/RTMP/RTMPTimestamp.swift b/Sources/RTMP/RTMPTimestamp.swift new file mode 100644 index 000000000..55e143547 --- /dev/null +++ b/Sources/RTMP/RTMPTimestamp.swift @@ -0,0 +1,82 @@ +import AVFoundation +import CoreMedia +import Foundation + +protocol RTMPTimeConvertible { + var seconds: TimeInterval { get } +} + +private let kRTMPTimestamp_defaultTimeInterval: TimeInterval = 0 +private let kRTMPTimestamp_compositiionTimeOffset = CMTime(value: 3, timescale: 30) + +struct RTMPTimestamp { + private var startedAt = kRTMPTimestamp_defaultTimeInterval + private var updatedAt = kRTMPTimestamp_defaultTimeInterval + private var timedeltaFraction: TimeInterval = kRTMPTimestamp_defaultTimeInterval + + mutating func update(_ value: T) -> UInt32 { + if startedAt == 0 { + startedAt = value.seconds + updatedAt = value.seconds + return 0 + } else { + var timedelta = (value.seconds - updatedAt) * 1000 + timedeltaFraction += timedelta.truncatingRemainder(dividingBy: 1) + if 1 <= timedeltaFraction { + timedeltaFraction -= 1 + timedelta += 1 + } + updatedAt = value.seconds + return UInt32(timedelta) + } + } + + mutating func update(_ message: RTMPMessage, chunkType: RTMPChunkType) { + switch chunkType { + case .zero: + if startedAt == 0 { + startedAt = TimeInterval(message.timestamp) / 1000 + updatedAt = TimeInterval(message.timestamp) / 1000 + } else { + updatedAt = TimeInterval(message.timestamp) / 1000 + } + default: + updatedAt += TimeInterval(message.timestamp) / 1000 + } + } + + mutating func clear() { + startedAt = kRTMPTimestamp_defaultTimeInterval + updatedAt = kRTMPTimestamp_defaultTimeInterval + timedeltaFraction = kRTMPTimestamp_defaultTimeInterval + } + + func getCompositionTime(_ sampleBuffer: CMSampleBuffer) -> Int32 { + guard sampleBuffer.decodeTimeStamp.isValid, sampleBuffer.decodeTimeStamp != sampleBuffer.presentationTimeStamp else { + return 0 + } + let compositionTime = (sampleBuffer.presentationTimeStamp + kRTMPTimestamp_compositiionTimeOffset).seconds - updatedAt + return Int32(compositionTime * 1000) + } +} + +extension AVAudioTime: RTMPTimeConvertible { + var seconds: TimeInterval { + AVAudioTime.seconds(forHostTime: hostTime) + } +} + +extension RTMPTimestamp where T == AVAudioTime { + var value: AVAudioTime { + return AVAudioTime(hostTime: AVAudioTime.hostTime(forSeconds: updatedAt)) + } +} + +extension CMTime: RTMPTimeConvertible { +} + +extension RTMPTimestamp where T == CMTime { + var value: CMTime { + return CMTime(seconds: updatedAt, preferredTimescale: 1000) + } +} diff --git a/Tests/RTMP/RTMPTimestampTests.swift b/Tests/RTMP/RTMPTimestampTests.swift new file mode 100644 index 000000000..f8debec2c --- /dev/null +++ b/Tests/RTMP/RTMPTimestampTests.swift @@ -0,0 +1,11 @@ +import Foundation +import XCTest +@testable import HaishinKit + +final class RTMPTimestampTests: XCTestCase { + func testCMTime() { + } + + func testAVAudioTime() { + } +}