How to Build a Message Stream Queue in Python
A beginner-friendly MessageStream class built on deque that sends messages one at a time, tracks unread counts, and records sent items.
Python code
35 linesfrom collections import deque
import time
class MessageStream:
def __init__(self, messages):
self._queue = deque(messages)
self._sent = []
def send_next(self):
if not self._queue:
return None
message = self._queue.popleft()
self._sent.append(message)
return message
def unread_count(self):
return len(self._queue)
def sent_messages(self):
return self._sent.copy()
def clear(self):
self._queue.clear()
self._sent.clear()
if __name__ == "__main__":
stream = MessageStream(["Hello", "How are you?", "Let's meet tomorrow"])
stream.send_next()
stream.send_next()
print("Unread:", stream.unread_count())
print("Sent:", stream.sent_messages())
print("Next:", stream.send_next())
print("Unread after all:", stream.unread_count())Output
Unread: 1
Sent: ['Hello', 'How are you?']
Next: Let's meet tomorrow
Unread after all: 0How it works
The collections.deque provides O(1) popleft operations, making it ideal for a queue-based message stream. sent_messages() returns a copy to protect internal state from accidental mutation. send_next is idempotent-safe by returning None on empty queues rather than raising an error. This simple pattern mirrors how message brokers drain queues while preserving processing history.
Common mistakes
- Using a list and pop(0) which is O(n) and slow for large streams
- Returning the internal sent list directly instead of a copy, risking external mutation
- Forgetting to handle the empty queue case with a clear return value
Variations
- Use queue.Queue with thread-safe put/get for multi-threaded producers/consumers
- Convert to a generator with yield for lazy message delivery to a processing loop
Real-world use cases
- Implementing a lightweight in-memory message buffer for a chat application prototyper.
- Simulating a consumer that drains a message queue while tracking what was processed in a pipeline.
- Teaching streaming fundamentals by replaying a fixed batch of events to a data processing script.
Sponsored
More from Streaming & messaging
- At Most Once Fire-and-Forget Mock in Python easy
- Batch Consume Process Commit Pattern in Python medium
- Build a Streaming Messaging Helper in Python easy
- Dead Letter Queue Failed Messages List Mock in Python easy
- Dedupe processed message IDs in Python easy
- Event Envelope with Schema Version Field in Python easy
Keep learning
Related tutorials and quizzes for this topic.