anchor.diff:       end
           || {ClientId,SubQoS} <- mqtt_sub_repo:get_matches(Topic)].
   
  -get_regs_by_state(Regs,MsgQoS) ->
  +split_regs_by_state(Regs,MsgQoS) ->
       RegStates = lists:filtermap(fun({ClientId,SubQoS,Reg}) ->
           QoS = min(MsgQoS,SubQoS),
           case Reg of
  
positive.code: -module(mqtt_router).
  -author("Kalin").
  
  -include("mqtt_internal_msgs.hrl").
  
  %% API
  
negative.code:     mqtt_topic_repo:enqueue(Topic,Msg),
      Regs = get_client_regs(Topic),
  
      {QoS_0,Live,Dead} = split_regs_by_state(Regs,MsgQoS),
  
      error_logger:info_msg("To enqueue ~p for topic ~p",[Msg,Topic]),
      CTRPacket = {Topic,Content,Seq},
      % Send out QoS messages, do NOT wait for response
      [cast_msg(NodeRegs ,CTRPacket) || NodeRegs <- QoS_0],
      %% Send out QoS 1/2 messages to registered processes and wait for response
      SyncResults = lists:flatten([call_msg(NodeRegs ,CTRPacket) || NodeRegs  <- Live]),
  
      %% Handle results
      %% @todo: Fault tolerance
      FailedRegs = [Reg || {error,_Reason,Reg} <- SyncResults],
      case FailedRegs of
          []  -> ok;
          _   -> error({failed_delivery,FailedRegs})
      end,
      StaleRegs = [Tuple || {noproc,Tuple} <- SyncResults],
      %% Get rid of any stale process registrations
      lists:map(fun({Pid,ClientId,_}) -> mqtt_reg_repo:unregister(Pid,ClientId) end,
                StaleRegs),
  
      %% persist if there are any subscribed clients that did not receive the message
      case {StaleRegs,Dead} of
          {[],[]} -> ok;
          _       -> persist_message(CTRPacket)
      end.
  
  
  -spec get_client_regs(binary()) ->
      [{ClientId::binary(),
        SubQos :: qos(),
        Reg :: {ok,Pid::pid()} | undefined}].
  
  %% Get the Pids of connected clients matching the topic
  get_client_regs(Topic) ->
      [begin
          Reg = mqtt_reg_repo:get_registration(ClientId),
  