ℹ️ Select 'Choose Exercise', or randomize 'Next Random Exercise' in selected language.

Choose Exercise:
Timer 00:00
WPM --
Score --
Acc --
Correct chars --

Elixir Message Queue Consumer with Rate Limiting

Elixir

Goal -- WPM

Ready
Exercise Algorithm Area
1defmodule RateLimitedConsumer do
2use GenServer
3
4@max_messages_per_second 5
5@sleep_interval_ms 1000 # 1 second
6
7# Client API
8def start_link(queue_name) do
9GenServer.start_link(__MODULE__, queue_name, name: __MODULE__)
10end
11
12def stop() do
13GenServer.cast(__MODULE__, :stop)
14end
15
16# Server Callbacks
17@impl true
18def init(queue_name) do
19IO.puts("RateLimitedConsumer started. Processing up to #{@max_messages_per_second} messages per second.")
20# In a real scenario, this would connect to a message queue.
21# For simulation, we'll use a GenServer as a mock queue.
22{:ok, %{queue_name: queue_name, message_count_in_interval: 0, last_interval_start: :os.system_time(:milli_seconds)}}
23end
24
25@impl true
26def handle_cast(:stop, state) do
27IO.puts("RateLimitedConsumer stopping.")
28{:stop, :normal, state}
29end
30
31@impl true
32def handle_info(:process_next, state) do
33current_time = :os.system_time(:milli_seconds)
34time_elapsed = current_time - state.last_interval_start
35
36# Reset message count if a new second has started
37if time_elapsed >= @sleep_interval_ms do
38state = %{state | message_count_in_interval: 0, last_interval_start: current_time}
39end
40
41# Check if we can process another message
42if state.message_count_in_interval < @max_messages_per_second do
43# Simulate fetching a message from the queue
44case MockQueue.fetch_message(state.queue_name) do
45{:ok, message} ->
46IO.puts("Processing message: #{inspect(message)}")
47# Simulate message processing time
48Process.sleep(50) # Short sleep for processing
49
50new_state = %{state | message_count_in_interval: state.message_count_in_interval + 1}
51# Schedule the next message processing attempt
52send(self(), :process_next)
53{:noreply, new_state}
54:empty ->
55# Queue is empty, wait a bit before checking again
56Process.sleep(100)
57send(self(), :process_next)
58{:noreply, state}
59end
60else
61# Rate limit reached, wait until the next interval
62time_to_wait = @sleep_interval_ms - time_elapsed
63Process.sleep(max(0, time_to_wait))
64send(self(), :process_next) # Try again after waiting
65{:noreply, state}
66end
67end
68
69# Handle unexpected messages
70def handle_info(message, state) do
71IO.puts("Received unexpected message: #{inspect(message)}")
72{:noreply, state}
73end
74end
75
76defmodule MockQueue do
77use GenServer
78
79@max_messages 10
80
81def start_link(name) do
82GenServer.start_link(__MODULE__, name, name: name)
83end
84
85def add_message(queue_name, message) do
86GenServer.cast(queue_name, {:add, message})
87end
88
89def fetch_message(queue_name) do
90GenServer.call(queue_name, :fetch)
91end
92
93@impl true
94def init(name) do
95IO.puts("MockQueue '#{name}' started.")
96{:ok, %{name: name, messages: :queue.new(), size: 0}}
97end
98
99@impl true
100def handle_cast({:add, message}, state) do
101if state.size < @max_messages do
102new_queue = :queue.in(message, state.messages)
103new_state = %{state | messages: new_queue, size: state.size + 1}
104IO.puts("Message added to queue '#{state.name}'. Current size: #{new_state.size}")
105{:noreply, new_state}
106else
107IO.puts("Queue '#{state.name}' is full. Message not added.")
108{:noreply, state}
109end
110end
111
112@impl true
113def handle_call(:fetch, _from, state) do
114case :queue.out(state.messages) do
115{:empty, _} ->
116{:reply, :empty, state}
117{{:value, message}, new_queue} ->
118new_state = %{state | messages: new_queue, size: state.size - 1}
119{:reply, {:ok, message}, new_state}
120end
121end
122end
Algorithm description viewbox

Elixir Message Queue Consumer with Rate Limiting

Algorithm description:

This Elixir code simulates a message queue consumer with rate limiting. The `RateLimitedConsumer` GenServer fetches messages from a `MockQueue` and processes them, but it enforces a maximum number of messages processed per second. It uses `Process.sleep/1` to pause execution when the rate limit is reached or when the queue is empty. This is a common pattern for integrating with external services that have API rate limits or for managing resource consumption to prevent overwhelming downstream systems.

Algorithm explanation:

The `RateLimitedConsumer` GenServer manages the message processing loop. It maintains `message_count_in_interval` and `last_interval_start` to track messages processed within the current second. When `handle_info(:process_next)` is called, it first checks if a new second has begun and resets the counter if necessary. It then checks if the `message_count_in_interval` is less than `@max_messages_per_second`. If so, it attempts to fetch a message from the `MockQueue`. If a message is fetched, it's processed (with a small simulated delay), the count is incremented, and `:process_next` is sent to `self()` to continue the loop. If the queue is empty, it sleeps briefly before rescheduling `:process_next`. If the rate limit is reached, it calculates the remaining time in the current second, sleeps for that duration, and then reschedules `:process_next`. The `MockQueue` is a simple GenServer simulating a queue with a maximum capacity. The time complexity for processing a batch of messages within one second is roughly O(M), where M is `@max_messages_per_second`, assuming message processing is constant time. The overall throughput is limited by the rate limit. Space complexity is O(Q) for the queue, where Q is the maximum number of messages it can hold, and O(1) for the consumer's state.

Pseudocode:

Define `RateLimitedConsumer` (GenServer).
  Constants: `MAX_MESSAGES_PER_SECOND`, `SLEEP_INTERVAL_MS`.
  State: `queue_name`, `message_count_in_interval`, `last_interval_start`.
  In `init`:
    Initialize state variables.
    Schedule the first `:process_next` message.
  In `handle_info(:process_next)`:
    Get current time.
    Calculate time elapsed since `last_interval_start`.
    If `time_elapsed >= SLEEP_INTERVAL_MS`:
      Reset `message_count_in_interval` to 0.
      Update `last_interval_start` to current time.
    If `message_count_in_interval < MAX_MESSAGES_PER_SECOND`:
      Fetch a message from the queue.
      If message exists:
        Process the message (simulate work).
        Increment `message_count_in_interval`.
        Send `:process_next` to self().
      If queue is empty:
        Sleep briefly.
        Send `:process_next` to self().
    Else (rate limit reached):
      Calculate time to wait until next interval.
      Sleep for that duration.
      Send `:process_next` to self().
Define `MockQueue` (GenServer).
  Constants: `MAX_MESSAGES`.
  State: `messages` (a queue), `size`.
  Implement `add_message` (cast) and `fetch_message` (call).
  `fetch_message` returns `:empty` if queue is empty.