-module(mqueue).

-export([timed_run/3, run/3]).

producer(0, ConsumerPid, _WithAck) ->
    ConsumerPid ! {done, self()},
    receive
        done -> ok
    end;
producer(N, ConsumerPid, WithAck = false) ->
    ConsumerPid ! {msg, self()},
    producer(N-1, ConsumerPid, WithAck);
producer(N, ConsumerPid, WithAck = true) ->
    ConsumerPid ! {acked_msg, self()},
    receive
        ack -> ok
    end,
    producer(N-1, ConsumerPid, WithAck).

consumer(M, EchoPid) ->
    receive
        {msg, _From} ->
            call_echo(M, EchoPid),
            consumer(M, EchoPid);
        {acked_msg, From} ->
            From ! ack,
            call_echo(M, EchoPid),
            consumer(M, EchoPid);
        {done, From} ->
            EchoPid ! done,
            From ! done
    end.

call_echo(0, _EchoPid) ->
    ok;
call_echo(M, EchoPid) ->
    EchoPid ! {hello, self()},
    receive
        hello -> call_echo(M-1, EchoPid)
    end.

echo() ->
    receive
        {Msg, From} ->
            From ! Msg,
            echo();
        done -> ok
    end.

run(N, M, WithAck) ->
    EchoPid     = spawn_link(fun echo/0),
    ConsumerPid = spawn_link(
                    fun () ->
                            consumer(M, EchoPid) end),
    producer(N, ConsumerPid, WithAck).

timed_run(N, M, WithAck) ->
    timer:tc(?MODULE, run, [N, M, WithAck]).