anchor.diff:      publish(Nodes, NrOfProcesses, NrOfMsgsPerProcess),
       done = receive_times(done, NrOfProcesses),
       ?_assertEqual(true, check_unique_client("connect-multiple", Nodes)).
  -
  -publish(Nodes, NrOfProcesses, NrOfMsgsPerProcess) ->
  -    publish(self(), Nodes, NrOfProcesses, NrOfMsgsPerProcess, []).
  -publish(_, _, 0, _, Pids) -> Pids;
  -publish(Self, Nodes, NrOfProcesses, NrOfMsgsPerProcess, Pids) ->
  -    Pid = spawn_link(fun() -> publish_(Self, Nodes, NrOfMsgsPerProcess) end),
  -    publish(Self, Nodes, NrOfProcesses -1, NrOfMsgsPerProcess, [Pid|Pids]).
  -
  -publish_(Self, Nodes, NrOfMsgsPerProcess) ->
  -    {A, B, C} =  now(),
  -    random:seed(A, B, C),
  -    publish__(Self, Nodes, NrOfMsgsPerProcess).
  -publish__(Self, _, 0) ->
  -    Self ! done;
  -publish__(Self, Nodes, NrOfMsgsPerProcess) ->
  -    Connect = packet:gen_connect("connect-multiple", [{keepalive, 10}]),
  -    Connack = packet:gen_connack(0),
  -    {ok, _Socket} = packet:do_client_connect(Connect, Connack, opts(Nodes)),
  -    check_unique_client("connect-multiple", Nodes),
  -    timer:sleep(random:uniform(100)),
  -    publish__(Self, Nodes, NrOfMsgsPerProcess - 1).
  -
  -publish_random(Nodes, N, Topic) ->
  -    publish_random(Nodes, N, Topic, []).
  -
  -publish_random(_, 0, _, Acc) -> Acc;
  -publish_random(Nodes, N, Topic, Acc) ->
  -    Connect = packet:gen_connect("connect-unclean-pub", [{clean_session, true},
  -                                                           {keepalive, 10}]),
  -    Connack = packet:gen_connack(0),
  -    Payload = crypto:rand_bytes(random:uniform(10000)),
  -    Publish = packet:gen_publish(Topic, 1, Payload, [{mid, 1}]),
  -    Puback = packet:gen_puback(1),
  -    Disconnect = packet:gen_disconnect(),
  -    {ok, Socket} = packet:do_client_connect(Connect, Connack, opts(Nodes)),
  -    ok = gen_tcp:send(Socket, Publish),
  -    ok = packet:expect_packet(Socket, "puback", Puback),
  -    ok = gen_tcp:send(Socket, Disconnect),
  -    publish_random(Nodes, N - 1, Topic, [Payload|Acc]).
  -
  -receive_publishes(_, _, []) -> ok;
  -receive_publishes(Nodes, Topic, Payloads) ->
  -    Connect = packet:gen_connect("connect-unclean", [{clean_session, false},
  -                                                           {keepalive, 10}]),
  -    Connack = packet:gen_connack(0),
  -    Disconnect = packet:gen_disconnect(),
  -    {ok, Socket} = packet:do_client_connect(Connect, Connack, opts(Nodes)),
  -    case recv(Socket, emqtt_frame:initial_state()) of
  -        {ok, #mqtt_frame{variable= #mqtt_frame_publish{
  -                                      message_id=MsgId
  -                                     },
  -                         payload=Payload}} ->
  -            ok = gen_tcp:send(Socket, packet:gen_puback(MsgId)),
  -            ok = gen_tcp:send(Socket, Disconnect),
  -            io:format(user, "+", []),
  -            receive_publishes(Nodes, Topic, Payloads -- [Payload]);
  -        E ->
  -            io:format(user, "got other ~p~n", [E]),
  -            gen_tcp:send(Socket, Disconnect),
  -            gen_tcp:close(Socket),
  -            receive_publishes(Nodes, Topic, Payloads)
  -    end.
  -
  -recv(Socket, ParserState) ->
  -    case gen_tcp:recv(Socket, 0) of
  -        {ok, Data} ->
  -            case emqtt_frame:parse(Data, ParserState) of
  -                {ok, F, _} -> {ok, F};
  -                {more, NewParserState} ->
  -                    recv(Socket, NewParserState);
  -                {error, Reason} -> {error, Reason}
  -            end;
  -        {error, Reason} -> {error, Reason}
  -    end.
  -
   
   multiple_connect_unclean(Nodes) ->
       Topic = "qos1/multiple/test",
  
positive.code: cluster_test_() ->
      [
       ?SETUP(fun multiple_connect/1),
       ?SETUP(fun multiple_connect_unclean/1)
      ].
  
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  
negative.code:     io:format(user, "!!!!!!!!!!!!!!!!!!! port_count ~p~n", [Ports()]),
      io:format(user, "!!!!!!!!!!!!!!!!!!! process_count ~p~n", [Procs()]),
      ?_assertEqual(true, true).
  
  distributed_subscribe(Nodes) ->
      Topic = "qos1/distributed/test",
      Sockets =
      [begin
           Connect = packet:gen_connect("connect-" ++ integer_to_list(Port),
                                        [{clean_session, true},
                                         {keepalive, 10}]),
           Connack = packet:gen_connack(0),
           Subscribe = packet:gen_subscribe(123, Topic, 1),
           Suback = packet:gen_suback(123, 1),
           {ok, Socket} = packet:do_client_connect(Connect, Connack, [{port, Port}]),
           ok = gen_tcp:send(Socket, Subscribe),
           ok = packet:expect_packet(Socket, "suback", Suback),
           Socket
       end || {_, Port} <- Nodes],
      [PubSocket|Rest] = Sockets,
      Publish = packet:gen_publish(Topic, 1, <<"test-message">>, [{mid, 1}]),
      Puback = packet:gen_puback(1),
      ok = gen_tcp:send(PubSocket, Publish),
      ok = packet:expect_packet(PubSocket, "puback", Puback),
      _ = [begin
               ok = packet:expect_packet(Socket, "publish", Publish),
               ok = gen_tcp:send(Socket, Puback),
               gen_tcp:close(Socket)
           end || Socket <- Rest],
      ?_assertEqual(true, true).
  
  
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  %%% Internal
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  
  publish(Nodes, NrOfProcesses, NrOfMsgsPerProcess) ->
      publish(self(), Nodes, NrOfProcesses, NrOfMsgsPerProcess, []).
  publish(_, _, 0, _, Pids) -> Pids;
  publish(Self, Nodes, NrOfProcesses, NrOfMsgsPerProcess, Pids) ->
      Pid = spawn_link(fun() -> publish_(Self, Nodes, NrOfMsgsPerProcess) end),
      publish(Self, Nodes, NrOfProcesses -1, NrOfMsgsPerProcess, [Pid|Pids]).
  
  publish_(Self, Nodes, NrOfMsgsPerProcess) ->
      {A, B, C} =  now(),
      random:seed(A, B, C),
      publish__(Self, Nodes, NrOfMsgsPerProcess).
  publish__(Self, _, 0) ->
      Self ! done;
  publish__(Self, Nodes, NrOfMsgsPerProcess) ->
      Connect = packet:gen_connect("connect-multiple", [{keepalive, 10}]),
      Connack = packet:gen_connack(0),
      {ok, _Socket} = packet:do_client_connect(Connect, Connack, opts(Nodes)),
      check_unique_client("connect-multiple", Nodes),
      timer:sleep(random:uniform(100)),
      publish__(Self, Nodes, NrOfMsgsPerProcess - 1).
  
  publish_random(Nodes, N, Topic) ->
      publish_random(Nodes, N, Topic, []).
  
  publish_random(_, 0, _, Acc) -> Acc;
  publish_random(Nodes, N, Topic, Acc) ->
      Connect = packet:gen_connect("connect-unclean-pub", [{clean_session, true},
                                                             {keepalive, 10}]),
      Connack = packet:gen_connack(0),
      Payload = crypto:rand_bytes(random:uniform(10000)),
      Publish = packet:gen_publish(Topic, 1, Payload, [{mid, 1}]),
      Puback = packet:gen_puback(1),
      Disconnect = packet:gen_disconnect(),
      {ok, Socket} = packet:do_client_connect(Connect, Connack, opts(Nodes)),
      ok = gen_tcp:send(Socket, Publish),
      ok = packet:expect_packet(Socket, "puback", Puback),
      ok = gen_tcp:send(Socket, Disconnect),
      publish_random(Nodes, N - 1, Topic, [Payload|Acc]).
  
  receive_publishes(_, _, []) -> ok;
  receive_publishes(Nodes, Topic, Payloads) ->
      Connect = packet:gen_connect("connect-unclean", [{clean_session, false},
                                                             {keepalive, 10}]),
      Connack = packet:gen_connack(0),
      Disconnect = packet:gen_disconnect(),
      {ok, Socket} = packet:do_client_connect(Connect, Connack, opts(Nodes)),
      case recv(Socket, emqtt_frame:initial_state()) of
          {ok, #mqtt_frame{variable= #mqtt_frame_publish{
                                        message_id=MsgId
                                       },
                           payload=Payload}} ->
              ok = gen_tcp:send(Socket, packet:gen_puback(MsgId)),
              ok = gen_tcp:send(Socket, Disconnect),
              io:format(user, "+", []),
              receive_publishes(Nodes, Topic, Payloads -- [Payload]);
          E ->
              io:format(user, "got other ~p~n", [E]),
              gen_tcp:send(Socket, Disconnect),
              gen_tcp:close(Socket),
              receive_publishes(Nodes, Topic, Payloads)
      end.
  
  recv(Socket, ParserState) ->
      case gen_tcp:recv(Socket, 0) of
          {ok, Data} ->
              case emqtt_frame:parse(Data, ParserState) of
                  {ok, F, _} -> {ok, F};
                  {more, NewParserState} ->
                      recv(Socket, NewParserState);
                  {error, Reason} -> {error, Reason}
              end;
          {error, Reason} -> {error, Reason}
      end.
  
  
  
  
  