anchor.diff:          when PatPub == undefined; PatPub == ?EMPTY_PATTERN ->
       {noreply, State};
   handle_info({uplink, _Node, Vars0}, #state{conn=Conn}=State) ->
  -    case ensure_connected(State) of
  +    case ensure_connected(ensure_gun(State)) of
           {ok, State2} ->
               {noreply, handle_uplink(Vars0, State2)};
           {error, State2} ->
  
positive.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)}.
  
negative.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]),
      {noreply, State#state{pid=undefined}};
  handle_info(Unknown, State) ->
      lager:debug("Unknown message: ~p", [Unknown]),
      {noreply, State}.
  
  terminate(normal, #state{conn=#connector{connid=ConnId}, pid=C}) ->
      lager:debug("Connector ~s terminated: normal", [ConnId]),
      disconnect(C);
  terminate(Reason, #state{conn=#connector{connid=ConnId}, pid=C}) ->
      lager:warning("Connector ~s terminated: ~p", [ConnId, Reason]),
      disconnect(C).
  
  code_change(_OldVsn, State, _Extra) ->
      {ok, State}.
  
  ensure_gun(#state{pid=Pid}=State) when is_pid(Pid) ->
      % is running
      State;
  ensure_gun(#state{conn=#connector{uri= <<"http:">>}, pid=undefined}=State) ->
      % should not be running
      State;
  ensure_gun(#state{conn=#connector{connid=ConnId, uri=Uri}, pid=undefined}=State) ->
      lager:debug("Connecting ~s to ~s", [ConnId, Uri]),
      {ok, ConnPid} =
          case http_uri:parse(binary_to_list(Uri), [{scheme_defaults, [{http, 80}, {https, 443}]}]) of
              {ok, {http, _UserInfo, HostName, Port, _Path, _Query}} ->
                  gun:open(HostName, Port);
              {ok, {https, _UserInfo, HostName, Port, _Path, _Query}} ->
                  gun:open(HostName, Port, #{transport=>ssl})
          end,
      MRef = monitor(process, ConnPid),
      State#state{pid=ConnPid, mref=MRef, ready=false, streams=#{}}.
  
  ensure_connected(#state{ready=true}=State) ->
      {ok, State};
  ensure_connected(#state{pid=undefined, ready=false}=State) ->
      {error, State};
  ensure_connected(#state{conn=Conn, pid=ConnPid, mref=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.
  
  disconnect(undefined) ->
      ok;
  disconnect(ConnPid) ->
      gun:close(ConnPid).
  
  handle_uplink(Vars0, State) when is_list(Vars0) ->
  