anchor.diff:          end,
       {noreply, fin_stream(StreamRef, Fin, State3)};
   handle_info({gun_response, C, StreamRef, Fin, Status, _Headers},
  -        State=#state{pid={C,_}, streams=Streams}) ->
  +        State=#state{pid=C, streams=Streams}) ->
       if
           Status < 300 ->
               ok;
  
positive.code:             ok
      end,
      {noreply, fin_stream(StreamRef, Fin, State)};
  handle_info({gun_data, C, StreamRef, Fin, _Data}, State=#state{pid={C,_}}) ->
      {noreply, fin_stream(StreamRef, Fin, State)};
  
  handle_info({'DOWN', _MRef, process, C, Reason}, #state{conn=Conn, pid={C,_}}=State) ->
      lager:warning("Connector ~s failed: ~p", [Conn#connector.connid, Reason]),
      % restart the gun process
      {noreply, State#state{pid=create_gun(Conn), ready=false, streams=#{}}};
  handle_info(Unknown, State) ->
      lager:debug("Unknown message: ~p", [Unknown]),
      {noreply, State}.
  
  terminate(normal, #state{conn=#connector{connid=ConnId}, pid=Pid}) ->
      lager:debug("Connector ~s terminated: normal", [ConnId]),
      disconnect(Pid);
  terminate(Reason, #state{conn=#connector{connid=ConnId}, pid=Pid}) ->
      lager:warning("Connector ~s terminated: ~p", [ConnId, Reason]),
      disconnect(Pid).
  
  code_change(_OldVsn, State, _Extra) ->
      {ok, State}.
  
  ensure_connected(#state{ready=true}=State) ->
      {ok, State};
  ensure_connected(#state{conn=Conn, pid={ConnPid, MRef}, ready=false}=State) ->
      case gun:await_up(ConnPid, MRef) of
          {ok, _Protocol} ->
              {ok, State#state{ready=true}};
          {error, Reason} ->
              lager:debug("~s failed to connect: ~p", [Conn#connector.connid, Reason]),
              {error, State}
      end;
  ensure_connected(#state{pid=undefined}=State) ->
      {error, State}.
  
  disconnect(undefined) ->
      ok;
  disconnect({ConnPid, MRef}) ->
      demonitor(MRef),
      gun:close(ConnPid).
  
  handle_uplink(Vars0, State) when is_list(Vars0) ->
  
negative.code:             do_publish({URI, [User, Pass], ContentType, Body}, [], State)
      end.
  
  do_publish({URI, _Auth, ContentType, Body}=Msg, Headers, State=#state{pid=C, streams=Streams}) ->
      StreamRef = gun:post(C, URI,
          [{<<"content-type">>, ContentType} | Headers], Body),
      State#state{streams=maps:put(StreamRef, Msg, Streams)}.
  