[RTE1] Implement non-blocking emitter

[sacOO7]

• Implement thread-safe, non-blocking emitter in kotlin as per spec RTE1
• Ideally, should expose methods that can work for both kotlin/java.
• Should be able to process events in provided coroutinescope.
• Ideally, events should be processed async in kotlin ( under coroutinescope ), since communication between coroutines mostly requires parent coroutine scope. i.e. users can choose to emit event via flow, channel etc while processing messages, room state changes ... This is not possible if subscribers are blocking or not running under coroutinescope.
• Implement proper tests for the same.
• Current event-emitter implementation in java is thread-safe, but blocks the emitter thread till all subscribers process the event
  https://github.com/ably/ably-java/blob/main/lib/src/main/java/io/ably/lib/util/EventEmitter.java

┆Issue is synchronized with this Jira Task by Unito

[sacOO7]

Currently, we have implemented event emitter with two different ways ->

1. Flow based eventemitter - [ECO-5013][CHA-RL4] Flow based Event Emitter #35

• Somehow, this doesn't fit our use-case well, since flow doesn't allow proper cancellation of collectors/subscribers.
• We need to use hot flows, those don't close until we explicitly specify it in the code.
• Since, this is a pull-based approach ( hot flows ), rather than emitter push, more likelyhood of causing memory leaks, without proper garbage collection
• Tests for those doesn't work as expected, we need to add workarounds to make emitter work.
• More information -> [ECO-5013][CHA-RL4] Flow based Event Emitter #35 (comment)
• Likely to cause change in public API interface, also hot flows can slow down other subscribers.

2. Emit based eventemitter

• This solution fits our requirement, each subscriber process events independently when notified by emitter.
• All events are processed based on emit and doesn't need to keep observing for incoming events, i.e. garbage collector will automatically clean resources when no jobs running.
• We have proper tests for the same and works as expected, see async emitter tests
• We have extended this emitter with GenericEmitter that works for both java/kotlin -> Replace blocking emitter with non-blocking emitter #40