from __future__ import annotations

from abc import ABC, abstractmethod
from asyncio import CancelledError, Queue, Task, create_task
from collections import defaultdict
from contextlib import suppress
from typing import TYPE_CHECKING, Any, DefaultDict, Sequence

import sniffio

from starlite.exceptions import ImproperlyConfiguredException

__all__ = ("BaseEventEmitterBackend", "SimpleEventEmitter")


if TYPE_CHECKING:
    from starlite.events.listener import EventListener


class BaseEventEmitterBackend(ABC):
    """Abstract class used to define event emitter backends."""

    __slots__ = ("listeners",)

    listeners: DefaultDict[str, set[EventListener]]

    def __init__(self, listeners: Sequence[EventListener]):
        """Create an event emitter instance.

        Args:
            listeners: A list of listeners.
        """
        self.listeners = defaultdict(set)
        for listener in listeners:
            for event_id in listener.event_ids:
                self.listeners[event_id].add(listener)

    @abstractmethod
    def emit(self, event_id: str, *args: Any, **kwargs: Any) -> None:  # pragma: no cover
        """Emit an event to all attached listeners.

        Args:
            event_id: The ID of the event to emit, e.g 'my_event'.
            *args: args to pass to the listener(s).
            **kwargs: kwargs to pass to the listener(s)

        Returns:
            None
        """
        raise NotImplementedError("not implemented")

    @abstractmethod
    async def on_startup(self) -> None:  # pragma: no cover
        """Hook called on application startup, used to establish connection or perform other async operations.

        Returns:
            None
        """
        raise NotImplementedError("not implemented")

    @abstractmethod
    async def on_shutdown(self) -> None:  # pragma: no cover
        """Hook called on application shutdown, used to perform cleanup.

        Returns:
            None
        """
        raise NotImplementedError("not implemented")


class SimpleEventEmitter(BaseEventEmitterBackend):
    """Event emitter the works only in the current process"""

    __slots__ = ("_queue", "_worker_task")

    _worker_task: Task | None

    def __init__(self, listeners: Sequence[EventListener]):
        """Create an event emitter instance.

        Args:
            listeners: A list of listeners.
        """
        super().__init__(listeners=listeners)
        self._queue: Queue | None = None
        self._worker_task = None

    async def _worker(self) -> None:
        """Worker that runs in a separate task and continuously pulls events from asyncio queue.

        Returns:
            None
        """
        while self._queue:
            fn, args, kwargs = await self._queue.get()
            await fn(*args, *kwargs)
            self._queue.task_done()

    async def on_startup(self) -> None:
        """Hook called on application startup, used to establish connection or perform other async operations.

        Returns:
            None
        """
        if sniffio.current_async_library() != "asyncio":
            return

        self._queue = Queue()
        self._worker_task = create_task(self._worker())

    async def on_shutdown(self) -> None:
        """Hook called on application shutdown, used to perform cleanup.

        Returns:
            None
        """

        if self._queue:
            await self._queue.join()

        if self._worker_task:
            self._worker_task.cancel()
            with suppress(CancelledError):
                await self._worker_task

        self._worker_task = None
        self._queue = None

    def emit(self, event_id: str, *args: Any, **kwargs: Any) -> None:
        """Emit an event to all attached listeners.

        Args:
            event_id: The ID of the event to emit, e.g 'my_event'.
            *args: args to pass to the listener(s).
            **kwargs: kwargs to pass to the listener(s)

        Returns:
            None
        """
        if not (self._worker_task and self._queue):
            if sniffio.current_async_library() != "asyncio":
                raise ImproperlyConfiguredException("{type(self).__name__} only supports 'asyncio' based event loops")

            raise ImproperlyConfiguredException("Worker not running")

        if listeners := self.listeners.get(event_id):
            for listener in listeners:
                self._queue.put_nowait((listener.fn, args, kwargs))
            return
        raise ImproperlyConfiguredException(f"no event listeners are registered for event ID: {event_id}")

starlite/events/emitter.py
from __future__ import annotations

from abc import ABC, abstractmethod
from asyncio import CancelledError, Queue, Task, create_task
from collections import defaultdict
from contextlib import suppress
from typing import TYPE_CHECKING, Any, DefaultDict, Sequence

import sniffio

from starlite.exceptions import ImproperlyConfiguredException

__all__ = ("BaseEventEmitterBackend", "SimpleEventEmitter")


if TYPE_CHECKING:
    from starlite.events.listener import EventListener


class BaseEventEmitterBackend(ABC):
    """Abstract class used to define event emitter backends."""

    __slots__ = ("listeners",)

    listeners: DefaultDict[str, set[EventListener]]

    def __init__(self, listeners: Sequence[EventListener]):
        """Create an event emitter instance.

        Args:
            listeners: A list of listeners.
        """
        self.listeners = defaultdict(set)
        for listener in listeners:
            for event_id in listener.event_ids:
                self.listeners[event_id].add(listener)

    @abstractmethod
    def emit(self, event_id: str, *args: Any, **kwargs: Any) -> None:  # pragma: no cover
        """Emit an event to all attached listeners.

        Args:
            event_id: The ID of the event to emit, e.g 'my_event'.
            *args: args to pass to the listener(s).
            **kwargs: kwargs to pass to the listener(s)

        Returns:
            None
        """
        raise NotImplementedError("not implemented")

    @abstractmethod
    async def on_startup(self) -> None:  # pragma: no cover
        """Hook called on application startup, used to establish connection or perform other async operations.

        Returns:
            None
        """
        raise NotImplementedError("not implemented")

    @abstractmethod
    async def on_shutdown(self) -> None:  # pragma: no cover
        """Hook called on application shutdown, used to perform cleanup.

        Returns:
            None
        """
        raise NotImplementedError("not implemented")


class SimpleEventEmitter(BaseEventEmitterBackend):
    """Event emitter the works only in the current process"""

    __slots__ = ("_queue", "_worker_task")

    _worker_task: Task | None

    def __init__(self, listeners: Sequence[EventListener]):
        """Create an event emitter instance.

        Args:
            listeners: A list of listeners.
        """
        super().__init__(listeners=listeners)
        self._queue: Queue | None = None
        self._worker_task = None

    async def _worker(self) -> None:
        """Worker that runs in a separate task and continuously pulls events from asyncio queue.

        Returns:
            None
        """
        while self._queue:
            fn, args, kwargs = await self._queue.get()
            await fn(*args, **kwargs)
            self._queue.task_done()

    async def on_startup(self) -> None:
        """Hook called on application startup, used to establish connection or perform other async operations.

        Returns:
            None
        """
        if sniffio.current_async_library() != "asyncio":
            return

        self._queue = Queue()
        self._worker_task = create_task(self._worker())

    async def on_shutdown(self) -> None:
        """Hook called on application shutdown, used to perform cleanup.

        Returns:
            None
        """

        if self._queue:
            await self._queue.join()

        if self._worker_task:
            self._worker_task.cancel()
            with suppress(CancelledError):
                await self._worker_task

        self._worker_task = None
        self._queue = None

    def emit(self, event_id: str, *args: Any, **kwargs: Any) -> None:
        """Emit an event to all attached listeners.

        Args:
            event_id: The ID of the event to emit, e.g 'my_event'.
            *args: args to pass to the listener(s).
            **kwargs: kwargs to pass to the listener(s)

        Returns:
            None
        """
        if not (self._worker_task and self._queue):
            if sniffio.current_async_library() != "asyncio":
                raise ImproperlyConfiguredException("{type(self).__name__} only supports 'asyncio' based event loops")

            raise ImproperlyConfiguredException("Worker not running")

        if listeners := self.listeners.get(event_id):
            for listener in listeners:
                self._queue.put_nowait((listener.fn, args, kwargs))
            return
        raise ImproperlyConfiguredException(f"no event listeners are registered for event ID: {event_id}")

starlite/events/emitter.py
diff --git a/starlite/events/emitter.py b/starlite/events/emitter.py
index 1131725893..0ff8be5a19 100644
--- a/starlite/events/emitter.py
+++ b/starlite/events/emitter.py
@@ -93,7 +93,7 @@ async def _worker(self) -> None:
         """
         while self._queue:
             fn, args, kwargs = await self._queue.get()
-            await fn(*args, *kwargs)
+            await fn(*args, **kwargs)
             self._queue.task_done()
 
     async def on_startup(self) -> None:
diff --git a/tests/events/test_listener.py b/tests/events/test_listener.py
index a092198d04..e72ae1d49e 100644
--- a/tests/events/test_listener.py
+++ b/tests/events/test_listener.py
@@ -38,17 +38,15 @@ async def listener_fn(*args: Any, **kwargs: Any) -> None:
 
 @pytest.mark.parametrize("listener", [lazy_fixture("sync_listener"), lazy_fixture("async_listener")])
 def test_event_listener(mock: MagicMock, listener: EventListener) -> None:
-    test_value = {"key": "123"}
-
     @get("/")
     def route_handler(request: Request[Any, Any, Any]) -> None:
-        request.app.emit("test_event", test_value)
+        request.app.emit("test_event", "positional", keyword="keyword-value")
 
     with create_test_client(route_handlers=[route_handler], listeners=[listener]) as client:
         response = client.get("/")
         assert response.status_code == HTTP_200_OK
         sleep(0.01)
-        mock.assert_called_with(test_value)
+        mock.assert_called_with("positional", keyword="keyword-value")
 
 
 async def test_shutdown_awaits_pending(async_listener: EventListener, mock: MagicMock) -> None:

StaticFilesConfig and virtual directories
I'm trying to write a ``FileSystemProtocol`` to load files from the package data using [importlib_resources](https://importlib-resources.readthedocs.io/en/latest/using.html#). But because ``directories`` is defined as ``DirectoryPath``, pydantic checks if the given directories exist in the local filesystem. 

This is not generally true, especially in any kind of virtual filesystem (e.g. a zipped package). I think this condition should be relaxed to support virtual filesystems.

https://github.com/starlite-api/starlite/blob/9bb6dcd57c10a591377cf8e3a537e9292566d5b9/starlite/config/static_files.py#L32
