anchor.diff:      end,Regs),
   
       {QoS_0,QoS_Reliable} = lists:partition(fun({_,_,QoS}) -> QoS =:= ?QOS_0 end,RegStates),
  -    {MaybeLive,Dead} = lists:partition(fun({State,_,_QoS}) -> State =:= live end, QoS_Reliable),
  -    {QoS_0,MaybeLive,Dead}.
  +    {Live,Dead} = lists:partition(fun({State,_,_QoS}) -> State =:= live end, QoS_Reliable),
   
  -fwd_message({live,{Pid,ClientId},QoS},CTRPacket) ->
  +    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,{Pid,ClientId,QoS}}
  +        exit:{noproc, _} -> {noproc,Reg}
       end.
   
   persist_message(_CTRPacket) ->
  
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: -module(mqtt_router).
  -author("Kalin").
  
  -define(BATCH_SIZE,20).
  
  -include("mqtt_internal_msgs.hrl").
  
  %% API
  