Allow to remove subscribers while iterating them

src/main/kotlin/ru/dbotthepony/mc/otm/core/ISubscripable.kt

@@ -2,6 +2,7 @@ package ru.dbotthepony.mc.otm.core
 
 import it.unimi.dsi.fastutil.booleans.BooleanConsumer
 import it.unimi.dsi.fastutil.floats.FloatConsumer
+import it.unimi.dsi.fastutil.objects.ReferenceArraySet
 import it.unimi.dsi.fastutil.objects.ReferenceLinkedOpenHashSet
 import java.util.function.Consumer
 import java.util.function.DoubleConsumer
@@ -23,22 +24,28 @@ interface ISubscriptable<V> {
 
 	class Impl<V> : ISubscriptable<V>, Consumer<V> {
 		private inner class L(val callback: Consumer<V>) : ISubscriptable.L {
+			private var isRemoved = false
+
 			init {
 				subscribers.add(this)
 			}
 
 			override fun remove() {
-				subscribers.remove(this)
+				isRemoved = true
+				queue.add(this)
 			}
 		}
 
 		private val subscribers = ReferenceLinkedOpenHashSet<L>(0)
+		private val queue = ReferenceArraySet<L>(0)
 
 		override fun addListener(listener: Consumer<V>): ISubscriptable.L {
 			return L(listener)
 		}
 
 		override fun accept(t: V) {
+			queue.forEach { subscribers.remove(it) }
+			queue.clear()
 			subscribers.forEach { it.callback.accept(t) }
 		}
 	}
@@ -67,22 +74,28 @@ interface IFloatSubcripable : ISubscriptable<Float> {
 
 	class Impl : IFloatSubcripable, Consumer<Float>, FloatConsumer {
 		private inner class L(val callback: FloatConsumer) : ISubscriptable.L {
+			private var isRemoved = false
+
 			init {
 				subscribers.add(this)
 			}
 
 			override fun remove() {
-				subscribers.remove(this)
+				isRemoved = true
+				queue.add(this)
 			}
 		}
 
 		private val subscribers = ReferenceLinkedOpenHashSet<L>(0)
+		private val queue = ReferenceArraySet<L>(0)
 
 		override fun addListener(listener: FloatConsumer): ISubscriptable.L {
 			return L(listener)
 		}
 
 		override fun accept(t: Float) {
+			queue.forEach { subscribers.remove(it) }
+			queue.clear()
 			subscribers.forEach { it.callback.accept(t) }
 		}
 	}
@@ -98,22 +111,28 @@ interface IDoubleSubcripable : ISubscriptable<Double> {
 
 	class Impl : IDoubleSubcripable, Consumer<Double>, DoubleConsumer {
 		private inner class L(val callback: DoubleConsumer) : ISubscriptable.L {
+			private var isRemoved = false
+
 			init {
 				subscribers.add(this)
 			}
 
 			override fun remove() {
-				subscribers.remove(this)
+				isRemoved = true
+				queue.add(this)
 			}
 		}
 
 		private val subscribers = ReferenceLinkedOpenHashSet<L>(0)
+		private val queue = ReferenceArraySet<L>(0)
 
 		override fun addListener(listener: DoubleConsumer): ISubscriptable.L {
 			return L(listener)
 		}
 
 		override fun accept(t: Double) {
+			queue.forEach { subscribers.remove(it) }
+			queue.clear()
 			subscribers.forEach { it.callback.accept(t) }
 		}
 	}
@@ -129,22 +148,28 @@ interface IIntSubcripable : ISubscriptable<Int> {
 
 	class Impl : IIntSubcripable, Consumer<Int>, IntConsumer {
 		private inner class L(val callback: IntConsumer) : ISubscriptable.L {
+			private var isRemoved = false
+
 			init {
 				subscribers.add(this)
 			}
 
 			override fun remove() {
-				subscribers.remove(this)
+				isRemoved = true
+				queue.add(this)
 			}
 		}
 
 		private val subscribers = ReferenceLinkedOpenHashSet<L>(0)
+		private val queue = ReferenceArraySet<L>(0)
 
 		override fun addListener(listener: IntConsumer): ISubscriptable.L {
 			return L(listener)
 		}
 
 		override fun accept(t: Int) {
+			queue.forEach { subscribers.remove(it) }
+			queue.clear()
 			subscribers.forEach { it.callback.accept(t) }
 		}
 	}
@@ -160,22 +185,28 @@ interface ILongSubcripable : ISubscriptable<Long> {
 
 	class Impl : ILongSubcripable, Consumer<Long>, LongConsumer {
 		private inner class L(val callback: LongConsumer) : ISubscriptable.L {
+			private var isRemoved = false
+
 			init {
 				subscribers.add(this)
 			}
 
 			override fun remove() {
-				subscribers.remove(this)
+				isRemoved = true
+				queue.add(this)
 			}
 		}
 
 		private val subscribers = ReferenceLinkedOpenHashSet<L>(0)
+		private val queue = ReferenceArraySet<L>(0)
 
 		override fun addListener(listener: LongConsumer): ISubscriptable.L {
 			return L(listener)
 		}
 
 		override fun accept(t: Long) {
+			queue.forEach { subscribers.remove(it) }
+			queue.clear()
 			subscribers.forEach { it.callback.accept(t) }
 		}
 	}
@@ -191,22 +222,28 @@ interface IBooleanSubscriptable : ISubscriptable<Boolean> {
 
 	class Impl : IBooleanSubscriptable, Consumer<Boolean>, BooleanConsumer {
 		private inner class L(val callback: BooleanConsumer) : ISubscriptable.L {
+			private var isRemoved = false
+
 			init {
 				subscribers.add(this)
 			}
 
 			override fun remove() {
-				subscribers.remove(this)
+				isRemoved = true
+				queue.add(this)
 			}
 		}
 
 		private val subscribers = ReferenceLinkedOpenHashSet<L>(0)
+		private val queue = ReferenceArraySet<L>(0)
 
 		override fun addListener(listener: BooleanConsumer): ISubscriptable.L {
 			return L(listener)
 		}
 
 		override fun accept(t: Boolean) {
+			queue.forEach { subscribers.remove(it) }
+			queue.clear()
 			subscribers.forEach { it.callback.accept(t) }
 		}
 	}