fix: reconnect should use both reconnect and retry_initial for reconnection

[jakub-hess]

Current behaviour

In the case where the SSE stream is available on connect and breaks mid-connection (e.g. because of server restart or network failure) and when polled in loop as such:

let es = es::ClientBuilder::for_url(
    url,
)?
.reconnect(
    es::ReconnectOptions::reconnect(true)
        .delay(Duration::from_millis(500))
        .delay_max(Duration::from_secs(20))
        .retry_initial(false)
        .build(),
)
.build();

let stream = es.stream();
loop {
      let result = stream.next().await {}
}

The Client goes through following State transitions New -> Connecting -> Connected -> WaitingToReconnect -> New -> Connecting -> New -> Connecting -> New -> Connecting -> ....

It never waits for the delay_duration (this causes reconnects happening in about 1 ms rhythm), since it never goes to WaitingForReconnect again.

Cause

In client.rs on line 457

let retry = self.props.reconnect_opts.retry_initial;

this is the retry flag, that is used further down on line 524, and it is always set to retry_initial, regardless of whether this is the initial connection, or reconnect of already established stream.

Err(e) => {
      // This seems basically impossible. AFAIK we can only get this way if we
      // poll after it was already ready
      warn!("request returned an error: {}", e);
      if !*retry {
          self.as_mut().project().state.set(State::New);
          return Poll::Ready(Some(Err(Error::HttpStream(Box::new(e)))));
      }

This error condition happens, when the server is still not ready to accept the connection, and therefore in my case this causes the stream to loop indefinitely (or break after first reconnect attempt depending on how i handle the errors)

Solution

Introduce a initial_connection flag, that is set to false once the Client reaches the Connected, and the in New state the reconnect or retry initial flag is used depending on the state of the initial_connection flag

[keelerm84]

@jakub-hess apologies for the delay in addressing this PR.

The provided code snippet you included, quoted again below, is not a supported use case.

let stream = es.stream();
loop {
      let result = stream.next().await {}
}

If you look at the tail example in this project, you will see an example of the usage we support.

The client is not intended to be consumed once an error has returned. Once an error is returned, the client is considered defunct and a new one would need to be instantiated.

[jakub-hess]

Yes, I understand that, I didn't mention the fact that the inside of the loop has been abbreviated.

Even if I break on error my main point still stands, the reconnection is driven by retry_initial instead of reconnect. If reconnect is true and retry_initial is false and the server drops the connection (e.g. because of restart) I get an HTTP Stream Error instead of the client trying to reconnect based on its reconnection settings.