anchor.diff:  -module(mqtt_router).
   -author("Kalin").
   
  +-define(BATCH_SIZE,20).
  +
   -include("mqtt_internal_msgs.hrl").
   
   %% API
  
positive.code:      end
          || {ClientId,SubQoS} <- mqtt_sub_repo:get_matches(Topic)].
  
  get_regs_by_state(Regs,MsgQoS) ->
      RegStates = lists:filtermap(fun({ClientId,SubQoS,Reg}) ->
          QoS = min(MsgQoS,SubQoS),
          case Reg of
  
negative.code:     end,Regs),
  
      {QoS_0,QoS_Reliable} = lists:partition(fun({_,_,QoS}) -> QoS =:= ?QOS_0 end,RegStates),
      {Live,Dead} = lists:partition(fun({State,_,_QoS}) -> State =:= live end, QoS_Reliable),
  
      QoS_0_PerNode = batch_up(QoS_0),
      Live_PerNode = batch_up(Live),
      {QoS_0_PerNode,Live_PerNode,Dead}.
  
  batch_up(Regs) ->
      RegsPerNode = group_by_node(Regs),
      lists:flatmap(fun({Node,NodeRegs}) ->
                      Batches = split_into_batches(NodeRegs,?BATCH_SIZE),
                      lists:map(fun(Batch) -> {Node,Batch} end,Batches)
                    end,
          RegsPerNode).
  
  split_into_batches(L,Len) ->
      split_into_batches(L,Len,[]).
  
  split_into_batches(L,Len,B) when length(L) > Len ->
      {H,T} = lists:split(L,Len),
      split_into_batches(T,Len,[H|B]);
  
  split_into_batches(L,Len,B) when length(L) =< Len ->
      [L|B].
  
  group_by_node(Regs) ->
      NodeRegs = [{node(Pid), Reg} || Reg = {Pid,_,_} <- Regs],
      Groups = lists:foldr(fun({K,V}, D) -> dict:append(K, V, D) end, dict:new(), NodeRegs),
      dict:to_list(Groups).
  
  cast_msg({Node,Regs},CTRPacket) ->
      rpc:cast(Node,?MODULE,cast_msg_local,[Regs,CTRPacket]).
  
  cast_msg_local(Regs,CTRPacket) ->
      [mqtt_session_out:push_qos0(Pid,CTRPacket) || {Pid,_,_} <- Regs].
  
  call_msg({Node,Regs},CTRPacket) ->
      Results = rpc:call(Node,?MODULE,call_msg_local,[Regs,CTRPacket]),
      [ case Result of
            {badrpc,Reason} -> {error,Reason,Regs};
            _ -> Result
        end
      || Result <- Results].
  
  call_msg_local(Regs,CTRPacket) ->
      rpc:pmap({?MODULE,fwd_message},[CTRPacket],Regs).
  
  fwd_message(Reg = {Pid,_,QoS},CTRPacket) ->
      %% @todo: use basic messaging (instead of depending on the specifics of gen_server:call)
      try mqtt_session_out:push_reliable(Pid,CTRPacket,QoS) of
          _ -> ok
      catch
          exit:{noproc, _} -> {noproc,Reg}
      end.
  
  persist_message(_CTRPacket) ->
  