Make the afterEOSE logic subscription dependent.

app/src/main/java/com/vitorpamplona/amethyst/service/relays/Relay.kt

@@ -59,7 +59,7 @@ class Relay(
 
     var closingTimeInSeconds = 0L
 
-    var afterEOSE = false
+    var afterEOSEPerSubscription = mutableMapOf<String, Boolean>()
 
     val authResponse = mutableMapOf<HexKey, Boolean>()
 
@@ -87,6 +87,7 @@ class Relay(
     private var connectingBlock = AtomicBoolean()
 
     fun connectAndRun(onConnected: (Relay) -> Unit) {
+        Log.d("Client", "Relay.connect $url")
         // BRB is crashing OkHttp Deflater object :(
         if (url.contains("brb.io")) return
 
@@ -184,7 +185,7 @@ class Relay(
     }
 
     fun markConnectionAsReady(pingInMs: Long, usingCompression: Boolean) {
-        this.afterEOSE = false
+        this.afterEOSEPerSubscription.clear()
         this.isReady = true
         this.pingInMs = pingInMs
         this.usingCompression = usingCompression
@@ -194,35 +195,39 @@ class Relay(
         this.socket = null
         this.isReady = false
         this.usingCompression = false
-        this.afterEOSE = false
+        this.afterEOSEPerSubscription.clear()
         this.closingTimeInSeconds = TimeUtils.now()
     }
 
     fun processNewRelayMessage(newMessage: String) {
         val msgArray = Event.mapper.readTree(newMessage)
         val type = msgArray.get(0).asText()
-        val channel = msgArray.get(1).asText()
 
         when (type) {
             "EVENT" -> {
+                val subscriptionId = msgArray.get(1).asText()
                 val event = Event.fromJson(msgArray.get(2))
 
-                // Log.w("Relay", "Relay onEVENT ${event.kind} $url, $channel ${msgArray.get(2)}")
+                // Log.w("Relay", "Relay onEVENT ${event.kind} $url, $subscriptionId ${msgArray.get(2)}")
                 listeners.forEach {
-                    it.onEvent(this@Relay, channel, event)
+                    it.onEvent(this@Relay, subscriptionId, event)
-                    if (afterEOSE) {
+                    if (afterEOSEPerSubscription[subscriptionId] == true) {
-                        it.onRelayStateChange(this@Relay, StateType.EOSE, channel)
+                        it.onRelayStateChange(this@Relay, StateType.EOSE, subscriptionId)
                     }
                 }
             }
             "EOSE" -> listeners.forEach {
-                afterEOSE = true
+                val subscriptionId = msgArray.get(1).asText()
-                // Log.w("Relay", "Relay onEOSE $url, $channel")
+
-                it.onRelayStateChange(this@Relay, StateType.EOSE, channel)
+                afterEOSEPerSubscription[subscriptionId] = true
+                // Log.w("Relay", "Relay onEOSE $url $subscriptionId")
+                it.onRelayStateChange(this@Relay, StateType.EOSE, subscriptionId)
             }
             "NOTICE" -> listeners.forEach {
-                Log.w("Relay", "Relay onNotice $url, $channel")
+                val message = msgArray.get(1).asText()
-                it.onError(this@Relay, channel, Error("Relay sent notice: " + channel))
+                Log.w("Relay", "Relay onNotice $url, $message")
+
+                it.onError(this@Relay, message, Error("Relay sent notice: $message"))
             }
             "OK" -> listeners.forEach {
                 val eventId = msgArray[1].asText()
@@ -241,25 +246,29 @@ class Relay(
                 it.onSendResponse(this@Relay, eventId, success, message)
             }
             "AUTH" -> listeners.forEach {
-                // Log.w("Relay", "Relay$url, ${msg[1].asString}")
+                // Log.w("Relay", "Relay onAuth $url, ${msg[1].asString}")
                 it.onAuth(this@Relay, msgArray[1].asText())
             }
             "NOTIFY" -> listeners.forEach {
-                // Log.w("Relay", "Relay$url, ${msg[1].asString}")
+                // Log.w("Relay", "Relay onNotify $url, ${msg[1].asString}")
                 it.onNotify(this@Relay, msgArray[1].asText())
             }
+            "CLOSED" -> listeners.forEach {
+                Log.w("Relay", "Relay onClosed $url, $newMessage")
+            }
             else -> listeners.forEach {
                 Log.w("Relay", "Unsupported message: $newMessage")
                 it.onError(
                     this@Relay,
-                    channel,
+                    "",
-                    Error("Unknown type $type on channel $channel. Msg was $newMessage")
+                    Error("Unknown type $type on channel. Msg was $newMessage")
                 )
             }
         }
     }
 
     fun disconnect() {
+        Log.d("Client", "Relay.disconnect $url")
         checkNotInMainThread()
 
         closingTimeInSeconds = TimeUtils.now()
@@ -267,7 +276,7 @@ class Relay(
         socket = null
         isReady = false
         usingCompression = false
-        afterEOSE = false
+        afterEOSEPerSubscription.clear()
     }
 
     fun sendFilter(requestId: String) {
@@ -280,10 +289,12 @@ class Relay(
                     if (filters.isNotEmpty()) {
                         val request =
                             """["REQ","$requestId",${filters.take(10).joinToString(",") { it.filter.toJson(url) }}]"""
+
                         // Log.d("Relay", "onFilterSent $url $requestId $request")
+
                         socket?.send(request)
                         eventUploadCounterInBytes += request.bytesUsedInMemory()
-                        afterEOSE = false
+                        afterEOSEPerSubscription.clear()
                     }
                 }
             } else {