Repository navigation
Add connectAsynchronously to Options allowing a nats connection initialization that does not block. - #33
Add connectAsynchronously to Options allowing a nats connection initialization that does not block.#33nviliunov-evolution-throwaway wants to merge 2 commits into
connectAsynchronously to Options allowing a nats connection initialization that does not block.#33Conversation
|
Important
This repository does not receive automatic reviews because it has fewer than 10 stars. ⚙️ Run configuration
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
…nitialization that does not block.
c948f9e to
200f542
Compare
| def connectUntilEstablished(javaConnection: Deferred[F, WrappedConnection[F]]): F[Unit] = | ||
| Async[F] | ||
| .uncancelable { _ => | ||
| Async[F].blocking(JNats.connect(javaOptions)).flatMap(c => javaConnection.complete(new WrappedConnection[F](c, options)).void) |
There was a problem hiding this comment.
Will reconnectOnConnect be ignored?
There was a problem hiding this comment.
Yes, reconnectOnConnect is used only when the connection is synchronous
| val wrappedConnection = connection match { | ||
| case wrapped: com.evolution.natseffect.impl.WrappedConnection[F] => wrapped | ||
| case _: com.evolution.natseffect.impl.AsyncConnection[F] => | ||
| throw new IllegalArgumentException("JetStream is not supported on a connection created with Options.connectAsynchronously") |
There was a problem hiding this comment.
The whole point of connectAsynchronously is that the app starts even when NATS is down. But JetStream.make now throws on an AsyncConnection, so anyone using JetStream or KeyValue crashes on startup the moment they turn the option on, broker up or not.
|
|
||
| override def flush(timeout: Duration): F[Unit] = withConnection(_.flush(timeout)) | ||
|
|
||
| override def status: F[Status] = withConnection(_.status) |
There was a problem hiding this comment.
Don't wait for the connection in this one method. Peek at the Deferred with tryGet. If it's empty, return Status.CONNECTING
| .uncancelable { _ => | ||
| Async[F].blocking(JNats.connect(javaOptions)).flatMap(c => javaConnection.complete(new WrappedConnection[F](c, options)).void) | ||
| } | ||
| .handleErrorWith(_ => Async[F].sleep(options.reconnectWait) >> connectUntilEstablished(javaConnection)) |
There was a problem hiding this comment.
All connect errors swallowed, retried forever. The handleErrorWith(_ => ...) drops AuthenticationException and misconfiguration errors alike. With bad credentials the fiber loops indefinitely and callers only ever see the generic NotEstablishedException, with no log or callback exposing the cause.
A conceptual replacement for jnats'
Nats.connectAsynchronously. API changes from the Java library:F[F[Connection[F]]]signature, but it makes things less ergonomic, as the easy path is toflattenthe result, which defeats the original purpose.Threadthat does connection initialization. I know that jnats is liberal with its usage of threads, but we can easily avoid this particular one by doing it in a cats-effect fiber.reconnectOnConnect, while I always retry a connection. The retry is done in cats-effect instead of jnats, providing cancellation windows between attempt rounds.