"Cleaning" the message queue

[Luke Gorrie]

"Vlad Dumitrescu" <vladdu@REDACTED> writes:

> Hi all,
> 
> I have an application where processes get messages posted quite often,
> in a periodical fashion. Sometimes they don't have time to finish
> processing one message before another drops in, so the message queue
> grows longer and longer.
> 
> What I'd want is that in case there are several such messages waiting,
> only the latest shall be processed, while the rest shall be
> dumped. (The messages are "virtual clock ticks")

Here's an example program with a "receiver" process that gets 'go'
messages, and discards duplicates that pile up on its message
queue. Hopefully its pretty similar to your situation.

Since I have no information in my "go" messages I'm not actually using
the "latest" one, but that'd be an easy change.

--- SNIP ---

-module(receive_one).

-export([test/0]).

-export([receiver/0]).                  % internal

%% Test: We create a "receiver" process that will handle some 'go'
%% messages. We expect it to only actually handle two of them and
%% discard the rest, due to the timings!
%%
%% So we expect the string "Going!" should be io:format'd just twice.
test() ->
    Receiver = spawn_link(?MODULE, receiver, []),
    Receiver ! go,
    %% After this wait, the receiver will be busy processing the first
    %% 'go', allowing the next ones to accumulate in its message queue
    wait(5),
    Receiver ! go,
    Receiver ! go,
    Receiver ! go,
    Receiver ! go.

receiver() ->
    receive_go(),
    process(),
    receiver().

%% Process a 'go' message. This takes some time.
process() ->
    io:format("Going!~n"),
    %% We spend 500ms "processing" each message, so that there's time
    %% for new ones to accumulate on our message queue.
    wait(500).

%% Wait to receive a 'go' message, and flush any extra ones.
receive_go() ->
    receive
        go ->
            flush_go_msgs(),
            ok
    end.

flush_go_msgs() ->
    %% Receive/discard go messages until we timeout with 'after 0' -
    %% i.e. until no go messages are left on message queue
    receive
        go -> flush_go_msgs()
    after
        0 -> no_more_gos
    end.

wait(Time) ->
    receive
        i_never_get_this -> exit(impossible)
    after
        Time -> ok
    end.