We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Bug: Exception in event listener breaks listener
### Description

In a separate production code base, we encountered what I believe to be an issue/behavior that could be fixed/improved and did not find anything other references of it when searching the issues here. 

Basically, if an exception occurs inside of an event listener you will then see
```log
INFO:     ASGI 'lifespan' protocol appears unsupported.
```
 and the listener stream is closed which then prevents any additional events from being picked up. Any time `request.app.emit` is then called you will get `ClosedResourceError`.

I don't think a single exception inside an event execution should prevent all future events from getting picked up or at least the error should be a little more obvious as to what is going on.

I've created an example repo that recreates the issue consistently.

### URL to code causing the issue

https://github.com/bnjmn/litestar-emit-fail-example

### MCVE

See [full example here](https://github.com/bnjmn/litestar-emit-fail-example/tree/main)
```python
from litestar import Litestar, get, Request
from litestar.events import listener

import logging

logger = logging.getLogger(__name__)

@listener("raise_exception")
async def raise_exception_if_odd(value) -> None:
    """Raise an exception to test Emit error."""
    if value is not None and value % 2 != 0:
        raise ValueError(f"{value} is odd")
    else:
        return "The value is even. No exception raised."

@get("/")
async def index() -> str:
    return "Hello, world!"

@get("/check-value/{value:int}")
async def check_value(request: Request, value: int) -> str:
    try:
        request.app.emit("raise_exception", value)
        return f"Checked {value}: No exception raised."
    except ValueError as e:
        return str(e)

app = Litestar([index, check_value], listeners=[raise_exception_if_odd])
```


### Steps to reproduce

1. Follow the steps outlined in the [README](https://github.com/bnjmn/litestar-emit-fail-example/blob/main/README.md#run-it)
```bash
pipenv install
litestar run -d
```
- Go to http://localhost:8000/check-value/2 to see a successful request
- Go to http://localhost:8000/check-value/3 to see an exception raised in the listener
- Attempt to go to http://localhost:8000/check-value/2 again, but you should get ClosedResourceError

### Litestar Version

Recreated on both 2.3.2 and 2.4.1

### Platform

- [ ] Linux
- [X] Mac
- [ ] Windows
- [x] Other (Please specify in the description above)

<!-- POLAR PLEDGE BADGE START -->
---
> [!NOTE]  
> While we are open for sponsoring on [GitHub Sponsors](https://github.com/sponsors/litestar-org/) and 
> [OpenCollective](https://opencollective.com/litestar), we also utilize [Polar.sh](https://polar.sh/) to engage in pledge-based sponsorship.
>
> Check out all issues funded or available for funding [on our Polar.sh Litestar dashboard](https://polar.sh/litestar-org)
> * If you would like to see an issue prioritized, make a pledge towards it!
> * We receive the pledge once the issue is completed & verified
> * This, along with engagement in the community, helps us know which features are a priority to our users.

<a href="https://polar.sh/litestar-org/litestar/issues/2809">
<picture>
  <source media="(prefers-color-scheme: dark)" srcset="https://polar.sh/api/github/litestar-org/litestar/issues/2809/pledge.svg?darkmode=1">
  <img alt="Fund with Polar" src="https://polar.sh/api/github/litestar-org/litestar/issues/2809/pledge.svg">
</picture>
</a>
<!-- POLAR PLEDGE BADGE END -->


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `litestar/events/emitter.py`
Content:
```
1 from __future__ import annotations
2 
3 import math
4 import sys
5 from abc import ABC, abstractmethod
6 from collections import defaultdict
7 from contextlib import AsyncExitStack
8 from functools import partial
9 from typing import TYPE_CHECKING, Any, Sequence
10 
11 if sys.version_info < (3, 9):
12     from typing import AsyncContextManager
13 else:
14     from contextlib import AbstractAsyncContextManager as AsyncContextManager
15 
16 import anyio
17 
18 from litestar.exceptions import ImproperlyConfiguredException
19 
20 __all__ = ("BaseEventEmitterBackend", "SimpleEventEmitter")
21 
22 
23 if TYPE_CHECKING:
24     from types import TracebackType
25 
26     from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
27 
28     from litestar.events.listener import EventListener
29 
30 
31 class BaseEventEmitterBackend(AsyncContextManager["BaseEventEmitterBackend"], ABC):
32     """Abstract class used to define event emitter backends."""
33 
34     __slots__ = ("listeners",)
35 
36     listeners: defaultdict[str, set[EventListener]]
37 
38     def __init__(self, listeners: Sequence[EventListener]) -> None:
39         """Create an event emitter instance.
40 
41         Args:
42             listeners: A list of listeners.
43         """
44         self.listeners = defaultdict(set)
45         for listener in listeners:
46             for event_id in listener.event_ids:
47                 self.listeners[event_id].add(listener)
48 
49     @abstractmethod
50     def emit(self, event_id: str, *args: Any, **kwargs: Any) -> None:
51         """Emit an event to all attached listeners.
52 
53         Args:
54             event_id: The ID of the event to emit, e.g 'my_event'.
55             *args: args to pass to the listener(s).
56             **kwargs: kwargs to pass to the listener(s)
57 
58         Returns:
59             None
60         """
61         raise NotImplementedError("not implemented")
62 
63 
64 class SimpleEventEmitter(BaseEventEmitterBackend):
65     """Event emitter the works only in the current process"""
66 
67     __slots__ = ("_queue", "_exit_stack", "_receive_stream", "_send_stream")
68 
69     def __init__(self, listeners: Sequence[EventListener]) -> None:
70         """Create an event emitter instance.
71 
72         Args:
73             listeners: A list of listeners.
74         """
75         super().__init__(listeners=listeners)
76         self._receive_stream: MemoryObjectReceiveStream | None = None
77         self._send_stream: MemoryObjectSendStream | None = None
78         self._exit_stack: AsyncExitStack | None = None
79 
80     @staticmethod
81     async def _worker(receive_stream: MemoryObjectReceiveStream) -> None:
82         """Run items from ``receive_stream`` in a task group.
83 
84         Returns:
85             None
86         """
87         async with receive_stream, anyio.create_task_group() as task_group:
88             async for item in receive_stream:
89                 fn, args, kwargs = item
90                 if kwargs:
91                     fn = partial(fn, **kwargs)
92                 task_group.start_soon(fn, *args)
93 
94     async def __aenter__(self) -> SimpleEventEmitter:
95         self._exit_stack = AsyncExitStack()
96         send_stream, receive_stream = anyio.create_memory_object_stream(math.inf)  # type: ignore[var-annotated]
97         self._send_stream = send_stream
98         task_group = anyio.create_task_group()
99 
100         await self._exit_stack.enter_async_context(task_group)
101         await self._exit_stack.enter_async_context(send_stream)
102         task_group.start_soon(self._worker, receive_stream)
103 
104         return self
105 
106     async def __aexit__(
107         self,
108         exc_type: type[BaseException] | None,
109         exc_val: BaseException | None,
110         exc_tb: TracebackType | None,
111     ) -> None:
112         if self._exit_stack:
113             await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)
114 
115         self._exit_stack = None
116         self._send_stream = None
117 
118     def emit(self, event_id: str, *args: Any, **kwargs: Any) -> None:
119         """Emit an event to all attached listeners.
120 
121         Args:
122             event_id: The ID of the event to emit, e.g 'my_event'.
123             *args: args to pass to the listener(s).
124             **kwargs: kwargs to pass to the listener(s)
125 
126         Returns:
127             None
128         """
129         if not (self._send_stream and self._exit_stack):
130             raise RuntimeError("Emitter not initialized")
131 
132         if listeners := self.listeners.get(event_id):
133             for listener in listeners:
134                 self._send_stream.send_nowait((listener.fn, args, kwargs))
135             return
136         raise ImproperlyConfiguredException(f"no event listeners are registered for event ID: {event_id}")
137
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/litestar/events/emitter.py b/litestar/events/emitter.py
--- a/litestar/events/emitter.py
+++ b/litestar/events/emitter.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import math
 import sys
 from abc import ABC, abstractmethod
@@ -17,9 +18,6 @@
 
 from litestar.exceptions import ImproperlyConfiguredException
 
-__all__ = ("BaseEventEmitterBackend", "SimpleEventEmitter")
-
-
 if TYPE_CHECKING:
     from types import TracebackType
 
@@ -27,6 +25,10 @@
 
     from litestar.events.listener import EventListener
 
+__all__ = ("BaseEventEmitterBackend", "SimpleEventEmitter")
+
+logger = logging.getLogger(__name__)
+
 
 class BaseEventEmitterBackend(AsyncContextManager["BaseEventEmitterBackend"], ABC):
     """Abstract class used to define event emitter backends."""
@@ -77,19 +79,25 @@
         self._send_stream: MemoryObjectSendStream | None = None
         self._exit_stack: AsyncExitStack | None = None
 
-    @staticmethod
-    async def _worker(receive_stream: MemoryObjectReceiveStream) -> None:
+    async def _worker(self, receive_stream: MemoryObjectReceiveStream) -> None:
         """Run items from ``receive_stream`` in a task group.
 
         Returns:
             None
         """
-        async with receive_stream, anyio.create_task_group() as task_group:
+        async with receive_stream:
             async for item in receive_stream:
-                fn, args, kwargs = item
+                await self._run_listener_in_task_group(*item)
+
+    @staticmethod
+    async def _run_listener_in_task_group(fn: Any, args: tuple[Any], kwargs: dict[str, Any]) -> None:
+        try:
+            async with anyio.create_task_group() as task_group:
                 if kwargs:
                     fn = partial(fn, **kwargs)
                 task_group.start_soon(fn, *args)
+        except Exception as exc:
+            logger.exception("Error in event listener: %s", exc)
 
     async def __aenter__(self) -> SimpleEventEmitter:
         self._exit_stack = AsyncExitStack()

{"golden_diff": "diff --git a/litestar/events/emitter.py b/litestar/events/emitter.py\n--- a/litestar/events/emitter.py\n+++ b/litestar/events/emitter.py\n@@ -1,5 +1,6 @@\n from __future__ import annotations\n \n+import logging\n import math\n import sys\n from abc import ABC, abstractmethod\n@@ -17,9 +18,6 @@\n \n from litestar.exceptions import ImproperlyConfiguredException\n \n-__all__ = (\"BaseEventEmitterBackend\", \"SimpleEventEmitter\")\n-\n-\n if TYPE_CHECKING:\n     from types import TracebackType\n \n@@ -27,6 +25,10 @@\n \n     from litestar.events.listener import EventListener\n \n+__all__ = (\"BaseEventEmitterBackend\", \"SimpleEventEmitter\")\n+\n+logger = logging.getLogger(__name__)\n+\n \n class BaseEventEmitterBackend(AsyncContextManager[\"BaseEventEmitterBackend\"], ABC):\n     \"\"\"Abstract class used to define event emitter backends.\"\"\"\n@@ -77,19 +79,25 @@\n         self._send_stream: MemoryObjectSendStream | None = None\n         self._exit_stack: AsyncExitStack | None = None\n \n-    @staticmethod\n-    async def _worker(receive_stream: MemoryObjectReceiveStream) -> None:\n+    async def _worker(self, receive_stream: MemoryObjectReceiveStream) -> None:\n         \"\"\"Run items from ``receive_stream`` in a task group.\n \n         Returns:\n             None\n         \"\"\"\n-        async with receive_stream, anyio.create_task_group() as task_group:\n+        async with receive_stream:\n             async for item in receive_stream:\n-                fn, args, kwargs = item\n+                await self._run_listener_in_task_group(*item)\n+\n+    @staticmethod\n+    async def _run_listener_in_task_group(fn: Any, args: tuple[Any], kwargs: dict[str, Any]) -> None:\n+        try:\n+            async with anyio.create_task_group() as task_group:\n                 if kwargs:\n                     fn = partial(fn, **kwargs)\n                 task_group.start_soon(fn, *args)\n+        except Exception as exc:\n+            logger.exception(\"Error in event listener: %s\", exc)\n \n     async def __aenter__(self) -> SimpleEventEmitter:\n         self._exit_stack = AsyncExitStack()\n", "issue": "Bug: Exception in event listener breaks listener\n### Description\r\n\r\nIn a separate production code base, we encountered what I believe to be an issue/behavior that could be fixed/improved and did not find anything other references of it when searching the issues here. \r\n\r\nBasically, if an exception occurs inside of an event listener you will then see\r\n```log\r\nINFO:     ASGI 'lifespan' protocol appears unsupported.\r\n```\r\n and the listener stream is closed which then prevents any additional events from being picked up. Any time `request.app.emit` is then called you will get `ClosedResourceError`.\r\n\r\nI don't think a single exception inside an event execution should prevent all future events from getting picked up or at least the error should be a little more obvious as to what is going on.\r\n\r\nI've created an example repo that recreates the issue consistently.\r\n\r\n### URL to code causing the issue\r\n\r\nhttps://github.com/bnjmn/litestar-emit-fail-example\r\n\r\n### MCVE\r\n\r\nSee [full example here](https://github.com/bnjmn/litestar-emit-fail-example/tree/main)\r\n```python\r\nfrom litestar import Litestar, get, Request\r\nfrom litestar.events import listener\r\n\r\nimport logging\r\n\r\nlogger = logging.getLogger(__name__)\r\n\r\n@listener(\"raise_exception\")\r\nasync def raise_exception_if_odd(value) -> None:\r\n    \"\"\"Raise an exception to test Emit error.\"\"\"\r\n    if value is not None and value % 2 != 0:\r\n        raise ValueError(f\"{value} is odd\")\r\n    else:\r\n        return \"The value is even. No exception raised.\"\r\n\r\n@get(\"/\")\r\nasync def index() -> str:\r\n    return \"Hello, world!\"\r\n\r\n@get(\"/check-value/{value:int}\")\r\nasync def check_value(request: Request, value: int) -> str:\r\n    try:\r\n        request.app.emit(\"raise_exception\", value)\r\n        return f\"Checked {value}: No exception raised.\"\r\n    except ValueError as e:\r\n        return str(e)\r\n\r\napp = Litestar([index, check_value], listeners=[raise_exception_if_odd])\r\n```\r\n\r\n\r\n### Steps to reproduce\r\n\r\n1. Follow the steps outlined in the [README](https://github.com/bnjmn/litestar-emit-fail-example/blob/main/README.md#run-it)\r\n```bash\r\npipenv install\r\nlitestar run -d\r\n```\r\n- Go to http://localhost:8000/check-value/2 to see a successful request\r\n- Go to http://localhost:8000/check-value/3 to see an exception raised in the listener\r\n- Attempt to go to http://localhost:8000/check-value/2 again, but you should get ClosedResourceError\r\n\r\n### Litestar Version\r\n\r\nRecreated on both 2.3.2 and 2.4.1\r\n\r\n### Platform\r\n\r\n- [ ] Linux\r\n- [X] Mac\r\n- [ ] Windows\r\n- [x] Other (Please specify in the description above)\r\n\r\n<!-- POLAR PLEDGE BADGE START -->\r\n---\r\n> [!NOTE]  \r\n> While we are open for sponsoring on [GitHub Sponsors](https://github.com/sponsors/litestar-org/) and \r\n> [OpenCollective](https://opencollective.com/litestar), we also utilize [Polar.sh](https://polar.sh/) to engage in pledge-based sponsorship.\r\n>\r\n> Check out all issues funded or available for funding [on our Polar.sh Litestar dashboard](https://polar.sh/litestar-org)\r\n> * If you would like to see an issue prioritized, make a pledge towards it!\r\n> * We receive the pledge once the issue is completed & verified\r\n> * This, along with engagement in the community, helps us know which features are a priority to our users.\r\n\r\n<a href=\"https://polar.sh/litestar-org/litestar/issues/2809\">\r\n<picture>\r\n  <source media=\"(prefers-color-scheme: dark)\" srcset=\"https://polar.sh/api/github/litestar-org/litestar/issues/2809/pledge.svg?darkmode=1\">\r\n  <img alt=\"Fund with Polar\" src=\"https://polar.sh/api/github/litestar-org/litestar/issues/2809/pledge.svg\">\r\n</picture>\r\n</a>\r\n<!-- POLAR PLEDGE BADGE END -->\r\n\n", "before_files": [{"content": "from __future__ import annotations\n\nimport math\nimport sys\nfrom abc import ABC, abstractmethod\nfrom collections import defaultdict\nfrom contextlib import AsyncExitStack\nfrom functools import partial\nfrom typing import TYPE_CHECKING, Any, Sequence\n\nif sys.version_info < (3, 9):\n    from typing import AsyncContextManager\nelse:\n    from contextlib import AbstractAsyncContextManager as AsyncContextManager\n\nimport anyio\n\nfrom litestar.exceptions import ImproperlyConfiguredException\n\n__all__ = (\"BaseEventEmitterBackend\", \"SimpleEventEmitter\")\n\n\nif TYPE_CHECKING:\n    from types import TracebackType\n\n    from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream\n\n    from litestar.events.listener import EventListener\n\n\nclass BaseEventEmitterBackend(AsyncContextManager[\"BaseEventEmitterBackend\"], ABC):\n    \"\"\"Abstract class used to define event emitter backends.\"\"\"\n\n    __slots__ = (\"listeners\",)\n\n    listeners: defaultdict[str, set[EventListener]]\n\n    def __init__(self, listeners: Sequence[EventListener]) -> None:\n        \"\"\"Create an event emitter instance.\n\n        Args:\n            listeners: A list of listeners.\n        \"\"\"\n        self.listeners = defaultdict(set)\n        for listener in listeners:\n            for event_id in listener.event_ids:\n                self.listeners[event_id].add(listener)\n\n    @abstractmethod\n    def emit(self, event_id: str, *args: Any, **kwargs: Any) -> None:\n        \"\"\"Emit an event to all attached listeners.\n\n        Args:\n            event_id: The ID of the event to emit, e.g 'my_event'.\n            *args: args to pass to the listener(s).\n            **kwargs: kwargs to pass to the listener(s)\n\n        Returns:\n            None\n        \"\"\"\n        raise NotImplementedError(\"not implemented\")\n\n\nclass SimpleEventEmitter(BaseEventEmitterBackend):\n    \"\"\"Event emitter the works only in the current process\"\"\"\n\n    __slots__ = (\"_queue\", \"_exit_stack\", \"_receive_stream\", \"_send_stream\")\n\n    def __init__(self, listeners: Sequence[EventListener]) -> None:\n        \"\"\"Create an event emitter instance.\n\n        Args:\n            listeners: A list of listeners.\n        \"\"\"\n        super().__init__(listeners=listeners)\n        self._receive_stream: MemoryObjectReceiveStream | None = None\n        self._send_stream: MemoryObjectSendStream | None = None\n        self._exit_stack: AsyncExitStack | None = None\n\n    @staticmethod\n    async def _worker(receive_stream: MemoryObjectReceiveStream) -> None:\n        \"\"\"Run items from ``receive_stream`` in a task group.\n\n        Returns:\n            None\n        \"\"\"\n        async with receive_stream, anyio.create_task_group() as task_group:\n            async for item in receive_stream:\n                fn, args, kwargs = item\n                if kwargs:\n                    fn = partial(fn, **kwargs)\n                task_group.start_soon(fn, *args)\n\n    async def __aenter__(self) -> SimpleEventEmitter:\n        self._exit_stack = AsyncExitStack()\n        send_stream, receive_stream = anyio.create_memory_object_stream(math.inf)  # type: ignore[var-annotated]\n        self._send_stream = send_stream\n        task_group = anyio.create_task_group()\n\n        await self._exit_stack.enter_async_context(task_group)\n        await self._exit_stack.enter_async_context(send_stream)\n        task_group.start_soon(self._worker, receive_stream)\n\n        return self\n\n    async def __aexit__(\n        self,\n        exc_type: type[BaseException] | None,\n        exc_val: BaseException | None,\n        exc_tb: TracebackType | None,\n    ) -> None:\n        if self._exit_stack:\n            await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)\n\n        self._exit_stack = None\n        self._send_stream = None\n\n    def emit(self, event_id: str, *args: Any, **kwargs: Any) -> None:\n        \"\"\"Emit an event to all attached listeners.\n\n        Args:\n            event_id: The ID of the event to emit, e.g 'my_event'.\n            *args: args to pass to the listener(s).\n            **kwargs: kwargs to pass to the listener(s)\n\n        Returns:\n            None\n        \"\"\"\n        if not (self._send_stream and self._exit_stack):\n            raise RuntimeError(\"Emitter not initialized\")\n\n        if listeners := self.listeners.get(event_id):\n            for listener in listeners:\n                self._send_stream.send_nowait((listener.fn, args, kwargs))\n            return\n        raise ImproperlyConfiguredException(f\"no event listeners are registered for event ID: {event_id}\")\n", "path": "litestar/events/emitter.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport logging\nimport math\nimport sys\nfrom abc import ABC, abstractmethod\nfrom collections import defaultdict\nfrom contextlib import AsyncExitStack\nfrom functools import partial\nfrom typing import TYPE_CHECKING, Any, Sequence\n\nif sys.version_info < (3, 9):\n    from typing import AsyncContextManager\nelse:\n    from contextlib import AbstractAsyncContextManager as AsyncContextManager\n\nimport anyio\n\nfrom litestar.exceptions import ImproperlyConfiguredException\n\nif TYPE_CHECKING:\n    from types import TracebackType\n\n    from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream\n\n    from litestar.events.listener import EventListener\n\n__all__ = (\"BaseEventEmitterBackend\", \"SimpleEventEmitter\")\n\nlogger = logging.getLogger(__name__)\n\n\nclass BaseEventEmitterBackend(AsyncContextManager[\"BaseEventEmitterBackend\"], ABC):\n    \"\"\"Abstract class used to define event emitter backends.\"\"\"\n\n    __slots__ = (\"listeners\",)\n\n    listeners: defaultdict[str, set[EventListener]]\n\n    def __init__(self, listeners: Sequence[EventListener]) -> None:\n        \"\"\"Create an event emitter instance.\n\n        Args:\n            listeners: A list of listeners.\n        \"\"\"\n        self.listeners = defaultdict(set)\n        for listener in listeners:\n            for event_id in listener.event_ids:\n                self.listeners[event_id].add(listener)\n\n    @abstractmethod\n    def emit(self, event_id: str, *args: Any, **kwargs: Any) -> None:\n        \"\"\"Emit an event to all attached listeners.\n\n        Args:\n            event_id: The ID of the event to emit, e.g 'my_event'.\n            *args: args to pass to the listener(s).\n            **kwargs: kwargs to pass to the listener(s)\n\n        Returns:\n            None\n        \"\"\"\n        raise NotImplementedError(\"not implemented\")\n\n\nclass SimpleEventEmitter(BaseEventEmitterBackend):\n    \"\"\"Event emitter the works only in the current process\"\"\"\n\n    __slots__ = (\"_queue\", \"_exit_stack\", \"_receive_stream\", \"_send_stream\")\n\n    def __init__(self, listeners: Sequence[EventListener]) -> None:\n        \"\"\"Create an event emitter instance.\n\n        Args:\n            listeners: A list of listeners.\n        \"\"\"\n        super().__init__(listeners=listeners)\n        self._receive_stream: MemoryObjectReceiveStream | None = None\n        self._send_stream: MemoryObjectSendStream | None = None\n        self._exit_stack: AsyncExitStack | None = None\n\n    async def _worker(self, receive_stream: MemoryObjectReceiveStream) -> None:\n        \"\"\"Run items from ``receive_stream`` in a task group.\n\n        Returns:\n            None\n        \"\"\"\n        async with receive_stream:\n            async for item in receive_stream:\n                await self._run_listener_in_task_group(*item)\n\n    @staticmethod\n    async def _run_listener_in_task_group(fn: Any, args: tuple[Any], kwargs: dict[str, Any]) -> None:\n        try:\n            async with anyio.create_task_group() as task_group:\n                if kwargs:\n                    fn = partial(fn, **kwargs)\n                task_group.start_soon(fn, *args)\n        except Exception as exc:\n            logger.exception(\"Error in event listener: %s\", exc)\n\n    async def __aenter__(self) -> SimpleEventEmitter:\n        self._exit_stack = AsyncExitStack()\n        send_stream, receive_stream = anyio.create_memory_object_stream(math.inf)  # type: ignore[var-annotated]\n        self._send_stream = send_stream\n        task_group = anyio.create_task_group()\n\n        await self._exit_stack.enter_async_context(task_group)\n        await self._exit_stack.enter_async_context(send_stream)\n        task_group.start_soon(self._worker, receive_stream)\n\n        return self\n\n    async def __aexit__(\n        self,\n        exc_type: type[BaseException] | None,\n        exc_val: BaseException | None,\n        exc_tb: TracebackType | None,\n    ) -> None:\n        if self._exit_stack:\n            await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)\n\n        self._exit_stack = None\n        self._send_stream = None\n\n    def emit(self, event_id: str, *args: Any, **kwargs: Any) -> None:\n        \"\"\"Emit an event to all attached listeners.\n\n        Args:\n            event_id: The ID of the event to emit, e.g 'my_event'.\n            *args: args to pass to the listener(s).\n            **kwargs: kwargs to pass to the listener(s)\n\n        Returns:\n            None\n        \"\"\"\n        if not (self._send_stream and self._exit_stack):\n            raise RuntimeError(\"Emitter not initialized\")\n\n        if listeners := self.listeners.get(event_id):\n            for listener in listeners:\n                self._send_stream.send_nowait((listener.fn, args, kwargs))\n            return\n        raise ImproperlyConfiguredException(f\"no event listeners are registered for event ID: {event_id}\")\n", "path": "litestar/events/emitter.py"}]}