You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

SampleUploader.swift 5.2KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154
  1. /*
  2. * Copyright @ 2021-present 8x8, Inc.
  3. *
  4. * Licensed under the Apache License, Version 2.0 (the "License");
  5. * you may not use this file except in compliance with the License.
  6. * You may obtain a copy of the License at
  7. *
  8. * http://www.apache.org/licenses/LICENSE-2.0
  9. *
  10. * Unless required by applicable law or agreed to in writing, software
  11. * distributed under the License is distributed on an "AS IS" BASIS,
  12. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. * See the License for the specific language governing permissions and
  14. * limitations under the License.
  15. */
  16. import Foundation
  17. import ReplayKit
  18. private enum Constants {
  19. static let bufferMaxLength = 10240
  20. }
  21. class SampleUploader {
  22. private static var imageContext = CIContext(options: nil)
  23. @Atomic private var isReady: Bool = false
  24. private var connection: SocketConnection
  25. private var dataToSend: Data?
  26. private var byteIndex = 0
  27. private let serialQueue: DispatchQueue
  28. init(connection: SocketConnection) {
  29. self.connection = connection
  30. self.serialQueue = DispatchQueue(label: "org.jitsi.meet.broadcast.sampleUploader")
  31. setupConnection()
  32. }
  33. @discardableResult func send(sample buffer: CMSampleBuffer) -> Bool {
  34. guard isReady == true else {
  35. return false
  36. }
  37. isReady = false
  38. dataToSend = prepare(sample: buffer)
  39. byteIndex = 0
  40. serialQueue.async { [weak self] in
  41. self?.sendDataChunk()
  42. }
  43. return true
  44. }
  45. }
  46. private extension SampleUploader {
  47. func setupConnection() {
  48. connection.didOpen = { [weak self] in
  49. self?.isReady = true
  50. }
  51. connection.streamHasSpaceAvailable = { [weak self] in
  52. self?.serialQueue.async {
  53. self?.isReady = !(self?.sendDataChunk() ?? true)
  54. }
  55. }
  56. }
  57. @discardableResult func sendDataChunk() -> Bool {
  58. guard let dataToSend = dataToSend else {
  59. return false
  60. }
  61. var bytesLeft = dataToSend.count - byteIndex
  62. var length = bytesLeft > Constants.bufferMaxLength ? Constants.bufferMaxLength : bytesLeft
  63. length = dataToSend[byteIndex..<(byteIndex + length)].withUnsafeBytes {
  64. guard let ptr = $0.bindMemory(to: UInt8.self).baseAddress else {
  65. return 0
  66. }
  67. return connection.writeToStream(buffer: ptr, maxLength: length)
  68. }
  69. if length > 0 {
  70. byteIndex += length
  71. bytesLeft -= length
  72. if bytesLeft == 0 {
  73. self.dataToSend = nil
  74. byteIndex = 0
  75. }
  76. } else {
  77. print("writeBufferToStream failure")
  78. }
  79. return true
  80. }
  81. func prepare(sample buffer: CMSampleBuffer) -> Data? {
  82. guard let imageBuffer = CMSampleBufferGetImageBuffer(buffer) else {
  83. print("image buffer not available")
  84. return nil
  85. }
  86. CVPixelBufferLockBaseAddress(imageBuffer, .readOnly)
  87. let scaleFactor = 2.0
  88. let width = CVPixelBufferGetWidth(imageBuffer)/Int(scaleFactor)
  89. let height = CVPixelBufferGetHeight(imageBuffer)/Int(scaleFactor)
  90. let orientation = CMGetAttachment(buffer, key: RPVideoSampleOrientationKey as CFString, attachmentModeOut: nil)?.uintValue ?? 0
  91. let scaleTransform = CGAffineTransform(scaleX: CGFloat(1.0/scaleFactor), y: CGFloat(1.0/scaleFactor))
  92. let bufferData = self.jpegData(from: imageBuffer, scale: scaleTransform)
  93. CVPixelBufferUnlockBaseAddress(imageBuffer, .readOnly)
  94. guard let messageData = bufferData else {
  95. print("corrupted image buffer")
  96. return nil
  97. }
  98. let httpResponse = CFHTTPMessageCreateResponse(nil, 200, nil, kCFHTTPVersion1_1).takeRetainedValue()
  99. CFHTTPMessageSetHeaderFieldValue(httpResponse, "Content-Length" as CFString, String(messageData.count) as CFString)
  100. CFHTTPMessageSetHeaderFieldValue(httpResponse, "Buffer-Width" as CFString, String(width) as CFString)
  101. CFHTTPMessageSetHeaderFieldValue(httpResponse, "Buffer-Height" as CFString, String(height) as CFString)
  102. CFHTTPMessageSetHeaderFieldValue(httpResponse, "Buffer-Orientation" as CFString, String(orientation) as CFString)
  103. CFHTTPMessageSetBody(httpResponse, messageData as CFData)
  104. let serializedMessage = CFHTTPMessageCopySerializedMessage(httpResponse)?.takeRetainedValue() as Data?
  105. return serializedMessage
  106. }
  107. func jpegData(from buffer: CVPixelBuffer, scale scaleTransform: CGAffineTransform) -> Data? {
  108. var image = CIImage(cvPixelBuffer: buffer)
  109. image = image.transformed(by: scaleTransform)
  110. guard let colorSpace = image.colorSpace else {
  111. return nil
  112. }
  113. let options: [CIImageRepresentationOption: Float] = [kCGImageDestinationLossyCompressionQuality as CIImageRepresentationOption: 1.0]
  114. let imageData = SampleUploader.imageContext.jpegRepresentation(of: image, colorSpace: colorSpace, options: options)
  115. return imageData
  116. }
  117. }