diff --git a/app/src/main/java/com/pedro/streamer/rotation/CameraFragment.kt b/app/src/main/java/com/pedro/streamer/rotation/CameraFragment.kt
index 1b085cf82d..54ff230ade 100644
--- a/app/src/main/java/com/pedro/streamer/rotation/CameraFragment.kt
+++ b/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
diff --git a/common/src/main/java/com/pedro/common/Extensions.kt b/common/src/main/java/com/pedro/common/Extensions.kt
index 06cdf9259b..f05bfa4f01 100644
--- a/common/src/main/java/com/pedro/common/Extensions.kt
+++ b/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())
 }
diff --git a/library/src/main/java/com/pedro/library/util/streamclient/RtmpStreamClient.kt b/library/src/main/java/com/pedro/library/util/streamclient/RtmpStreamClient.kt
index 77f72f432a..8f23a163e5 100644
--- a/library/src/main/java/com/pedro/library/util/streamclient/RtmpStreamClient.kt
+++ b/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)
+  }
 }
\ No newline at end of file
diff --git a/rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfData.kt b/rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfData.kt
index fddfb23812..801e8add74 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/amf/v0/AmfData.kt
+++ b/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()
diff --git a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Array.kt b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Array.kt
index 5b1f2e6602..1c1b829a46 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Array.kt
+++ b/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}" }})"
   }
-}
\ No newline at end of file
+}
diff --git a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Data.kt b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Data.kt
index bbaec9a610..d08b5320ee 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Data.kt
+++ b/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}")
diff --git a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Dictionary.kt b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Dictionary.kt
index 1c480ff3dd..4ef37be75a 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Dictionary.kt
+++ b/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}" }})"
   }
-}
\ No newline at end of file
+}
diff --git a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Integer.kt b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Integer.kt
index fe908fef16..7a1842f80d 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Integer.kt
+++ b/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()
 }
\ No newline at end of file
diff --git a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Object.kt b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Object.kt
index d7109b76b0..df40acfab6 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Object.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Object.kt
@@ -16,16 +16,36 @@
 
 package com.pedro.rtmp.amf.v3
 
-import com.pedro.rtmp.amf.v0.AmfNull
+import com.pedro.common.getUInt29Size
+import com.pedro.common.readUInt16
+import com.pedro.common.readUInt29
+import com.pedro.common.readUntil
+import com.pedro.common.writeUInt29
+import java.io.IOException
 import java.io.InputStream
 import java.io.OutputStream
 
 /**
  * Created by pedro on 29/04/21.
+ *
+ * Written always as an anonymous dynamic object (U29O-traits 0x0B, empty class name).
+ * Reading supports inline traits with sealed and dynamic members; references to the
+ * implicit tables can't be resolved without payload scope, so they are rejected.
  */
-open class Amf3Object(private val properties: HashMap<Amf3String, Amf3Data> = LinkedHashMap()): Amf3Data() {
+open class Amf3Object(protected val properties: LinkedHashMap<Amf3String, Amf3Data> = LinkedHashMap()): Amf3Data() {
+
+  //traits U29 + empty class name + end of dynamic members
+  protected var bodySize = 3
+  protected var reference = -1
+  var className = ""
+    protected set
 
-  protected var bodySize = 0
+  init {
+    properties.forEach {
+      bodySize += it.key.getSize()
+      bodySize += it.value.getSize() + 1
+    }
+  }
 
   fun getProperty(name: String): Amf3Data? {
     properties.forEach {
@@ -60,27 +80,134 @@ open class Amf3Object(private val properties: HashMap<Amf3String, Amf3Data> = Li
   }
 
   private fun putProperty(name: String, value: Amf3Data) {
-    val key = Amf3String(name)
-    val previous = properties.put(key, value)
-    bodySize += if (previous != null) {
-      value.getSize() - previous.getSize()
+    //Amf3String doesn't implement equals, so an existing key must be located by value
+    val existingKey = properties.keys.find { it.value == name }
+    if (existingKey != null) {
+      val previous = properties.put(existingKey, value)
+      bodySize += value.getSize() - (previous?.getSize() ?: 0)
     } else {
-      key.getSize() + value.getSize() + 1
+      val key = Amf3String(name)
+      properties[key] = value
+      bodySize += key.getSize() + value.getSize() + 1
     }
   }
 
-
+  @Throws(IOException::class)
   override fun readBody(input: InputStream) {
-    TODO("Not yet implemented")
+    properties.clear()
+    className = ""
+    val u29 = input.readUInt29()
+    bodySize = u29.getUInt29Size()
+    if (u29 and 0x01 == 0) { //reference to a previous object of this payload, no tables to resolve it
+      reference = u29 ushr 1
+      return
+    }
+    reference = -1
+    if (u29 and 0x02 == 0) throw IOException("AMF3 traits references are not supported")
+    val isDynamic = u29 and 0x08 != 0
+    val sealedCount = u29 ushr 4
+
+    val name = Amf3String()
+    name.readBody(input)
+    if (name.isReference) throw IOException("AMF3 string reference as class name is not supported")
+    bodySize += name.getSize()
+    className = name.value
+
+    if (u29 and 0x04 != 0) { //U29O-traits-ext: opaque bytes defined by the class
+      readExternalizable(input)
+      return
+    }
+
+    val sealedNames = mutableListOf<Amf3String>()
+    repeat(sealedCount) {
+      val key = Amf3String()
+      key.readBody(input)
+      if (key.isReference) throw IOException("AMF3 string reference as property name is not supported")
+      bodySize += key.getSize()
+      sealedNames.add(key)
+    }
+    sealedNames.forEach { key ->
+      val value = getAmf3Data(input)
+      bodySize += value.getSize() + 1
+      properties[key] = value
+    }
+    if (isDynamic) {
+      while (true) {
+        val key = Amf3String()
+        key.readBody(input)
+        if (key.isReference) throw IOException("AMF3 string reference as property name is not supported")
+        bodySize += key.getSize()
+        if (key.value.isEmpty()) break
+        val value = getAmf3Data(input)
+        bodySize += value.getSize() + 1
+        properties[key] = value
+      }
+    }
+  }
+
+  /**
+   * Externalizable payloads are opaque bytes only known by the class, so only the Red5
+   * status classes (the ones a client receives in practice) are supported:
+   * Status (also with empty class name): double clientid + UTF code, description, details, level
+   * StatusObject: UTF code, description, level + AMF3 values application, additional
+   */
+  @Throws(IOException::class)
+  private fun readExternalizable(input: InputStream) {
+    when {
+      className.isEmpty() || className.endsWith(".Status") -> {
+        val clientId = Amf3Double()
+        clientId.readBody(input)
+        bodySize += clientId.getSize()
+        properties[Amf3String("clientid")] = clientId
+        properties[Amf3String("code")] = readExternalUtf(input)
+        properties[Amf3String("description")] = readExternalUtf(input)
+        properties[Amf3String("details")] = readExternalUtf(input)
+        properties[Amf3String("level")] = readExternalUtf(input)
+      }
+      className.endsWith("StatusObject") -> {
+        properties[Amf3String("code")] = readExternalUtf(input)
+        properties[Amf3String("description")] = readExternalUtf(input)
+        properties[Amf3String("level")] = readExternalUtf(input)
+        properties[Amf3String("application")] = readExternalValue(input)
+        properties[Amf3String("additional")] = readExternalValue(input)
+      }
+      else -> throw IOException("AMF3 externalizable class $className is not supported")
+    }
   }
 
+  @Throws(IOException::class)
+  private fun readExternalUtf(input: InputStream): Amf3String {
+    val length = input.readUInt16()
+    val bytes = ByteArray(length)
+    input.readUntil(bytes)
+    bodySize += 2 + length
+    return Amf3String(String(bytes, Charsets.UTF_8))
+  }
+
+  @Throws(IOException::class)
+  private fun readExternalValue(input: InputStream): Amf3Data {
+    val value = getAmf3Data(input)
+    bodySize += value.getSize() + 1
+    return value
+  }
+
+  @Throws(IOException::class)
   override fun writeBody(output: OutputStream) {
-    TODO("Not yet implemented")
+    output.writeUInt29(0x0B) //U29O-traits: inline, dynamic, 0 sealed members
+    output.write(0x01) //empty class name (anonymous object)
+    properties.forEach { (key, value) ->
+      key.writeBody(output)
+      value.writeHeader(output)
+      value.writeBody(output)
+    }
+    output.write(0x01) //empty string, end of dynamic members
   }
 
   override fun getType(): Amf3Type = Amf3Type.OBJECT
 
-  override fun getSize(): Int {
-    TODO("Not yet implemented")
+  override fun getSize(): Int = bodySize
+
+  override fun toString(): String {
+    return "Amf3Object(className='$className', properties=${properties.entries.joinToString { "${it.key.value}=${it.value}" }})"
   }
-}
\ No newline at end of file
+}
diff --git a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3String.kt b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3String.kt
index 9301d89bfd..05b34ef5a3 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3String.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3String.kt
@@ -16,25 +16,54 @@
 
 package com.pedro.rtmp.amf.v3
 
+import com.pedro.common.getUInt29Size
+import com.pedro.common.readUInt29
+import com.pedro.common.readUntil
+import com.pedro.common.writeUInt29
+import java.io.IOException
 import java.io.InputStream
 import java.io.OutputStream
 
 /**
  * Created by pedro on 8/04/21.
  */
-class Amf3String(var value: String = ""): Amf3Data() {
+open class Amf3String(var value: String = ""): Amf3Data() {
 
+  private var bodySize: Int
+  private var reference = -1
+  val isReference: Boolean
+    get() = reference != -1
+
+  init {
+    val length = value.toByteArray(Charsets.UTF_8).size
+    bodySize = length + ((length shl 1) or 1).getUInt29Size()
+  }
+
+  @Throws(IOException::class)
   override fun readBody(input: InputStream) {
-    TODO("Not yet implemented")
+    val u29 = input.readUInt29()
+    val length = u29 ushr 1
+    if (u29 and 0x01 == 0) { //reference
+      reference = length
+      value = ""
+      bodySize = u29.getUInt29Size()
+      return
+    }
+    reference = -1
+    bodySize = length + u29.getUInt29Size()
+    val bytes = ByteArray(length)
+    input.readUntil(bytes)
+    value = String(bytes, Charsets.UTF_8)
   }
 
+  @Throws(IOException::class)
   override fun writeBody(output: OutputStream) {
-    TODO("Not yet implemented")
+    val bytes = value.toByteArray(Charsets.UTF_8)
+    output.writeUInt29((bytes.size shl 1) or 1)
+    output.write(bytes)
   }
 
   override fun getType(): Amf3Type = Amf3Type.STRING
 
-  override fun getSize(): Int {
-    TODO("Not yet implemented")
-  }
+  override fun getSize(): Int = bodySize
 }
\ No newline at end of file
diff --git a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Undefined.kt b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Undefined.kt
index cb27dbc935..611951124f 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Undefined.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3Undefined.kt
@@ -25,11 +25,11 @@ import java.io.OutputStream
 class Amf3Undefined: Amf3Data() {
 
   override fun readBody(input: InputStream) {
-    //no body to read
+    //nothing to read
   }
 
   override fun writeBody(output: OutputStream) {
-    //no body to write
+    //nothing to write
   }
 
   override fun getType(): Amf3Type = Amf3Type.UNDEFINED
diff --git a/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3XmlDocument.kt b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3XmlDocument.kt
new file mode 100644
index 0000000000..75b20fbf86
--- /dev/null
+++ b/rtmp/src/main/java/com/pedro/rtmp/amf/v3/Amf3XmlDocument.kt
@@ -0,0 +1,23 @@
+package com.pedro.rtmp.amf.v3
+
+import org.w3c.dom.Document
+import org.xml.sax.InputSource
+import org.xml.sax.SAXException
+import java.io.IOException
+import java.io.StringReader
+import javax.xml.parsers.DocumentBuilderFactory
+
+class Amf3XmlDocument(value: String = ""): Amf3String(value) {
+
+  @Throws(IOException::class, SAXException::class)
+  fun getDocument(): Document {
+    val dom = DocumentBuilderFactory.newInstance().newDocumentBuilder()
+    return dom.parse(InputSource(StringReader(value)))
+  }
+
+  override fun getType(): Amf3Type = Amf3Type.XML_DOC
+
+  override fun toString(): String {
+    return "Amf3XmlDocument value: $value"
+  }
+}
\ No newline at end of file
diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/CommandsManager.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/CommandsManager.kt
index 07ea708a74..ea2c9ee8f6 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/CommandsManager.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/CommandsManager.kt
@@ -242,4 +242,38 @@ abstract class CommandsManager {
     lastAcknowledgementSequence = 0
     bytesRead = 0
   }
+
+  fun copyInto(commandsManager: CommandsManager): CommandsManager {
+    commandsManager.sessionHistory.copyFrom(sessionHistory)
+    commandsManager.timestamp = timestamp
+    commandsManager.commandId = commandId
+    commandsManager.streamId = streamId
+    commandsManager.host = host
+    commandsManager.port = port
+    commandsManager.appName = appName
+    commandsManager.streamName = streamName
+    commandsManager.tcUrl = tcUrl
+    commandsManager.flashVersion = flashVersion
+    commandsManager.user = user
+    commandsManager.password = password
+    commandsManager.onAuth = onAuth
+    commandsManager.startTs = startTs
+    commandsManager.config.readChunkSize = config.readChunkSize
+    commandsManager.config.writeChunkSize = config.writeChunkSize
+    commandsManager.config.acknowledgementWindowSize = config.acknowledgementWindowSize
+    commandsManager.audioDisabled = audioDisabled
+    commandsManager.videoDisabled = videoDisabled
+    commandsManager.customAmfObject = customAmfObject
+    commandsManager.customMetadata = customMetadata
+    commandsManager.bytesRead = bytesRead
+    commandsManager.lastAcknowledgementSequence = lastAcknowledgementSequence
+    commandsManager.width = width
+    commandsManager.height = height
+    commandsManager.fps = fps
+    commandsManager.sampleRate = sampleRate
+    commandsManager.isStereo = isStereo
+    commandsManager.videoCodec = videoCodec
+    commandsManager.audioCodec = audioCodec
+    return commandsManager
+  }
 }
diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/CommandsManagerAmf3.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/CommandsManagerAmf3.kt
index 6c6fa5b934..6ea81632c2 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/CommandsManagerAmf3.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/CommandsManagerAmf3.kt
@@ -17,44 +17,53 @@
 package com.pedro.rtmp.rtmp
 
 import android.util.Log
+import com.pedro.common.AudioCodec
 import com.pedro.common.VideoCodec
-import com.pedro.rtmp.amf.v3.Amf3Array
-import com.pedro.rtmp.amf.v3.Amf3Data
-import com.pedro.rtmp.amf.v3.Amf3Dictionary
-import com.pedro.rtmp.amf.v3.Amf3Null
-import com.pedro.rtmp.amf.v3.Amf3Object
+import com.pedro.rtmp.amf.v0.AmfData
+import com.pedro.rtmp.amf.v0.AmfEcmaArray
+import com.pedro.rtmp.amf.v0.AmfNull
+import com.pedro.rtmp.amf.v0.AmfObject
+import com.pedro.rtmp.amf.v0.AmfStrictArray
+import com.pedro.rtmp.amf.v0.AmfString
 import com.pedro.rtmp.amf.v3.Amf3String
 import com.pedro.rtmp.flv.audio.AudioFormat
 import com.pedro.rtmp.flv.video.VideoFormat
 import com.pedro.rtmp.rtmp.chunk.ChunkStreamId
 import com.pedro.rtmp.rtmp.chunk.ChunkType
 import com.pedro.rtmp.rtmp.message.BasicHeader
+import com.pedro.rtmp.rtmp.message.command.CommandAmf0
 import com.pedro.rtmp.rtmp.message.command.CommandAmf3
-import com.pedro.rtmp.rtmp.message.data.DataAmf3
+import com.pedro.rtmp.rtmp.message.data.DataAmf0
 import com.pedro.rtmp.utils.socket.RtmpSocket
 
 class CommandsManagerAmf3: CommandsManager() {
   override suspend fun sendConnectImp(auth: String, socket: RtmpSocket) {
-    val connect = CommandAmf3("connect", ++commandId, getCurrentTimestamp(), streamId,
-        BasicHeader(ChunkType.TYPE_0, ChunkStreamId.OVER_CONNECTION.mark))
-    val connectInfo = Amf3Object()
+    val connect = CommandAmf0(
+      "connect", ++commandId, getCurrentTimestamp(), streamId,
+      BasicHeader(ChunkType.TYPE_0, ChunkStreamId.OVER_CONNECTION.mark)
+    )
+    val connectInfo = AmfObject()
     connectInfo.setProperty("app", appName + auth)
     connectInfo.setProperty("flashVer", flashVersion)
     connectInfo.setProperty("tcUrl", tcUrl + auth)
+    val list = mutableListOf<AmfData>()
     if (!videoDisabled) {
-      if (videoCodec == VideoCodec.H265) {
-        val list = mutableListOf<Amf3Data>()
-        list.add(Amf3String("hvc1"))
-        val array = Amf3Array(list)
-        connectInfo.setProperty("fourCcList", array)
-      } else if (videoCodec == VideoCodec.AV1) {
-        val list = mutableListOf<Amf3Data>()
-        list.add(Amf3String("av01"))
-        val array = Amf3Array(list)
-        connectInfo.setProperty("fourCcList", array)
-      }
+      if (videoCodec == VideoCodec.H265) list.add(AmfString("hvc1"))
+      else if (videoCodec == VideoCodec.AV1) list.add(AmfString("av01"))
+    }
+    if (!audioDisabled) {
+      if (audioCodec == AudioCodec.OPUS) list.add(AmfString("Opus"))
+    }
+    if (list.isNotEmpty()) {
+      val array = AmfStrictArray(list)
+      connectInfo.setProperty("fourCcList", array)
     }
     connectInfo.setProperty("objectEncoding", 3.0)
+
+    // Inject other custom AMF fields as-is
+    customAmfObject.forEach { (key, value) ->
+      connectInfo.setProperty(key, value)
+    }
     connect.addData(connectInfo)
 
     connect.writeHeader(socket)
@@ -66,7 +75,7 @@ class CommandsManagerAmf3: CommandsManager() {
   override suspend fun createStreamImp(socket: RtmpSocket) {
     val releaseStream = CommandAmf3("releaseStream", ++commandId, getCurrentTimestamp(), streamId,
         BasicHeader(ChunkType.TYPE_0, ChunkStreamId.OVER_STREAM.mark))
-    releaseStream.addData(Amf3Null())
+    releaseStream.addData(AmfNull())
     releaseStream.addData(Amf3String(streamName))
 
     releaseStream.writeHeader(socket)
@@ -76,7 +85,7 @@ class CommandsManagerAmf3: CommandsManager() {
 
     val fcPublish = CommandAmf3("FCPublish", ++commandId, getCurrentTimestamp(), streamId,
         BasicHeader(ChunkType.TYPE_0, ChunkStreamId.OVER_STREAM.mark))
-    fcPublish.addData(Amf3Null())
+    fcPublish.addData(AmfNull())
     fcPublish.addData(Amf3String(streamName))
 
     fcPublish.writeHeader(socket)
@@ -86,7 +95,7 @@ class CommandsManagerAmf3: CommandsManager() {
 
     val createStream = CommandAmf3("createStream", ++commandId, getCurrentTimestamp(), streamId,
         BasicHeader(ChunkType.TYPE_0, ChunkStreamId.OVER_CONNECTION.mark))
-    createStream.addData(Amf3Null())
+    createStream.addData(AmfNull())
 
     createStream.writeHeader(socket)
     createStream.writeBody(socket, config.writeChunkSize)
@@ -96,22 +105,30 @@ class CommandsManagerAmf3: CommandsManager() {
 
   override suspend fun sendMetadataImp(socket: RtmpSocket) {
     val name = "@setDataFrame"
-    val metadata = DataAmf3(name, getCurrentTimestamp(), streamId)
-    metadata.addData(Amf3String("onMetaData"))
-    val amfEcmaArray = Amf3Dictionary()
+    val metadata = DataAmf0(name, getCurrentTimestamp(), streamId)
+    metadata.addData(AmfString("onMetaData"))
+    val amfEcmaArray = AmfEcmaArray()
     amfEcmaArray.setProperty("duration", 0.0)
     if (!videoDisabled) {
       amfEcmaArray.setProperty("width", width.toDouble())
       amfEcmaArray.setProperty("height", height.toDouble())
       //few servers don't support it even if it is in the standard rtmp enhanced
-      //val codecValue = if (videoCodec == VideoCodec.H265) VideoFormat.HEVC.value else VideoFormat.AVC.value
-      //amfEcmaArray.setProperty("videocodecid", codecValue.toDouble())
-      amfEcmaArray.setProperty("videocodecid", VideoFormat.AVC.value.toDouble())
+      val codecValue = when (videoCodec) {
+        VideoCodec.H264 -> VideoFormat.AVC.value
+        VideoCodec.H265 -> VideoFormat.HEVC.value
+        VideoCodec.AV1 -> VideoFormat.AV1.value
+      }
+      amfEcmaArray.setProperty("videocodecid", codecValue.toDouble())
       amfEcmaArray.setProperty("framerate", fps.toDouble())
       amfEcmaArray.setProperty("videodatarate", 0.0)
     }
     if (!audioDisabled) {
-      amfEcmaArray.setProperty("audiocodecid", AudioFormat.AAC.value.toDouble())
+      val codecValue = when (audioCodec) {
+        AudioCodec.G711 -> AudioFormat.G711_A.value
+        AudioCodec.AAC -> AudioFormat.AAC.value
+        AudioCodec.OPUS -> AudioFormat.OPUS.value
+      }
+      amfEcmaArray.setProperty("audiocodecid", codecValue.toDouble())
       amfEcmaArray.setProperty("audiosamplerate", sampleRate.toDouble())
       amfEcmaArray.setProperty("audiosamplesize", 16.0)
       amfEcmaArray.setProperty("audiodatarate", 0.0)
@@ -132,7 +149,7 @@ class CommandsManagerAmf3: CommandsManager() {
     val name = "publish"
     val publish = CommandAmf3(name, ++commandId, getCurrentTimestamp(), streamId,
         BasicHeader(ChunkType.TYPE_0, ChunkStreamId.OVER_STREAM.mark))
-    publish.addData(Amf3Null())
+    publish.addData(AmfNull())
     publish.addData(Amf3String(streamName))
     publish.addData(Amf3String("live"))
 
@@ -145,7 +162,7 @@ class CommandsManagerAmf3: CommandsManager() {
   override suspend fun sendCloseImp(socket: RtmpSocket) {
     val name = "closeStream"
     val closeStream = CommandAmf3(name, ++commandId, getCurrentTimestamp(), streamId, BasicHeader(ChunkType.TYPE_0, ChunkStreamId.OVER_STREAM.mark))
-    closeStream.addData(Amf3Null())
+    closeStream.addData(AmfNull())
 
     closeStream.writeHeader(socket)
     closeStream.writeBody(socket, config.writeChunkSize)
diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt
index 50263c4c99..d4135d1aaa 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt
@@ -43,7 +43,6 @@ import com.pedro.rtmp.rtmp.message.command.Command
 import com.pedro.rtmp.rtmp.message.control.Type
 import com.pedro.rtmp.rtmp.message.control.UserControl
 import com.pedro.rtmp.utils.AuthUtil
-import com.pedro.rtmp.utils.RtmpConfig
 import com.pedro.rtmp.utils.socket.RtmpSocket
 import com.pedro.rtmp.utils.socket.TcpSocket
 import com.pedro.rtmp.utils.socket.TcpTunneledSocket
@@ -142,8 +141,9 @@ class RtmpClient(private val connectChecker: ConnectChecker) {
     if (!isStreaming) {
       commandsManager = when (amfVersion) {
         AmfVersion.VERSION_0 -> CommandsManagerAmf0()
-        AmfVersion.VERSION_3 -> TODO("Not yet implemented")
+        AmfVersion.VERSION_3 -> CommandsManagerAmf3()
       }
+      rtmpSender.commandsManager = commandsManager
     }
   }
 
@@ -437,6 +437,14 @@ class RtmpClient(private val connectChecker: ConnectChecker) {
           "_result" -> {
             when (commandName) {
               "connect" -> {
+                if (commandsManager is CommandsManagerAmf3) {
+                  val objectEncoding = command.getObjectEncoding()
+                  if (objectEncoding != 3) {
+                    Log.e(TAG, "Server doesn't support AMF3, falling to AMF0")
+                    commandsManager = commandsManager.copyInto(CommandsManagerAmf0())
+                    rtmpSender.commandsManager = commandsManager
+                  }
+                }
                 if (commandsManager.onAuth) {
                   onMainThread { connectChecker.onAuthSuccess() }
                   commandsManager.onAuth = false
diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpSender.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpSender.kt
index c6297835b4..49996878c7 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpSender.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpSender.kt
@@ -43,7 +43,7 @@ import java.nio.ByteBuffer
  */
 class RtmpSender(
   connectChecker: ConnectChecker,
-  private val commandsManager: CommandsManager
+  var commandsManager: CommandsManager
 ): BaseSender(connectChecker, "RtmpSender") {
 
   private var audioPacket: BasePacket = AacPacket()
diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/Command.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/Command.kt
index e5c192c175..5671a15db9 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/Command.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/Command.kt
@@ -38,6 +38,7 @@ abstract class Command(var name: String = "", var commandId: Int, private val ti
   }
 
   abstract fun getStreamId(): Int
+  abstract fun getObjectEncoding(): Int
   abstract fun getDescription(): String
   abstract fun getCode(): String
 
diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/CommandAmf0.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/CommandAmf0.kt
index 8baf010f43..380403d397 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/CommandAmf0.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/CommandAmf0.kt
@@ -20,28 +20,36 @@ import com.pedro.rtmp.amf.v0.AmfData
 import com.pedro.rtmp.amf.v0.AmfNumber
 import com.pedro.rtmp.amf.v0.AmfObject
 import com.pedro.rtmp.amf.v0.AmfString
+import com.pedro.rtmp.amf.v3.Amf3Data
+import com.pedro.rtmp.amf.v3.Amf3Double
+import com.pedro.rtmp.amf.v3.Amf3Integer
+import com.pedro.rtmp.amf.v3.Amf3Object
+import com.pedro.rtmp.amf.v3.Amf3String
 import com.pedro.rtmp.rtmp.chunk.ChunkStreamId
 import com.pedro.rtmp.rtmp.chunk.ChunkType
 import com.pedro.rtmp.rtmp.message.BasicHeader
 import com.pedro.rtmp.rtmp.message.MessageType
 import java.io.ByteArrayOutputStream
+import java.io.IOException
 import java.io.InputStream
 
 /**
  * Created by pedro on 21/04/21.
+ *
+ * In AMF3 sessions servers like Red5 keep replying with this message type but encode
+ * arguments as AMF3 values escaped with the 0x11 avmplus marker, so reading must accept
+ * both encodings per value.
  */
 class CommandAmf0(name: String = "", commandId: Int = 0, private val timestamp: Int = 0, private val streamId: Int = 0, basicHeader: BasicHeader =
     BasicHeader(ChunkType.TYPE_0, ChunkStreamId.OVER_CONNECTION.mark)): Command(name, commandId, timestamp, streamId, basicHeader = basicHeader) {
 
-  private val data: MutableList<AmfData> = mutableListOf()
+  private val data: MutableList<Any> = mutableListOf()
 
   init {
     val amfString = AmfString(name)
-    data.add(amfString)
     bodySize += amfString.getSize() + 1
     val amfNumber = AmfNumber(commandId.toDouble())
     bodySize += amfNumber.getSize() + 1
-    data.add(amfNumber)
     header.messageLength = bodySize
   }
 
@@ -52,42 +60,104 @@ class CommandAmf0(name: String = "", commandId: Int = 0, private val timestamp:
   }
 
   override fun getStreamId(): Int {
-    return (data[3] as AmfNumber).value.toInt()
+    return when (val value = data[1]) {
+      is AmfNumber -> value.value.toInt()
+      is Amf3Double -> value.value.toInt()
+      is Amf3Integer -> value.value
+      else -> throw IOException("Unexpected stream id type: $value")
+    }
+  }
+
+  override fun getObjectEncoding(): Int {
+    return when (val value = getInfoProperty("objectEncoding")) {
+      is AmfNumber -> value.value.toInt()
+      is Amf3Double -> value.value.toInt()
+      is Amf3Integer -> value.value
+      else -> 0
+    }
   }
 
   override fun getDescription(): String {
-    return ((data[3] as AmfObject).getProperty("description") as AmfString).value
+    return when (val value = getInfoProperty("description")) {
+      is AmfString -> value.value
+      is Amf3String -> value.value
+      else -> throw IOException("Unexpected description type: $value")
+    }
   }
 
   override fun getCode(): String {
-    return ((data[3] as AmfObject).getProperty("code") as AmfString).value
+    return when (val value = getInfoProperty("code")) {
+      is AmfString -> value.value
+      is Amf3String -> value.value
+      else -> throw IOException("Unexpected code type: $value")
+    }
+  }
+
+  private fun getInfoProperty(name: String): Any? {
+    return when (val info = data.getOrNull(1)) {
+      is AmfObject -> info.getProperty(name)
+      is Amf3Object -> info.getProperty(name)
+      else -> null
+    }
   }
 
   override fun readBody(input: InputStream) {
     data.clear()
-    var bytesRead = 0
-    while (bytesRead < header.messageLength) {
-      val amfData = AmfData.getAmfData(input)
-      bytesRead += amfData.getSize() + 1
-      data.add(amfData)
+    bodySize = 0
+
+    //Red5 in AMF3 sessions escapes every value including command name and transaction id
+    this.name = when (val value = readMixedValue(input)) {
+      is AmfString -> value.value
+      is Amf3String -> value.value
+      else -> throw IOException("Unexpected command name type: $value")
     }
-    if (data.isNotEmpty()) {
-      if (data[0] is AmfString) {
-        name = (data[0] as AmfString).value
-      }
-      if (data.size >= 2 && data[1] is AmfNumber) {
-        commandId = (data[1] as AmfNumber).value.toInt()
-      }
+    this.commandId = when (val value = readMixedValue(input)) {
+      is AmfNumber -> value.value.toInt()
+      is Amf3Integer -> value.value
+      is Amf3Double -> value.value.toInt()
+      else -> throw IOException("Unexpected transaction id type: $value")
+    }
+    while (bodySize < header.messageLength) {
+      data.add(readMixedValue(input))
     }
-    bodySize = bytesRead
     header.messageLength = bodySize
   }
 
+  private fun readMixedValue(input: InputStream): Any {
+    val mark = input.read()
+    bodySize += 1
+    return if (mark == 0x11) {
+      val amf3Data = Amf3Data.getAmf3Data(input)
+      bodySize += amf3Data.getSize() + 1
+      amf3Data
+    } else {
+      val amfData = AmfData.getAmfData(mark, input)
+      bodySize += amfData.getSize()
+      amfData
+    }
+  }
+
   override fun storeBody(): ByteArray {
     val byteArrayOutputStream = ByteArrayOutputStream()
+    val amfString = AmfString(name)
+    amfString.writeHeader(byteArrayOutputStream)
+    amfString.writeBody(byteArrayOutputStream)
+    val amfNumber = AmfNumber(commandId.toDouble())
+    amfNumber.writeHeader(byteArrayOutputStream)
+    amfNumber.writeBody(byteArrayOutputStream)
+
     data.forEach {
-      it.writeHeader(byteArrayOutputStream)
-      it.writeBody(byteArrayOutputStream)
+      when (it) {
+        is AmfData -> {
+          it.writeHeader(byteArrayOutputStream)
+          it.writeBody(byteArrayOutputStream)
+        }
+        is Amf3Data -> {
+          byteArrayOutputStream.write(0x11)
+          it.writeHeader(byteArrayOutputStream)
+          it.writeBody(byteArrayOutputStream)
+        }
+      }
     }
     return byteArrayOutputStream.toByteArray()
   }
diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/CommandAmf3.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/CommandAmf3.kt
index bb919b3ef6..8a3099e33c 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/CommandAmf3.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/command/CommandAmf3.kt
@@ -16,6 +16,10 @@
 
 package com.pedro.rtmp.rtmp.message.command
 
+import com.pedro.rtmp.amf.v0.AmfData
+import com.pedro.rtmp.amf.v0.AmfNumber
+import com.pedro.rtmp.amf.v0.AmfObject
+import com.pedro.rtmp.amf.v0.AmfString
 import com.pedro.rtmp.amf.v3.Amf3Data
 import com.pedro.rtmp.amf.v3.Amf3Double
 import com.pedro.rtmp.amf.v3.Amf3Object
@@ -33,59 +37,119 @@ import java.io.InputStream
 class CommandAmf3(name: String = "", commandId: Int = 0, private val timestamp: Int = 0, private val streamId: Int = 0, basicHeader: BasicHeader =
     BasicHeader(ChunkType.TYPE_0, ChunkStreamId.OVER_CONNECTION.mark)): Command(name, commandId, timestamp, streamId, basicHeader = basicHeader) {
 
-  private val data: MutableList<Amf3Data> = mutableListOf()
+  private val dataAmf0: MutableList<AmfData> = mutableListOf()
+  private val dataAmf3: MutableList<Amf3Data> = mutableListOf()
+  private var isAmf3 = true
+  private var infoIndex = 0
 
   init {
-    val amf3String = Amf3String(name)
-    data.add(amf3String)
-    bodySize += amf3String.getSize() + 1
-    val amf3Double = Amf3Double(commandId.toDouble())
-    bodySize += amf3Double.getSize() + 1
-    data.add(amf3Double)
+    val amfString = AmfString(name)
+    bodySize += amfString.getSize() + 1
+    val amfNumber = AmfNumber(commandId.toDouble())
+    bodySize += amfNumber.getSize() + 1
+    bodySize += 1
     header.messageLength = bodySize
   }
 
   fun addData(amf3Data: Amf3Data) {
-    data.add(amf3Data)
-    bodySize += amf3Data.getSize() + 1
+    dataAmf3.add(amf3Data)
+    bodySize += amf3Data.getSize() + 2
     header.messageLength = bodySize
   }
 
+  fun addData(amfData: AmfData) {
+    dataAmf0.add(amfData)
+    bodySize += amfData.getSize() + 1
+    header.messageLength = bodySize
+  }
+
+  override fun getObjectEncoding(): Int {
+    return if (isAmf3) {
+      ((dataAmf3[infoIndex] as? Amf3Object)?.getProperty("objectEncoding") as? Amf3Double)?.value?.toInt() ?: 0
+    } else {
+      ((dataAmf0[infoIndex] as? AmfObject)?.getProperty("objectEncoding") as? AmfNumber)?.value?.toInt() ?: 0
+    }
+  }
+
   override fun getStreamId(): Int {
-    return (data[3] as Amf3Double).value.toInt()
+    return if (isAmf3) {
+      (dataAmf3[infoIndex] as Amf3Double).value.toInt()
+    } else {
+      (dataAmf0[infoIndex] as AmfNumber).value.toInt()
+    }
   }
 
   override fun getDescription(): String {
-    return ((data[3] as Amf3Object).getProperty("description") as Amf3String).value
+    return if (isAmf3) {
+      ((dataAmf3[infoIndex] as Amf3Object).getProperty("description") as Amf3String).value
+    } else {
+      ((dataAmf0[infoIndex] as AmfObject).getProperty("description") as AmfString).value
+    }
   }
 
   override fun getCode(): String {
-    return ((data[3] as Amf3Object).getProperty("code") as Amf3String).value
+    return if (isAmf3) {
+      ((dataAmf3[infoIndex] as Amf3Object).getProperty("code") as Amf3String).value
+    } else {
+      ((dataAmf0[infoIndex] as AmfObject).getProperty("code") as AmfString).value
+    }
   }
 
   override fun readBody(input: InputStream) {
-    data.clear()
-    var bytesRead = 0
-    while (bytesRead < header.messageLength) {
-      val amf3Data = Amf3Data.getAmf3Data(input)
-      bytesRead += amf3Data.getSize() + 1
-      data.add(amf3Data)
-    }
-    if (data.isNotEmpty()) {
-      if (data[0] is Amf3String) {
-        name = (data[0] as Amf3String).value
-      }
-      if (data.size >= 2 && data[1] is Amf3Double) {
-        commandId = (data[1] as Amf3Double).value.toInt()
+    dataAmf0.clear()
+    dataAmf3.clear()
+    bodySize = 0
+
+    input.read()
+    bodySize += 1
+    val name = AmfData.getAmfData(input) as AmfString
+    bodySize += name.getSize() + 1
+    this.name = name.value
+    val commandId = AmfData.getAmfData(input) as AmfNumber
+    bodySize += commandId.getSize() + 1
+    this.commandId = commandId.value.toInt()
+
+    var cont = 0
+    while (bodySize < header.messageLength) {
+      val mark = input.read()
+      bodySize += 1
+      if (mark == 0x11) {
+        val amf3Data = Amf3Data.getAmf3Data(input)
+        dataAmf3.add(amf3Data)
+        bodySize += amf3Data.getSize() + 1
+        if (cont == 1) {
+          isAmf3 = true
+          infoIndex = dataAmf3.size - 1
+        }
+      } else {
+        val amfData = AmfData.getAmfData(mark, input)
+        dataAmf0.add(amfData)
+        bodySize += amfData.getSize()
+        if (cont == 1) {
+          isAmf3 = false
+          infoIndex = dataAmf0.size - 1
+        }
       }
+      cont++
     }
-    bodySize = bytesRead
     header.messageLength = bodySize
   }
 
   override fun storeBody(): ByteArray {
     val byteArrayOutputStream = ByteArrayOutputStream()
-    data.forEach {
+    byteArrayOutputStream.write(0x00)
+    val amfString = AmfString(name)
+    amfString.writeHeader(byteArrayOutputStream)
+    amfString.writeBody(byteArrayOutputStream)
+    val amfNumber = AmfNumber(commandId.toDouble())
+    amfNumber.writeHeader(byteArrayOutputStream)
+    amfNumber.writeBody(byteArrayOutputStream)
+    dataAmf0.forEach {
+      it.writeHeader(byteArrayOutputStream)
+      it.writeBody(byteArrayOutputStream)
+    }
+    dataAmf3.forEach {
+      byteArrayOutputStream.write(0x11)
       it.writeHeader(byteArrayOutputStream)
       it.writeBody(byteArrayOutputStream)
     }
@@ -95,6 +159,6 @@ class CommandAmf3(name: String = "", commandId: Int = 0, private val timestamp:
   override fun getType(): MessageType = MessageType.COMMAND_AMF3
 
   override fun toString(): String {
-    return "Command(name='$name', transactionId=$commandId, timeStamp=$timestamp, streamId=$streamId, data=$data, bodySize=$bodySize)"
+    return "Command(name='$name', transactionId=$commandId, timeStamp=$timestamp, streamId=$streamId, data=${dataAmf0.plus(dataAmf3)}, bodySize=$bodySize)"
   }
 }
\ No newline at end of file
diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/data/DataAmf0.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/data/DataAmf0.kt
index 8b8e14b687..c70328c72c 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/data/DataAmf0.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/data/DataAmf0.kt
@@ -36,9 +36,6 @@ class DataAmf0(private var name: String = "", timeStamp: Int = 0, streamId: Int
   init {
     val amfString = AmfString(name)
     bodySize += amfString.getSize() + 1
-    data.forEach {
-      bodySize += it.getSize() + 1
-    }
     header.messageLength = bodySize
   }
 
diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/data/DataAmf3.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/data/DataAmf3.kt
index e2ef45309f..d425bd491e 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/data/DataAmf3.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/message/data/DataAmf3.kt
@@ -16,8 +16,9 @@
 
 package com.pedro.rtmp.rtmp.message.data
 
+import com.pedro.rtmp.amf.v0.AmfData
+import com.pedro.rtmp.amf.v0.AmfString
 import com.pedro.rtmp.amf.v3.Amf3Data
-import com.pedro.rtmp.amf.v3.Amf3String
 import com.pedro.rtmp.rtmp.chunk.ChunkStreamId
 import com.pedro.rtmp.rtmp.chunk.ChunkType
 import com.pedro.rtmp.rtmp.message.BasicHeader
@@ -28,46 +29,62 @@ import java.io.InputStream
 /**
  * Created by pedro on 21/04/21.
  */
-class DataAmf3(private val name: String = "", timeStamp: Int = 0, streamId: Int = 0, basicHeader: BasicHeader = BasicHeader(ChunkType.TYPE_0, ChunkStreamId.OVER_CONNECTION.mark)):
+class DataAmf3(private var name: String = "", timeStamp: Int = 0, streamId: Int = 0, basicHeader: BasicHeader = BasicHeader(ChunkType.TYPE_0, ChunkStreamId.OVER_CONNECTION.mark)):
   Data(timeStamp, streamId, basicHeader) {
 
-  private val data: MutableList<Amf3Data> = mutableListOf()
+  private val dataAmf0: MutableList<AmfData> = mutableListOf()
+  private val dataAmf3: MutableList<Amf3Data> = mutableListOf()
 
   init {
-    val amf3String = Amf3String(name)
-    bodySize += amf3String.getSize() + 1
-    data.forEach {
-      bodySize += it.getSize() + 1
-    }
+    val amfString = AmfString(name)
+    bodySize += amfString.getSize() + 1
+    bodySize += 1
     header.messageLength = bodySize
   }
 
   fun addData(amf3Data: Amf3Data) {
-    data.add(amf3Data)
-    bodySize += amf3Data.getSize() + 1
+    dataAmf3.add(amf3Data)
+    bodySize += amf3Data.getSize() + 2
     header.messageLength = bodySize
   }
 
   override fun readBody(input: InputStream) {
-    data.clear()
+    dataAmf0.clear()
+    dataAmf3.clear()
     bodySize = 0
-    val amf3String = Amf3String()
-    amf3String.readHeader(input)
-    amf3String.readBody(input)
-    bodySize += amf3String.getSize() + 1
+
+    input.read()
+    bodySize += 1
+    val amfString = AmfString()
+    amfString.readHeader(input)
+    amfString.readBody(input)
+    bodySize += amfString.getSize() + 1
+    name = amfString.value
+
     while (bodySize < header.messageLength) {
-      val amf3Data = Amf3Data.getAmf3Data(input)
-      data.add(amf3Data)
-      bodySize += amf3Data.getSize() + 1
+      val mark = input.read()
+      bodySize += 1
+      if (mark == 0x11) {
+        val amf3Data = Amf3Data.getAmf3Data(input)
+        dataAmf3.add(amf3Data)
+        bodySize += amf3Data.getSize() + 1
+      } else {
+        val amfData = AmfData.getAmfData(mark, input)
+        dataAmf0.add(amfData)
+        bodySize += amfData.getSize()
+      }
     }
   }
 
   override fun storeBody(): ByteArray {
     val byteArrayOutputStream = ByteArrayOutputStream()
-    val amf3String = Amf3String(name)
-    amf3String.writeHeader(byteArrayOutputStream)
-    amf3String.writeBody(byteArrayOutputStream)
-    data.forEach {
+    byteArrayOutputStream.write(0x00)
+    val amfString = AmfString(name)
+    amfString.writeHeader(byteArrayOutputStream)
+    amfString.writeBody(byteArrayOutputStream)
+
+    dataAmf3.forEach {
+      byteArrayOutputStream.write(0x11)
       it.writeHeader(byteArrayOutputStream)
       it.writeBody(byteArrayOutputStream)
     }
@@ -77,6 +94,6 @@ class DataAmf3(private val name: String = "", timeStamp: Int = 0, streamId: Int
   override fun getType(): MessageType = MessageType.DATA_AMF3
 
   override fun toString(): String {
-    return "Data(name='$name', data=$data, bodySize=$bodySize)"
+    return "Data(name='$name', data=$dataAmf3, bodySize=$bodySize)"
   }
 }
\ No newline at end of file
diff --git a/rtmp/src/main/java/com/pedro/rtmp/utils/CommandSessionHistory.kt b/rtmp/src/main/java/com/pedro/rtmp/utils/CommandSessionHistory.kt
index d124b9d1bc..7a3a2ef724 100644
--- a/rtmp/src/main/java/com/pedro/rtmp/utils/CommandSessionHistory.kt
+++ b/rtmp/src/main/java/com/pedro/rtmp/utils/CommandSessionHistory.kt
@@ -46,4 +46,11 @@ class CommandSessionHistory {
     commandHistory.clear()
     headerHistory.clear()
   }
+
+  fun copyFrom(other: CommandSessionHistory) {
+    commandHistory.clear()
+    headerHistory.clear()
+    commandHistory.putAll(other.commandHistory)
+    headerHistory.putAll(other.headerHistory)
+  }
 }
\ No newline at end of file
diff --git a/rtmp/src/test/java/com/pedro/rtmp/Amf3ConnectionTest.kt b/rtmp/src/test/java/com/pedro/rtmp/Amf3ConnectionTest.kt
new file mode 100644
index 0000000000..e5e6800995
--- /dev/null
+++ b/rtmp/src/test/java/com/pedro/rtmp/Amf3ConnectionTest.kt
@@ -0,0 +1,95 @@
+/*
+ * Copyright (C) 2026 pedroSG94.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.pedro.rtmp
+
+import com.pedro.common.ConnectChecker
+import com.pedro.rtmp.amf.AmfVersion
+import com.pedro.rtmp.rtmp.CommandsManagerAmf0
+import com.pedro.rtmp.rtmp.CommandsManagerAmf3
+import com.pedro.rtmp.rtmp.RtmpClient
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.setMain
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Assume.assumeTrue
+import org.junit.Test
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+/**
+ * Integration tests against real RTMP servers, skipped unless the env var is set:
+ *
+ * AMF3 session (server with AMF3 support, e.g. Red5):
+ * RTMP_AMF3_URL=rtmp://localhost/live/test ./gradlew :rtmp:testDebugUnitTest --tests "*Amf3ConnectionTest*"
+ *
+ * AMF0 fallback (server without AMF3, e.g. MediaMTX):
+ * RTMP_AMF3_FALLBACK_URL=rtmp://localhost:1936/live/test ./gradlew :rtmp:testDebugUnitTest --tests "*Amf3ConnectionTest*"
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class Amf3ConnectionTest {
+
+  @Test
+  fun `amf3 session against a server with amf3 support`() {
+    val url = System.getenv("RTMP_AMF3_URL")
+    assumeTrue("skipped: RTMP_AMF3_URL not set", url != null)
+    val client = connectAndAwaitPublish(url ?: "")
+    assertTrue("server negotiated amf3, manager should still be amf3",
+      getCommandsManager(client) is CommandsManagerAmf3)
+    client.disconnect()
+  }
+
+  @Test
+  fun `amf0 fallback against a server without amf3 support`() {
+    val url = System.getenv("RTMP_AMF3_FALLBACK_URL")
+    assumeTrue("skipped: RTMP_AMF3_FALLBACK_URL not set", url != null)
+    val client = connectAndAwaitPublish(url ?: "")
+    assertTrue("server without amf3, manager should have fallen back to amf0",
+      getCommandsManager(client) is CommandsManagerAmf0)
+    client.disconnect()
+  }
+
+  private fun connectAndAwaitPublish(url: String): RtmpClient {
+    Dispatchers.setMain(Dispatchers.Default)
+    val latch = CountDownLatch(1)
+    var failedReason: String? = null
+    val client = RtmpClient(object : ConnectChecker {
+      override fun onConnectionStarted(url: String) {}
+      override fun onConnectionSuccess() { latch.countDown() }
+      override fun onConnectionFailed(reason: String) {
+        failedReason = reason
+        latch.countDown()
+      }
+      override fun onDisconnect() {}
+      override fun onAuthError() {}
+      override fun onAuthSuccess() {}
+    })
+    client.setAmfVersion(AmfVersion.VERSION_3)
+    client.connect(url)
+    val responded = latch.await(10, TimeUnit.SECONDS)
+    if (!responded) client.disconnect()
+    assertTrue("no connection result in 10s", responded)
+    assertNull("connection failed: $failedReason", failedReason)
+    return client
+  }
+
+  private fun getCommandsManager(client: RtmpClient): Any {
+    val field = RtmpClient::class.java.getDeclaredField("commandsManager")
+    field.isAccessible = true
+    return field.get(client)
+  }
+}
diff --git a/rtmp/src/test/java/com/pedro/rtmp/amf/v3/Amf3DataTest.kt b/rtmp/src/test/java/com/pedro/rtmp/amf/v3/Amf3DataTest.kt
new file mode 100644
index 0000000000..f10cd43ac6
--- /dev/null
+++ b/rtmp/src/test/java/com/pedro/rtmp/amf/v3/Amf3DataTest.kt
@@ -0,0 +1,156 @@
+/*
+ * Copyright (C) 2026 pedroSG94.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.pedro.rtmp.amf.v3
+
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertArrayEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
+
+class Amf3DataTest {
+
+  private fun write(data: Amf3Data): ByteArray {
+    val output = ByteArrayOutputStream()
+    data.writeHeader(output)
+    data.writeBody(output)
+    return output.toByteArray()
+  }
+
+  private fun read(bytes: ByteArray): Amf3Data = Amf3Data.getAmf3Data(ByteArrayInputStream(bytes))
+
+  @Test
+  fun `object writes anonymous dynamic format`() {
+    val obj = Amf3Object()
+    obj.setProperty("name", "pedro")
+    val expected = byteArrayOf(
+      0x0A, 0x0B, 0x01, //object marker, U29O-traits dynamic 0 sealed, empty class name
+      0x09, 'n'.code.toByte(), 'a'.code.toByte(), 'm'.code.toByte(), 'e'.code.toByte(),
+      0x06, 0x0B, 'p'.code.toByte(), 'e'.code.toByte(), 'd'.code.toByte(), 'r'.code.toByte(), 'o'.code.toByte(),
+      0x01 //end of dynamic members
+    )
+    val bytes = write(obj)
+    assertArrayEquals(expected, bytes)
+    assertEquals(bytes.size - 1, obj.getSize())
+  }
+
+  @Test
+  fun `object round trip with nested values`() {
+    val obj = Amf3Object()
+    obj.setProperty("code", "NetConnection.Connect.Success")
+    obj.setProperty("objectEncoding", 3.0)
+    obj.setProperty("count", 500)
+    obj.setProperty("enabled", true)
+    obj.setProperty("empty")
+    val nested = Amf3Object()
+    nested.setProperty("inner", "value")
+    obj.setProperty("data", nested)
+
+    val bytes = write(obj)
+    assertEquals(bytes.size - 1, obj.getSize())
+    val result = read(bytes) as Amf3Object
+    assertEquals(bytes.size - 1, result.getSize())
+    assertEquals("NetConnection.Connect.Success", (result.getProperty("code") as Amf3String).value)
+    assertEquals(3.0, (result.getProperty("objectEncoding") as Amf3Double).value, 0.0)
+    assertEquals(500, (result.getProperty("count") as Amf3Integer).value)
+    assertTrue(result.getProperty("enabled") is Amf3True)
+    assertTrue(result.getProperty("empty") is Amf3Null)
+    assertEquals("value", ((result.getProperty("data") as Amf3Object).getProperty("inner") as Amf3String).value)
+  }
+
+  @Test
+  fun `object read sealed members with class name`() {
+    //typed object: 1 sealed member "level" of class "org.Test", value "status"
+    val output = ByteArrayOutputStream()
+    output.write(0x0A)
+    output.write(0x13) //U29O-traits: 1 sealed, not dynamic (0b10011)
+    Amf3String("org.Test").writeBody(output)
+    Amf3String("level").writeBody(output)
+    val value = Amf3String("status")
+    value.writeHeader(output)
+    value.writeBody(output)
+
+    val bytes = output.toByteArray()
+    val result = read(bytes) as Amf3Object
+    assertEquals("org.Test", result.className)
+    assertEquals("status", (result.getProperty("level") as Amf3String).value)
+    assertEquals(bytes.size - 1, result.getSize())
+  }
+
+  @Test
+  fun `object reference is read without consuming extra bytes`() {
+    val result = read(byteArrayOf(0x0A, 0x00)) as Amf3Object
+    assertEquals(1, result.getSize())
+  }
+
+  @Test
+  fun `array round trip with dense and associative parts`() {
+    val array = Amf3Array(mutableListOf(Amf3String("hvc1"), Amf3Integer(7)))
+    array.setProperty("extra", Amf3String("yes"))
+
+    val bytes = write(array)
+    assertEquals(bytes.size - 1, array.getSize())
+    val result = read(bytes) as Amf3Array
+    assertEquals(bytes.size - 1, result.getSize())
+    assertEquals(2, result.items.size)
+    assertEquals("hvc1", (result.items[0] as Amf3String).value)
+    assertEquals(7, (result.items[1] as Amf3Integer).value)
+    assertEquals("yes", (result.getProperty("extra") as Amf3String).value)
+  }
+
+  @Test
+  fun `dictionary round trip`() {
+    val dictionary = Amf3Dictionary()
+    dictionary.setProperty("width", 1920.0)
+    dictionary.setProperty("codec", "avc1")
+
+    val bytes = write(dictionary)
+    assertEquals(bytes.size - 1, dictionary.getSize())
+    val result = read(bytes) as Amf3Dictionary
+    assertEquals(bytes.size - 1, result.getSize())
+    assertEquals(1920.0, (result.getProperty("width") as Amf3Double).value, 0.0)
+    assertEquals("avc1", (result.getProperty("codec") as Amf3String).value)
+  }
+
+  @Test
+  fun `integer u29 boundaries round trip`() {
+    val cases = mapOf(
+      0 to 1, 127 to 1, 128 to 2, 0x3FFF to 2, 0x4000 to 3,
+      0x1FFFFF to 3, 0x200000 to 4, 0xFFFFFFF to 4, -1 to 4, -0x10000000 to 4
+    )
+    cases.forEach { (value, expectedSize) ->
+      val integer = Amf3Integer(value)
+      assertEquals("size of $value", expectedSize, integer.getSize())
+      val bytes = write(integer)
+      assertEquals("written size of $value", expectedSize, bytes.size - 1)
+      assertEquals("round trip of $value", value, (read(bytes) as Amf3Integer).value)
+    }
+  }
+
+  @Test
+  fun `string u29 length boundary round trip`() {
+    //length 64: the shifted U29 (129) needs 2 bytes while the raw length only needs 1
+    val text = "a".repeat(64)
+    val string = Amf3String(text)
+    val bytes = write(string)
+    assertEquals(bytes.size - 1, string.getSize())
+    val result = read(bytes) as Amf3String
+    assertEquals(text, result.value)
+    assertEquals(bytes.size - 1, result.getSize())
+  }
+}