anchor.diff:              State
       end.
   
  +internal_flush(#state{reachable=false} = State) -> State;
   internal_flush(#state{pending=[]} = State) -> State;
   internal_flush(#state{pending=Pending, node=Node, transport=Transport,
  -                      reconnect_tref=TRef, socket=Socket, bytes_send={{M, S, _}, V}} = State) ->
  +                      socket=Socket, bytes_send={{M, S, _}, V}} = State) ->
       L = iolist_size(Pending),
       Msg = [<<"vmq-send", L:32>>|lists:reverse(Pending)],
       case Transport:send(Socket, Msg) of
  
positive.code:                     Msg = [<<"vmq-connect">>, <<L:32, NodeName/binary>>],
                      case Transport:send(Socket, Msg) of
                          ok ->
                              self() ! remonitor, %% let remonitor decide if we are reachable
                              State#state{socket=Socket, transport=Transport};
                          {error, Reason} ->
                              lager:warning("can't initiate connect to cluster node ~p due to ~p", [RemoteNode, Reason]),
                              State#state{reachable=false}
                      end;
                  {error, Reason} ->
                      lager:warning("can't connect to cluster node ~p due to ~p", [RemoteNode, Reason]),
                      TRef = reconnect_timer(),
                      State#state{reachable=false, reconnect_tref=TRef}
              end;
          {badrpc, nodedown} ->
              %% we don't scream.. vmq_cluster_mon screams
  
negative.code:             %% we're already reconnecting
              TRef
      end,
      State#state{reachable=false, reconnect_tref=NewTRef};
  handle_message(reconnect, #state{reachable=false} = State) ->
      connect(State#state{reconnect_tref=undefined});
  handle_message(Msg, #state{node=Node, reachable=Reachable} = State) ->
  