%%
%% %CopyrightBegin%
%% 
%% Copyright Ericsson AB 2014-2017. All Rights Reserved.
%% 
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%% 
%% %CopyrightEnd%
%%

-module(message_queue_data_SUITE).

-export([all/0, suite/0]).
-export([basic/1, process_info_messages/1, total_heap_size/1]).

-export([basic_test/1]).

-include_lib("common_test/include/ct.hrl").

suite() ->
    [{ct_hooks,[ts_install_cth]},
     {timetrap, {minutes, 2}}].

all() -> 
    [basic, process_info_messages, total_heap_size].

%%
%%
%% Test cases
%%
%%

basic(Config) when is_list(Config) ->

    basic_test(erlang:system_info(message_queue_data)),

    {ok, Node1} = start_node(Config, "+hmqd off_heap"),
    ok = rpc:call(Node1, ?MODULE, basic_test, [off_heap]),
    stop_node(Node1),

    {ok, Node2} = start_node(Config, "+hmqd on_heap"),
    ok = rpc:call(Node2, ?MODULE, basic_test, [on_heap]),
    stop_node(Node2),

    ok.

is_valid_mqd_value(off_heap) ->
    true;
is_valid_mqd_value(on_heap) ->
    true;
is_valid_mqd_value(_) ->
    false.


basic_test(Default) ->

    Default = erlang:system_info(message_queue_data),
    true = is_valid_mqd_value(Default),

    {message_queue_data, Default} = process_info(self(), message_queue_data),
    Default = process_flag(message_queue_data, off_heap),
    {message_queue_data, off_heap} = process_info(self(), message_queue_data),
    off_heap = process_flag(message_queue_data, on_heap),
    {message_queue_data, on_heap} = process_info(self(), message_queue_data),
    {'EXIT', _} = (catch process_flag(message_queue_data, blupp)),

    P1 = spawn_opt(fun () -> receive after infinity -> ok end end,
		   [link]),
    {message_queue_data, Default} = process_info(P1, message_queue_data),
    unlink(P1),
    exit(P1, bye),

    P2 = spawn_opt(fun () -> receive after infinity -> ok end end,
		   [link, {message_queue_data, off_heap}]),
    {message_queue_data, off_heap} = process_info(P2, message_queue_data),
    unlink(P2),
    exit(P2, bye),

    P3 = spawn_opt(fun () -> receive after infinity -> ok end end,
		   [link, {message_queue_data, on_heap}]),
    {message_queue_data, on_heap} = process_info(P3, message_queue_data),
    unlink(P3),
    exit(P3, bye),

    {'EXIT', _} = (catch spawn_opt(fun () -> receive after infinity -> ok end end,
		   [link, {message_queue_data, blapp}])),

    ok.

process_info_messages(Config) when is_list(Config) ->
    Tester = self(),
    P1 = spawn_opt(fun () ->
			   receive after 500 -> ok end,
			   on_heap = process_flag(message_queue_data, off_heap),
			   Tester ! first,
			   receive after 500 -> ok end,
			   off_heap = process_flag(message_queue_data, on_heap),
			   Tester ! second,
			   receive after 500 -> ok end,
			   on_heap = process_flag(message_queue_data, off_heap),
			   Tester ! third,
			    
			   receive after infinity -> ok end
		   end,
		   [link, {message_queue_data, on_heap}]),
    
    P1 ! "A",
    receive first -> ok end,
    P1 ! "B",
    receive second -> ok end,
    P1 ! "C",
    receive third -> ok end,
    P1 ! "D",

    {messages, ["A", "B", "C", "D"]} = process_info(P1, messages),

    P2 = spawn_opt(fun () ->
			   receive after 500 -> ok end,
			   on_heap = process_flag(message_queue_data, off_heap),
			   Tester ! first,
			   receive after 500 -> ok end,
			   off_heap = process_flag(message_queue_data, on_heap),
			   Tester ! second,
			   receive after 500 -> ok end,
			   on_heap = process_flag(message_queue_data, off_heap),
			   Tester ! third,
			   receive after 500 -> ok end,

			   Tester ! process_info(self(), messages),

			   receive M1 -> M1 = "A" end,
			   receive M2 -> M2 = "B" end,
			   receive M3 -> M3 = "C" end,
			   receive M4 -> M4 = "D" end,

			   Tester ! self()
		   end,
		   [link, {message_queue_data, on_heap}]),

    P2 ! "A",
    receive first -> ok end,
    P2 ! "B",
    receive second -> ok end,
    P2 ! "C",
    receive third -> ok end,
    P2 ! "D",

    receive
	Msg ->
	    {messages, ["A", "B", "C", "D"]} = Msg
    end,
    
    receive P2 -> ok end,

    ok.

total_heap_size(_Config) ->

    Fun = fun F() -> receive Pid when is_pid(Pid) -> Pid ! ok,F() end end,

    %% Test that on_heap messages grow the heap even if they are not received
    OnPid = spawn_opt(Fun, [{message_queue_data, on_heap},link]),
    {total_heap_size, OnSize} = erlang:process_info(OnPid, total_heap_size),
    [OnPid ! lists:duplicate(N,N) || N <- lists:seq(1,100)],
    OnPid ! self(), receive ok -> ok end,
    {total_heap_size, OnSizeAfter} = erlang:process_info(OnPid, total_heap_size),
    ct:log("OnSize = ~p, OnSizeAfter = ~p",[OnSize, OnSizeAfter]),
    true = OnSize < OnSizeAfter,

    %% Test that off_heap messages do not grow the heap if they are not received
    OffPid = spawn_opt(Fun, [{message_queue_data, off_heap},link]),
    {total_heap_size, OffSize} = erlang:process_info(OffPid, total_heap_size),
    [OffPid ! lists:duplicate(N,N) || N <- lists:seq(1,100)],
    OffPid ! self(), receive ok -> ok end,
    {total_heap_size, OffSizeAfter} = erlang:process_info(OffPid, total_heap_size),
    ct:log("OffSize = ~p, OffSizeAfter = ~p",[OffSize, OffSizeAfter]),
    true = OffSize == OffSizeAfter.

%%
%%
%% helpers
%%
%%

start_node(Config, Opts) when is_list(Config), is_list(Opts) ->
    Pa = filename:dirname(code:which(?MODULE)),
    Name = list_to_atom(atom_to_list(?MODULE)
			++ "-"
			++ atom_to_list(proplists:get_value(testcase, Config))
			++ "-"
			++ integer_to_list(erlang:system_time(second))
			++ "-"
			++ integer_to_list(erlang:unique_integer([positive]))),
    test_server:start_node(Name, slave, [{args, Opts++" -pa "++Pa}]).

stop_node(Node) ->
    test_server:stop_node(Node).