fix command and data amf3

app/src/main/java/com/pedro/streamer/rotation/CameraFragment.kt

@@ -37,7 +37,9 @@ import com.pedro.extrasources.CameraXSource
 import com.pedro.library.base.StreamBase
 import com.pedro.library.base.recording.RecordController
 import com.pedro.library.generic.GenericStream
+import com.pedro.library.rtmp.RtmpStream
 import com.pedro.library.util.BitrateAdapter
+import com.pedro.rtmp.amf.AmfVersion
 import com.pedro.streamer.R
 import com.pedro.streamer.utils.PathUtils
 import com.pedro.streamer.utils.toast
@@ -77,8 +79,9 @@ class CameraFragment: Fragment(), ConnectChecker {
   }
 
   val genericStream: StreamBase by lazy {
-    GenericStream(requireContext(), this).apply {
+    RtmpStream(requireContext(), this).apply {
       getGlInterface().autoHandleOrientation = true
+      getStreamClient().setAmfVersion(AmfVersion.VERSION_3)
     }
   }
   private lateinit var surfaceView: SurfaceView

common/src/main/java/com/pedro/common/Extensions.kt

@@ -179,6 +179,16 @@ fun MediaFormat.getLongSafe(name: String): Long? {
   return try { getLong(name) } catch (e: Exception) { null }
 }
 
+fun Int.getUInt29Size(): Int {
+  val v = this and 0x1FFFFFFF
+  return when {
+    v < 0x80 -> 1
+    v < 0x4000 -> 2
+    v < 0x200000 -> 3
+    else -> 4
+  }
+}
+
 fun Int.toUInt16(): ByteArray = byteArrayOf((this ushr 8).toByte(), this.toByte())
 
 fun Int.toUInt24(): ByteArray {
@@ -243,6 +253,24 @@ fun InputStream.readUInt32(): Int {
   return data.toUInt32()
 }
 
+@Throws(IOException::class)
+fun InputStream.readUInt29(): Int {
+  var result = 0
+  var bytesRead = 0
+  while (bytesRead < 3) {
+    val b = read()
+    bytesRead++
+    if (b and 0x80 != 0) {
+      result = (result shl 7) or (b and 0x7F)
+    } else {
+      result = (result shl 7) or b
+      return if (result >= 0x10000000) result - 0x20000000 else result
+    }
+  }
+  result = (result shl 8) or read()
+  return if (result >= 0x10000000) result - 0x20000000 else result
+}
+
 @Throws(IOException::class)
 fun InputStream.readUInt24(): Int {
   val data = ByteArray(3)
@@ -265,6 +293,28 @@ fun OutputStream.writeUInt32(value: Int) {
   write(value.toUInt32())
 }
 
+fun OutputStream.writeUInt29(value: Int) {
+  val u29 = value and 0x1FFFFFFF
+  when {
+    u29 < 0x80 -> write(u29)
+    u29 < 0x4000 -> {
+      write((u29 shr 7) or 0x80)
+      write(u29 and 0x7F)
+    }
+    u29 < 0x200000 -> {
+      write((u29 shr 14) or 0x80)
+      write(((u29 shr 7) and 0x7F) or 0x80)
+      write(u29 and 0x7F)
+    }
+    else -> {
+      write((u29 shr 22) or 0x80)
+      write(((u29 shr 15) and 0x7F) or 0x80)
+      write(((u29 shr 8) and 0x7F) or 0x80)
+      write(u29 and 0xFF)
+    }
+  }
+}
+
 fun OutputStream.writeUInt24(value: Int) {
   write(value.toUInt24())
 }

library/src/main/java/com/pedro/library/util/streamclient/RtmpStreamClient.kt

@@ -17,6 +17,7 @@
 package com.pedro.library.util.streamclient
 
 import com.pedro.common.socket.base.SocketType
+import com.pedro.rtmp.amf.AmfVersion
 import com.pedro.rtmp.rtmp.RtmpClient
 import javax.net.ssl.TrustManager
 
@@ -223,4 +224,8 @@ class RtmpStreamClient(
   fun shouldFailOnRead(enabled: Boolean) {
     rtmpClient.shouldFailOnRead = enabled
   }
+
+  fun setAmfVersion(version: AmfVersion) {
+    rtmpClient.setAmfVersion(AmfVersion.VERSION_3)
+  }
 }

rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfData.kt

@@ -31,8 +31,11 @@ abstract class AmfData {
      * Read unknown AmfData and convert it to specific class
      */
     @Throws(IOException::class)
-    fun getAmfData(input: InputStream): AmfData {
-      val amfData = when (val type = getMarkType(input.read())) {
+    fun getAmfData(input: InputStream) = getAmfData(input.read(), input)
+
+    @Throws(IOException::class)
+    fun getAmfData(type: Int, input: InputStream): AmfData {
+      val amfData = when (val type = getMarkType(type)) {
         AmfType.NUMBER -> AmfNumber()
         AmfType.BOOLEAN -> AmfBoolean()
         AmfType.STRING -> AmfString()

rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Array.kt

@@ -16,25 +16,91 @@
 
 package com.pedro.rtmp.amf.v3
 
+import com.pedro.common.getUInt29Size
+import com.pedro.common.readUInt29
+import com.pedro.common.writeUInt29
+import java.io.IOException
 import java.io.InputStream
 import java.io.OutputStream
 
 /**
  * Created by pedro on 29/04/21.
+ *
+ * AMF3 arrays have an associative part (name/value pairs, empty string terminated)
+ * followed by a dense part of U29A-value items. References can't be resolved without
+ * payload scope, so they are rejected.
  */
 class Amf3Array(val items: MutableList<Amf3Data> = mutableListOf()): Amf3Data() {
 
+  private val associative = LinkedHashMap<Amf3String, Amf3Data>()
+  private var referenceSize = -1
+
+  fun getProperty(name: String): Amf3Data? {
+    associative.forEach {
+      if (it.key.value == name) {
+        return it.value
+      }
+    }
+    return null
+  }
+
+  fun setProperty(name: String, data: Amf3Data) {
+    val existingKey = associative.keys.find { it.value == name }
+    if (existingKey != null) associative[existingKey] = data
+    else associative[Amf3String(name)] = data
+  }
+
+  @Throws(IOException::class)
   override fun readBody(input: InputStream) {
-    TODO("Not yet implemented")
+    items.clear()
+    associative.clear()
+    val u29 = input.readUInt29()
+    if (u29 and 0x01 == 0) { //reference to a previous array of this payload, no tables to resolve it
+      referenceSize = u29.getUInt29Size()
+      return
+    }
+    referenceSize = -1
+    val denseCount = u29 ushr 1
+    while (true) {
+      val key = Amf3String()
+      key.readBody(input)
+      if (key.isReference) throw IOException("AMF3 string reference as array key is not supported")
+      if (key.value.isEmpty()) break
+      associative[key] = getAmf3Data(input)
+    }
+    repeat(denseCount) {
+      items.add(getAmf3Data(input))
+    }
   }
 
+  @Throws(IOException::class)
   override fun writeBody(output: OutputStream) {
-    TODO("Not yet implemented")
+    output.writeUInt29((items.size shl 1) or 1)
+    associative.forEach { (key, value) ->
+      key.writeBody(output)
+      value.writeHeader(output)
+      value.writeBody(output)
+    }
+    output.write(0x01) //empty string, end of associative part
+    items.forEach {
+      it.writeHeader(output)
+      it.writeBody(output)
+    }
   }
 
   override fun getType(): Amf3Type = Amf3Type.ARRAY
 
   override fun getSize(): Int {
-    TODO("Not yet implemented")
+    if (referenceSize != -1) return referenceSize
+    var size = ((items.size shl 1) or 1).getUInt29Size() + 1
+    associative.forEach { (key, value) ->
+      size += key.getSize() + value.getSize() + 1
+    }
+    items.forEach { size += it.getSize() + 1 }
+    return size
+  }
+
+  override fun toString(): String {
+    return "Amf3Array(items=$items, associative=${associative.entries.joinToString { "${it.key.value}=${it.value}" }})"
   }
-}
+}

rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Data.kt

@@ -41,6 +41,7 @@ abstract class Amf3Data {
         Amf3Type.UNDEFINED -> Amf3Undefined()
         Amf3Type.ARRAY -> Amf3Array()
         Amf3Type.DICTIONARY -> Amf3Dictionary()
+        Amf3Type.XML_DOC -> Amf3XmlDocument()
         Amf3Type.TRUE -> Amf3True()
         Amf3Type.FALSE -> Amf3False()
         else -> throw IOException("Unimplemented AMF3 data type: ${type.name}")

rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Dictionary.kt

@@ -16,25 +16,71 @@
 
 package com.pedro.rtmp.amf.v3
 
+import com.pedro.common.getUInt29Size
+import com.pedro.common.readUInt29
+import com.pedro.common.writeUInt29
+import java.io.IOException
 import java.io.InputStream
 import java.io.OutputStream
 
 /**
  * Created by pedro on 29/04/21.
+ *
+ * Unlike Amf3Object, dictionary keys are full value-types (marker included) and entries
+ * are preceded by a U29 count plus a weak-keys flag byte. Only string keys are supported.
+ * Introduced in Flash Player 10; several servers don't understand it, prefer Amf3Object
+ * or the associative part of Amf3Array for interop.
  */
-class Amf3Dictionary(private val properties: HashMap<Amf3String, Amf3Data> = LinkedHashMap()): Amf3Object(properties) {
+class Amf3Dictionary(properties: LinkedHashMap<Amf3String, Amf3Data> = LinkedHashMap()): Amf3Object(properties) {
 
+  @Throws(IOException::class)
   override fun readBody(input: InputStream) {
-    TODO("Not yet implemented")
+    properties.clear()
+    val u29 = input.readUInt29()
+    bodySize = u29.getUInt29Size()
+    if (u29 and 0x01 == 0) { //reference to a previous dictionary of this payload, no tables to resolve it
+      reference = u29 ushr 1
+      return
+    }
+    reference = -1
+    val count = u29 ushr 1
+    input.read() //weak keys flag, ignored
+    bodySize += 1
+    repeat(count) {
+      val key = getAmf3Data(input)
+      if (key !is Amf3String) throw IOException("Only string keys are supported in AMF3 dictionary")
+      if (key.isReference) throw IOException("AMF3 string reference as dictionary key is not supported")
+      bodySize += key.getSize() + 1
+      val value = getAmf3Data(input)
+      bodySize += value.getSize() + 1
+      properties[key] = value
+    }
   }
 
+  @Throws(IOException::class)
   override fun writeBody(output: OutputStream) {
-    TODO("Not yet implemented")
+    output.writeUInt29((properties.size shl 1) or 1)
+    output.write(0x00) //strong keys
+    properties.forEach { (key, value) ->
+      key.writeHeader(output)
+      key.writeBody(output)
+      value.writeHeader(output)
+      value.writeBody(output)
+    }
   }
 
   override fun getType(): Amf3Type = Amf3Type.DICTIONARY
 
   override fun getSize(): Int {
-    TODO("Not yet implemented")
+    if (reference != -1) return bodySize
+    var size = ((properties.size shl 1) or 1).getUInt29Size() + 1
+    properties.forEach { (key, value) ->
+      size += key.getSize() + value.getSize() + 2
+    }
+    return size
+  }
+
+  override fun toString(): String {
+    return "Amf3Dictionary(properties=${properties.entries.joinToString { "${it.key.value}=${it.value}" }})"
   }
-}
+}

rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Integer.kt

@@ -16,25 +16,27 @@
 
 package com.pedro.rtmp.amf.v3
 
+import com.pedro.common.getUInt29Size
+import com.pedro.common.readUInt29
+import com.pedro.common.writeUInt29
 import java.io.InputStream
 import java.io.OutputStream
 
 /**
  * Created by pedro on 29/04/21.
  */
-class Amf3Integer(private val value: Int = 0): Amf3Data() {
+class Amf3Integer(var value: Int = 0): Amf3Data() {
+
 
   override fun readBody(input: InputStream) {
-    TODO("Not yet implemented")
+    this.value = input.readUInt29()
   }
 
   override fun writeBody(output: OutputStream) {
-    TODO("Not yet implemented")
+    output.writeUInt29(value)
   }
 
   override fun getType(): Amf3Type = Amf3Type.INTEGER
 
-  override fun getSize(): Int {
-    TODO("Not yet implemented")
-  }
+  override fun getSize(): Int = value.getUInt29Size()
 }