diff --git a/splitio/storage/inmemmory.py b/splitio/storage/inmemmory.py
index d51bb0ac..f551c0aa 100644
--- a/splitio/storage/inmemmory.py
+++ b/splitio/storage/inmemmory.py
@@ -5,7 +5,9 @@
 from collections import Counter
 
 from splitio.models.segments import Segment
-from splitio.models.telemetry import HTTPErrors, HTTPLatencies, MethodExceptions, MethodLatencies, LastSynchronization, StreamingEvents, TelemetryConfig, TelemetryCounters, CounterConstants
+from splitio.models.telemetry import HTTPErrors, HTTPLatencies, MethodExceptions, MethodLatencies, LastSynchronization, StreamingEvents, TelemetryConfig, TelemetryCounters, CounterConstants, \
+        HTTPErrorsAsync, HTTPLatenciesAsync, MethodExceptionsAsync, MethodLatenciesAsync, LastSynchronizationAsync, StreamingEventsAsync, TelemetryConfigAsync, TelemetryCountersAsync
+
 from splitio.storage import SplitStorage, SegmentStorage, ImpressionStorage, EventStorage, TelemetryStorage
 from splitio.optional.loaders import asyncio
 
@@ -575,7 +577,7 @@ def get_segments_keys_count(self):
                 total_count += len(self._segments[segment]._keys)
             return total_count
 
-
+          
 class InMemorySegmentStorageAsync(SegmentStorage):
     """In-memory implementation of a segment async storage."""
 
@@ -704,7 +706,43 @@ async def get_segments_keys_count(self):
             return total_count
 
 
-class InMemoryImpressionStorage(ImpressionStorage):
+class InMemoryImpressionStorageBase(ImpressionStorage):
+    """In memory implementation of an impressions base storage."""
+
+    def set_queue_full_hook(self, hook):
+        """
+        Set a hook to be called when the queue is full.
+
+        :param h: Hook to be called when the queue is full
+        """
+        if callable(hook):
+            self._queue_full_hook = hook
+
+    def put(self, impressions):
+        """
+        Put one or more impressions in storage.
+
+        :param impressions: List of one or more impressions to store.
+        :type impressions: list
+        """
+        pass
+
+    def pop_many(self, count):
+        """
+        Pop the oldest N impressions from storage.
+
+        :param count: Number of impressions to pop.
+        :type count: int
+        """
+        pass
+
+    def clear(self):
+        """
+        Clear data.
+        """
+        pass
+
+class InMemoryImpressionStorage(InMemoryImpressionStorageBase):
     """In memory implementation of an impressions storage."""
 
     def __init__(self, queue_size, telemetry_runtime_producer):
@@ -719,15 +757,6 @@ def __init__(self, queue_size, telemetry_runtime_producer):
         self._queue_full_hook = None
         self._telemetry_runtime_producer = telemetry_runtime_producer
 
-    def set_queue_full_hook(self, hook):
-        """
-        Set a hook to be called when the queue is full.
-
-        :param h: Hook to be called when the queue is full
-        """
-        if callable(hook):
-            self._queue_full_hook = hook
-
     def put(self, impressions):
         """
         Put one or more impressions in storage.
@@ -776,6 +805,71 @@ def clear(self):
             self._impressions = queue.Queue(maxsize=self._queue_size)
 
 
+class InMemoryImpressionStorageAsync(InMemoryImpressionStorageBase):
+    """In memory implementation of an impressions async storage."""
+
+    def __init__(self, queue_size, telemetry_runtime_producer):
+        """
+        Construct an instance.
+
+        :param eventsQueueSize: How many events to queue before forcing a submission
+        """
+        self._queue_size = queue_size
+        self._impressions = asyncio.Queue(maxsize=queue_size)
+        self._lock = asyncio.Lock()
+        self._queue_full_hook = None
+        self._telemetry_runtime_producer = telemetry_runtime_producer
+
+    async def put(self, impressions):
+        """
+        Put one or more impressions in storage.
+
+        :param impressions: List of one or more impressions to store.
+        :type impressions: list
+        """
+        impressions_stored = 0
+        try:
+            async with self._lock:
+                for impression in impressions:
+                    if self._impressions.qsize() == self._queue_size:
+                        raise asyncio.QueueFull
+                    await self._impressions.put(impression)
+                    impressions_stored += 1
+            await self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_QUEUED, len(impressions))
+            return True
+        except asyncio.QueueFull:
+            await self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_DROPPED, len(impressions) - impressions_stored)
+            await self._telemetry_runtime_producer.record_impression_stats(CounterConstants.IMPRESSIONS_QUEUED, impressions_stored)
+            if self._queue_full_hook is not None and callable(self._queue_full_hook):
+                await self._queue_full_hook()
+            _LOGGER.warning(
+                'Impression queue is full, failing to add more impressions. \n'
+                'Consider increasing parameter `impressionsQueueSize` in configuration'
+            )
+            return False
+
+    async def pop_many(self, count):
+        """
+        Pop the oldest N impressions from storage.
+
+        :param count: Number of impressions to pop.
+        :type count: int
+        """
+        impressions = []
+        async with self._lock:
+            while not self._impressions.empty() and count > 0:
+                impressions.append(await self._impressions.get())
+                count -= 1
+        return impressions
+
+    async def clear(self):
+        """
+        Clear data.
+        """
+        async with self._lock:
+            self._impressions = asyncio.Queue(maxsize=self._queue_size)
+
+
 class InMemoryEventStorage(EventStorage):
     """
     In memory storage for events.
@@ -856,14 +950,158 @@ def clear(self):
         with self._lock:
             self._events = queue.Queue(maxsize=self._queue_size)
 
-class InMemoryTelemetryStorage(TelemetryStorage):
+class InMemoryTelemetryStorageBase(TelemetryStorage):
+    """In-memory telemetry storage base."""
+
+    def _reset_tags(self):
+        self._tags = []
+
+    def _reset_config_tags(self):
+        self._config_tags = []
+
+    def record_config(self, config, extra_config):
+        """Record configurations."""
+        pass
+
+    def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):
+        """Record active and redundant factories."""
+        pass
+
+    def record_ready_time(self, ready_time):
+        """Record ready time."""
+        pass
+
+    def add_tag(self, tag):
+        """Record tag string."""
+        pass
+
+    def add_config_tag(self, tag):
+        """Record tag string."""
+        pass
+
+    def record_bur_time_out(self):
+        """Record block until ready timeout."""
+        pass
+
+    def record_not_ready_usage(self):
+        """record non-ready usage."""
+        pass
+
+    def record_latency(self, method, latency):
+        """Record method latency time."""
+        pass
+
+    def record_exception(self, method):
+        """Record method exception."""
+        pass
+
+    def record_impression_stats(self, data_type, count):
+        """Record impressions stats."""
+        pass
+
+    def record_event_stats(self, data_type, count):
+        """Record events stats."""
+        pass
+
+    def record_successful_sync(self, resource, time):
+        """Record successful sync."""
+        pass
+
+    def record_sync_error(self, resource, status):
+        """Record sync http error."""
+        pass
+
+    def record_sync_latency(self, resource, latency):
+        """Record latency time."""
+        pass
+
+    def record_auth_rejections(self):
+        """Record auth rejection."""
+        pass
+
+    def record_token_refreshes(self):
+        """Record sse token refresh."""
+        pass
+
+    def record_streaming_event(self, streaming_event):
+        """Record incoming streaming event."""
+        pass
+
+    def record_session_length(self, session):
+        """Record session length."""
+        pass
+
+    def get_bur_time_outs(self):
+        """Get block until ready timeout."""
+        pass
+
+    def get_non_ready_usage(self):
+        """Get non-ready usage."""
+        pass
+
+    def get_config_stats(self):
+        """Get all config info."""
+        pass
+
+    def pop_exceptions(self):
+        """Get and reset method exceptions."""
+        pass
+
+    def pop_tags(self):
+        """Get and reset tags."""
+        pass
+
+    def pop_config_tags(self):
+        """Get and reset tags."""
+        pass
+
+    def pop_latencies(self):
+        """Get and reset eval latencies."""
+        pass
+
+    def get_impressions_stats(self, type):
+        """Get impressions stats"""
+        pass
+
+    def get_events_stats(self, type):
+        """Get events stats"""
+        pass
+
+    def get_last_synchronization(self):
+        """Get last sync"""
+        pass
+
+    def pop_http_errors(self):
+        """Get and reset http errors."""
+        pass
+
+    def pop_http_latencies(self):
+        """Get and reset http latencies."""
+        pass
+
+    def pop_auth_rejections(self):
+        """Get and reset auth rejections."""
+        pass
+
+    def pop_token_refreshes(self):
+        """Get and reset token refreshes."""
+        pass
+
+    def pop_streaming_events(self):
+        """Get and reset streaming events"""
+        pass
+
+    def get_session_length(self):
+        """Get session length"""
+        pass
+
+
+class InMemoryTelemetryStorage(InMemoryTelemetryStorageBase):
     """In-memory telemetry storage."""
 
     def __init__(self):
         """Constructor"""
         self._lock = threading.RLock()
-        self._reset_tags()
-        self._reset_config_tags()
         self._method_exceptions = MethodExceptions()
         self._last_synchronization = LastSynchronization()
         self._counters = TelemetryCounters()
@@ -872,14 +1110,9 @@ def __init__(self):
         self._http_latencies = HTTPLatencies()
         self._streaming_events = StreamingEvents()
         self._tel_config = TelemetryConfig()
-
-    def _reset_tags(self):
-        with self._lock:
-            self._tags = []
-
-    def _reset_config_tags(self):
         with self._lock:
-            self._config_tags = []
+            self._reset_tags()
+            self._reset_config_tags()
 
     def record_config(self, config, extra_config):
         """Record configurations."""
@@ -1026,6 +1259,173 @@ def get_session_length(self):
         """Get session length"""
         return self._counters.get_session_length()
 
+
+class InMemoryTelemetryStorageAsync(InMemoryTelemetryStorageBase):
+    """In-memory telemetry async storage."""
+
+    async def create():
+        """Constructor"""
+        self = InMemoryTelemetryStorageAsync()
+        self._lock = asyncio.Lock()
+        self._method_exceptions = await MethodExceptionsAsync.create()
+        self._last_synchronization = await LastSynchronizationAsync.create()
+        self._counters = await TelemetryCountersAsync.create()
+        self._http_sync_errors = await HTTPErrorsAsync.create()
+        self._method_latencies = await MethodLatenciesAsync.create()
+        self._http_latencies = await HTTPLatenciesAsync.create()
+        self._streaming_events = await StreamingEventsAsync.create()
+        self._tel_config = await TelemetryConfigAsync.create()
+        async with self._lock:
+            self._reset_tags()
+            self._reset_config_tags()
+        return self
+
+    async def record_config(self, config, extra_config):
+        """Record configurations."""
+        await self._tel_config.record_config(config, extra_config)
+
+    async def record_active_and_redundant_factories(self, active_factory_count, redundant_factory_count):
+        """Record active and redundant factories."""
+        await self._tel_config.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
+
+    async def record_ready_time(self, ready_time):
+        """Record ready time."""
+        await self._tel_config.record_ready_time(ready_time)
+
+    async def add_tag(self, tag):
+        """Record tag string."""
+        async with self._lock:
+            if len(self._tags) < MAX_TAGS:
+                self._tags.append(tag)
+
+    async def add_config_tag(self, tag):
+        """Record tag string."""
+        async with self._lock:
+            if len(self._config_tags) < MAX_TAGS:
+                self._config_tags.append(tag)
+
+    async def record_bur_time_out(self):
+        """Record block until ready timeout."""
+        await self._tel_config.record_bur_time_out()
+
+    async def record_not_ready_usage(self):
+        """record non-ready usage."""
+        await self._tel_config.record_not_ready_usage()
+
+    async def record_latency(self, method, latency):
+        """Record method latency time."""
+        await self._method_latencies.add_latency(method,latency)
+
+    async def record_exception(self, method):
+        """Record method exception."""
+        await self._method_exceptions.add_exception(method)
+
+    async def record_impression_stats(self, data_type, count):
+        """Record impressions stats."""
+        await self._counters.record_impressions_value(data_type, count)
+
+    async def record_event_stats(self, data_type, count):
+        """Record events stats."""
+        await self._counters.record_events_value(data_type, count)
+
+    async def record_successful_sync(self, resource, time):
+        """Record successful sync."""
+        await self._last_synchronization.add_latency(resource, time)
+
+    async def record_sync_error(self, resource, status):
+        """Record sync http error."""
+        await self._http_sync_errors.add_error(resource, status)
+
+    async def record_sync_latency(self, resource, latency):
+        """Record latency time."""
+        await self._http_latencies.add_latency(resource, latency)
+
+    async def record_auth_rejections(self):
+        """Record auth rejection."""
+        await self._counters.record_auth_rejections()
+
+    async def record_token_refreshes(self):
+        """Record sse token refresh."""
+        await self._counters.record_token_refreshes()
+
+    async def record_streaming_event(self, streaming_event):
+        """Record incoming streaming event."""
+        await self._streaming_events.record_streaming_event(streaming_event)
+
+    async def record_session_length(self, session):
+        """Record session length."""
+        await self._counters.record_session_length(session)
+
+    async def get_bur_time_outs(self):
+        """Get block until ready timeout."""
+        return await self._tel_config.get_bur_time_outs()
+
+    async def get_non_ready_usage(self):
+        """Get non-ready usage."""
+        return await self._tel_config.get_non_ready_usage()
+
+    async def get_config_stats(self):
+        """Get all config info."""
+        return await self._tel_config.get_stats()
+
+    async def pop_exceptions(self):
+        """Get and reset method exceptions."""
+        return await self._method_exceptions.pop_all()
+
+    async def pop_tags(self):
+        """Get and reset tags."""
+        async with self._lock:
+            tags = self._tags
+            self._reset_tags()
+            return tags
+
+    async def pop_config_tags(self):
+        """Get and reset tags."""
+        async with self._lock:
+            tags = self._config_tags
+            self._reset_config_tags()
+            return tags
+
+    async def pop_latencies(self):
+        """Get and reset eval latencies."""
+        return await self._method_latencies.pop_all()
+
+    async def get_impressions_stats(self, type):
+        """Get impressions stats"""
+        return await self._counters.get_counter_stats(type)
+
+    async def get_events_stats(self, type):
+        """Get events stats"""
+        return await self._counters.get_counter_stats(type)
+
+    async def get_last_synchronization(self):
+        """Get last sync"""
+        return await self._last_synchronization.get_all()
+
+    async def pop_http_errors(self):
+        """Get and reset http errors."""
+        return await self._http_sync_errors.pop_all()
+
+    async def pop_http_latencies(self):
+        """Get and reset http latencies."""
+        return await self._http_latencies.pop_all()
+
+    async def pop_auth_rejections(self):
+        """Get and reset auth rejections."""
+        return await self._counters.pop_auth_rejections()
+
+    async def pop_token_refreshes(self):
+        """Get and reset token refreshes."""
+        return await self._counters.pop_token_refreshes()
+
+    async def pop_streaming_events(self):
+        return await self._streaming_events.pop_streaming_events()
+
+    async def get_session_length(self):
+        """Get session length"""
+        return await self._counters.get_session_length()
+
+
 class LocalhostTelemetryStorage():
     """Localhost telemetry storage."""
     def do_nothing(*_, **__):
diff --git a/tests/storage/test_inmemory_storage.py b/tests/storage/test_inmemory_storage.py
index 96f75416..b66a515e 100644
--- a/tests/storage/test_inmemory_storage.py
+++ b/tests/storage/test_inmemory_storage.py
@@ -8,10 +8,11 @@
 from splitio.models.impressions import Impression
 from splitio.models.events import Event, EventWrapper
 import splitio.models.telemetry as ModelTelemetry
-from splitio.engine.telemetry import TelemetryStorageProducer
+from splitio.engine.telemetry import TelemetryStorageProducer, TelemetryStorageProducerAsync
+
+from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, InMemorySegmentStorageAsync, InMemorySplitStorageAsync, \
+    InMemoryImpressionStorage, InMemoryEventStorage, InMemoryTelemetryStorage, InMemoryImpressionStorageAsync, InMemoryTelemetryStorageAsync
 
-from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \
-    InMemoryImpressionStorage, InMemoryEventStorage, InMemoryTelemetryStorage, InMemorySegmentStorageAsync, InMemorySplitStorageAsync
 
 class InMemorySplitStorageTests(object):
     """In memory split storage test cases."""
@@ -584,7 +585,7 @@ def test_clear(self, mocker):
         storage.clear()
         assert storage._impressions.qsize() == 0
 
-    def test_push_pop_impressions(self, mocker):
+    def test_impressions_dropped(self, mocker):
         """Test pushing and retrieving impressions."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
@@ -597,6 +598,98 @@ def test_push_pop_impressions(self, mocker):
         assert(telemetry_storage._counters._impressions_dropped == 1)
         assert(telemetry_storage._counters._impressions_queued == 2)
 
+
+class InMemoryImpressionsStorageAsyncTests(object):
+    """InMemory impressions async storage test cases."""
+
+    @pytest.mark.asyncio
+    async def test_push_pop_impressions(self, mocker):
+        """Test pushing and retrieving impressions."""
+        telemetry_storage = await InMemoryTelemetryStorageAsync.create()
+        telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage)
+        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+        storage = InMemoryImpressionStorageAsync(100, telemetry_runtime_producer)
+        await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)])
+        await storage.put([Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654)])
+        await storage.put([Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654)])
+        assert(telemetry_storage._counters._impressions_queued == 3)
+
+        # Assert impressions are retrieved in the same order they are inserted.
+        assert await storage.pop_many(1) == [
+            Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)
+        ]
+        assert await storage.pop_many(1) == [
+            Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654)
+        ]
+        assert await storage.pop_many(1) == [
+            Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654)
+        ]
+
+        # Assert inserting multiple impressions at once works and maintains order.
+        impressions = [
+            Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654),
+            Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654),
+            Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654)
+        ]
+        assert await storage.put(impressions)
+
+        # Assert impressions are retrieved in the same order they are inserted.
+        assert await storage.pop_many(1) == [
+            Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)
+        ]
+        assert await storage.pop_many(1) == [
+            Impression('key2', 'feature1', 'on', 'l1', 123456, 'b1', 321654)
+        ]
+        assert await storage.pop_many(1) == [
+            Impression('key3', 'feature1', 'on', 'l1', 123456, 'b1', 321654)
+        ]
+
+    @pytest.mark.asyncio
+    async def test_queue_full_hook(self, mocker):
+        """Test queue_full_hook is executed when the queue is full."""
+        telemetry_storage = await InMemoryTelemetryStorageAsync.create()
+        telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage)
+        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+        storage = InMemoryImpressionStorageAsync(100, telemetry_runtime_producer)
+        self.hook_called = False
+        async def queue_full_hook():
+            self.hook_called = True
+
+        storage.set_queue_full_hook(queue_full_hook)
+        impressions = [
+            Impression('key%d' % i, 'feature1', 'on', 'l1', 123456, 'b1', 321654)
+            for i in range(0, 101)
+        ]
+        await storage.put(impressions)
+        await queue_full_hook()
+        assert self.hook_called == True
+
+    @pytest.mark.asyncio
+    async def test_clear(self, mocker):
+        """Test clear method."""
+        telemetry_storage = await InMemoryTelemetryStorageAsync.create()
+        telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage)
+        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+        storage = InMemoryImpressionStorageAsync(100, telemetry_runtime_producer)
+        await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)])
+        assert storage._impressions.qsize() == 1
+        await storage.clear()
+        assert storage._impressions.qsize() == 0
+
+    @pytest.mark.asyncio
+    async def test_impressions_dropped(self, mocker):
+        """Test pushing and retrieving impressions."""
+        telemetry_storage = await InMemoryTelemetryStorageAsync.create()
+        telemetry_producer = TelemetryStorageProducerAsync(telemetry_storage)
+        telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
+        storage = InMemoryImpressionStorageAsync(2, telemetry_runtime_producer)
+        await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)])
+        await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)])
+        await storage.put([Impression('key1', 'feature1', 'on', 'l1', 123456, 'b1', 321654)])
+        assert(telemetry_storage._counters._impressions_dropped == 1)
+        assert(telemetry_storage._counters._impressions_queued == 2)
+
+
 class InMemoryEventsStorageTests(object):
     """InMemory events storage test cases."""