How to retrieve the buffer on connection close ?

[Zetanova]

Is it possible to get all unpublished messages from the buffer after the connection gets closed (by the server) ?

I run into this problem when a component tried to publish a message on a not allowed subject with a shared connection.
The shared connection got closed server-side and it seams there is no why to retrieve/rescue buffered message.

[scottf]

I assume you mean core publishing versus JetStream since for JS you can use the PubAck to know if something is published. I suppose you could treat the library as internal and access the connection data structures and manually look through the queue.

But otherwise I'm not seeing a solution to your question.

[Zetanova]

Best location would be to extend the ClosedEventHandler event with the message-buffer.

The main issue is not that messages can be lost,
but that there is no mechanism to implement a full-reconnect and/or idle state.

Once the connection instance is successfully created and registered in an DI container.
then it gets very hard to safely replace it.

[scottf]

The client will try to reconnect to other servers in the cluster, and when it does it will send any messages in it's queue. It will also try to reconnect to single servers, but it might happen so fast, before the server is back up.

[Zetanova]

The connection state after a permission valuation is "closed"
It for a shared connection instance, its not ideal

I made an implantation with an IHostedService in the DI
to subscribe and request the lasted alive shared connection.

Its not ideal but better then be stuck with an closed connection.

readonly ConcurrentDictionary<string, IConnection> _connections = new (1, 8, StringComparer.OrdinalIgnoreCase);

Subject<(string Name, IConnection? Connection)> _subject = default!;

//...

public IConnection GetConnection(string? name = null)
{
    if (!_running)
        throw new InvalidOperationException("nats service not running");

    if (string.IsNullOrEmpty(name))
        name = "Default";

    return _connections.GetOrAdd(name, CreateConnection);
}

public IObservable<IConnection> GetLatestConnection(string? name = null)
{
    if (!_running)
        throw new InvalidOperationException("nats service not running");

    if (string.IsNullOrEmpty(name))
        name = "Default";

    var connection = GetConnection(name);

    return _subject.Where(n => StringComparer.OrdinalIgnoreCase.Equals(name, n.Name))
        .Select(n => n.Connection is null ? GetConnection(name) : n.Connection)
        .Append(connection)
        .DistinctUntilChanged();
}

private IConnection CreateConnection(string name)
{
    if (!_running)
        throw new InvalidOperationException("nats service not running");

    var opt = _optionsMonitor.Get(StringComparer.OrdinalIgnoreCase.Equals(name, "Default")
        ? Microsoft.Extensions.Options.Options.DefaultName
        : name);
    
    var connection = MyNatsUtils.CreateConnection(opt, o =>
    {
        o.Name = name;

        o.DisconnectedEventHandler += (s, a) =>
        {
            if(a.Error is not null)
                _logger.LogError(a.Error, "Connection[{name}] disconnected", a.Conn.Opts.Name);
            else if(_logger.IsEnabled(LogLevel.Debug))
                _logger.LogDebug("Connection[{name}] disconnected", a.Conn.Opts.Name);
        };

        o.ClosedEventHandler += (s, a) =>
        {
            var removed = _connections.TryRemove(new KeyValuePair<string, IConnection>(a.Conn.Opts.Name, a.Conn));

            if (a.Error is not null)
                _logger.LogError(a.Error, "Connection[{name}] closed", a.Conn.Opts.Name);
            else if (_logger.IsEnabled(LogLevel.Debug))
                _logger.LogDebug("Connection[{name}] closed", a.Conn.Opts.Name);

            if(removed)
                _subject.OnNext((name, null));
        };
    });

    _subject.OnNext((name, connection));
    
    return connection;
}

[scottf]

I think we need to start over. Can you elaborate on "a not allowed subject"?
Are you talking core publish or JetStream publish? How is the subject not allowed? Is it a security configuration, or is it no responders for core request, or is it just an invalid subject for JetStream

[scottf]

As far as DI, maybe don't inject the connection, maybe wrap the connection in a class with event awareness sort of a proxy or for publishing, you could even use something like a pool of connections.

[Zetanova]

Its fine, its only a logical misalignment with the nats IConnection behavior.

It behaves for the user as auto recovery, always active connection
until the server makes a force close.

Example would be on an `Error from processErr(): permissions violation for publish to "msg.notallowed.subject"
then the shared connection is for every closed and need to be recreated.

The users of the instance would need to take it into account before each publish.

Yes, everything can be resolved with one extra layer of abstraction.

But its not always a good idea, beside the normal IConnection
I would need then to abstract JS and KV and OS too, its never ending.

Better solution would be to "update" the nats.net client,
the implementation locks a bit off.

Improvments:

• Interface methods that could be implemented over extensions.
• A lot of locks that could be replaced with lock-free structure or a System.Threading.Channels
• buffer copying and, no span and memory support
• No ValueTask return for synchronies-like methods
• No Publishing of multiple messages (IEnumeration and IAsyncEnumeration)

At the end I tested the nats.net now for 2 weeks and there was no issue,
beside that I tried to write to an subject without permission "force close".

The performance does not really matter for a message rate under 200msg/s
But to have a height message truth put like for the golang-client,
the current client will be the bottleneck