[FEA]: Support for Async Operators

[cwharris]

Is this a new feature, an improvement, or a change to existing functionality?

New Feature

How would you describe the priority of this feature request

High

Please provide a clear description of problem this feature solves

MRC supports 1 of the following 4 mapping operations. Supporting all 4 would open up a multitude of opportunities.

# mrc.operators.map
def on_data(self, value: int):
    return value

# mrc.operators.flatmap
def on_data_iterator(self, value: int):
    for i in range(value):
        yield i

# mrc.operators.map_async
async def on_data_async(self, value: int):
    await asyncio.sleep(0.25)
    return value

# mrc.operators.flatmap_async
async def on_data_async_gen(self, value: int):
    for i in range(value):
        await asyncio.sleep(0.25)
        yield i

Describe your ideal solution

Write a custom fiber-enabled asyncio Event Loop and create one-per-thread to be used from within MRC operators.

Describe any alternatives you have considered

No response

Additional context

No response

Code of Conduct

• I agree to follow MRC's Code of Conduct
• I have searched the open feature requests and have found no duplicates for this feature request