Added Additional Try Commands to KVStore

src/NATS.Client.Core/NatsResult.cs

@@ -2,6 +2,34 @@
 
 namespace NATS.Client.Core;
 
+public readonly struct NatsResult
+{
+    private readonly Exception? _error;
+
+    public NatsResult()
+    {
+        _error = null;
+    }
+
+    public NatsResult(Exception error)
+    {
+        _error = error;
+    }
+
+    public static NatsResult Default => new();
+
+    public Exception Error => _error ?? ThrowErrorIsNotSetException();
+
+    public bool Success => _error == null;
+
+    public static implicit operator NatsResult(Exception error) => new(error);
+
+    private static Exception ThrowErrorIsNotSetException() => throw CreateInvalidOperationException("Result error is not set");
+
+    [MethodImpl(MethodImplOptions.NoInlining)]
+    private static Exception CreateInvalidOperationException(string message) => new InvalidOperationException(message);
+}
+
 public readonly struct NatsResult<T>
 {
     private readonly T? _value;

src/NATS.Client.JetStream/INatsJSContext.cs

@@ -196,6 +196,39 @@ ValueTask<PubAckResponse> PublishAsync<T>(
         NatsHeaders? headers = default,
         CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Tries to send data to a stream associated with the subject.
+    /// </summary>
+    /// <param name="subject">Subject to publish the data to.</param>
+    /// <param name="data">Data to publish.</param>
+    /// <param name="serializer">Serializer to use for the message type.</param>
+    /// <param name="opts">Publish options.</param>
+    /// <param name="headers">Optional message headers.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the publishing call or the wait for response.</param>
+    /// <typeparam name="T">Type of the data being sent.</typeparam>
+    /// <returns>
+    /// The ACK response to indicate if stream accepted the message as well as additional
+    /// information like the sequence number of the message stored by the stream.
+    /// </returns>
+    /// <exception cref="NatsJSException">There was a problem receiving the response.</exception>
+    /// <remarks>
+    /// <para>
+    /// Use this method to avoid exceptions.
+    /// </para>
+    /// <para>
+    /// Note that if the subject isn't backed by a stream or the connected NATS server
+    /// isn't running with JetStream enabled, this call will hang waiting for an ACK
+    /// until the request times out.
+    /// </para>
+    /// </remarks>
+    ValueTask<NatsResult<PubAckResponse>> TryPublishAsync<T>(
+        string subject,
+        T? data,
+        INatsSerialize<T>? serializer = default,
+        NatsJSPubOpts? opts = default,
+        NatsHeaders? headers = default,
+        CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Creates a new stream if it doesn't exist or returns an existing stream with the same name.
     /// </summary>

src/NATS.Client.JetStream/NatsJSContext.cs

@@ -81,6 +81,53 @@ public async ValueTask<PubAckResponse> PublishAsync<T>(
         NatsJSPubOpts? opts = default,
         NatsHeaders? headers = default,
         CancellationToken cancellationToken = default)
+    {
+        var result = await TryPublishAsync(subject, data, serializer, opts, headers, cancellationToken);
+        if (!result.Success)
+        {
+            throw result.Error;
+        }
+
+        return result.Value;
+    }
+
+    /// <summary>
+    /// Sends data to a stream associated with the subject.
+    /// </summary>
+    /// <param name="subject">Subject to publish the data to.</param>
+    /// <param name="data">Data to publish.</param>
+    /// <param name="serializer">Serializer to use for the message type.</param>
+    /// <param name="opts">Publish options.</param>
+    /// <param name="headers">Optional message headers.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the publishing call or the wait for response.</param>
+    /// <typeparam name="T">Type of the data being sent.</typeparam>
+    /// <returns>
+    /// The ACK response to indicate if stream accepted the message as well as additional
+    /// information like the sequence number of the message stored by the stream.
+    /// </returns>
+    /// <exception cref="NatsJSException">There was a problem receiving the response.</exception>
+    /// <remarks>
+    /// <para>
+    /// Use this method to avoid exceptions.
+    /// </para>
+    /// <para>
+    /// Note that if the subject isn't backed by a stream or the connected NATS server
+    /// isn't running with JetStream enabled, this call will hang waiting for an ACK
+    /// until the request times out.
+    /// </para>
+    /// <para>
+    /// By setting <c>msgId</c> you can ensure messages written to a stream only once. JetStream support idempotent
+    /// message writes by ignoring duplicate messages as indicated by the Nats-Msg-Id header. If both <c>msgId</c>
+    /// and the <c>Nats-Msg-Id</c> header value was set, <c>msgId</c> parameter value will be used.
+    /// </para>
+    /// </remarks>
+    public async ValueTask<NatsResult<PubAckResponse>> TryPublishAsync<T>(
+        string subject,
+        T? data,
+        INatsSerialize<T>? serializer = default,
+        NatsJSPubOpts? opts = default,
+        NatsHeaders? headers = default,
+        CancellationToken cancellationToken = default)
     {
         if (opts != null)
         {
@@ -136,28 +183,25 @@ public async ValueTask<PubAckResponse> PublishAsync<T>(
                         // without the timeout the publish call will hang forever since the server
                         // which received the request won't be there to respond anymore.
                         Timeout = Connection.Opts.RequestTimeout,
-
-                        // If JetStream is disabled, a no responders error will be returned
-                        // No responders error might also happen when reconnecting to cluster
-                        ThrowIfNoResponders = true,
                     },
                     cancellationToken)
                 .ConfigureAwait(false);
 
-            try
+            await foreach (var msg in sub.Msgs.ReadAllAsync(cancellationToken).ConfigureAwait(false))
             {
-                await foreach (var msg in sub.Msgs.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+                // If JetStream is disabled, a no responders error will be returned.
+                // No responders error might also happen when reconnecting to cluster.
+                // We should retry in those cases.
+                if (msg.HasNoResponders)
                 {
-                    if (msg.Data == null)
-                    {
-                        throw new NatsJSException("No response data received");
-                    }
-
-                    return msg.Data;
+                    break;
                 }
-            }
-            catch (NatsNoRespondersException)
-            {
+                else if (msg.Data == null)
+                {
+                    return new NatsJSException("No response data received");
+                }
+
+                return msg.Data;
             }
 
             if (i < retryMax)
@@ -169,7 +213,7 @@ public async ValueTask<PubAckResponse> PublishAsync<T>(
 
         // We throw a specific exception here for convenience so that the caller doesn't
         // have to check for the exception message etc.
-        throw new NatsJSPublishNoResponseException();
+        return new NatsJSPublishNoResponseException();
     }
 
     public async ValueTask<NatsJSPublishConcurrentFuture> PublishConcurrentAsync<T>(

src/NATS.Client.KeyValueStore/INatsKVStore.cs

@@ -26,6 +26,20 @@ public interface INatsKVStore
     /// <returns>Revision number</returns>
     ValueTask<ulong> PutAsync<T>(string key, T value, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Tries to put a value into the bucket using the key
+    /// </summary>
+    /// <param name="key">Key of the entry</param>
+    /// <param name="value">Value of the entry</param>
+    /// <param name="serializer">Serializer to use for the message type.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
+    /// <typeparam name="T">Serialized value type</typeparam>
+    /// <returns>Revision number</returns>
+    /// <remarks>
+    /// Use this method to avoid exceptions
+    /// </remarks>
+    ValueTask<NatsResult<ulong>> TryPutAsync<T>(string key, T value, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Create a new entry in the bucket only if it doesn't exist
     /// </summary>
@@ -37,6 +51,20 @@ public interface INatsKVStore
     /// <returns>The revision number of the entry</returns>
     ValueTask<ulong> CreateAsync<T>(string key, T value, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Tries to create a new entry in the bucket only if it doesn't exist
+    /// </summary>
+    /// <param name="key">Key of the entry</param>
+    /// <param name="value">Value of the entry</param>
+    /// <param name="serializer">Serializer to use for the message type.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
+    /// <typeparam name="T">Serialized value type</typeparam>
+    /// <returns>A NatsResult object representing the revision number of the created entry or an error.</returns>
+    /// <remarks>
+    /// Use this method to avoid exceptions
+    /// </remarks>
+    ValueTask<NatsResult<ulong>> TryCreateAsync<T>(string key, T value, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Update an entry in the bucket only if last update revision matches
     /// </summary>
@@ -49,6 +77,21 @@ public interface INatsKVStore
     /// <returns>The revision number of the updated entry</returns>
     ValueTask<ulong> UpdateAsync<T>(string key, T value, ulong revision, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Tries to update an entry in the bucket only if last update revision matches
+    /// </summary>
+    /// <param name="key">Key of the entry</param>
+    /// <param name="value">Value of the entry</param>
+    /// <param name="revision">Last revision number to match</param>
+    /// <param name="serializer">Serializer to use for the message type.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
+    /// <typeparam name="T">Serialized value type</typeparam>
+    /// <returns>A NatsResult object representing the revision number of the updated entry or an error.</returns>
+    /// <remarks>
+    /// Use this method to avoid exceptions
+    /// </remarks>
+    ValueTask<NatsResult<ulong>> TryUpdateAsync<T>(string key, T value, ulong revision, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Delete an entry from the bucket
     /// </summary>
@@ -57,6 +100,18 @@ public interface INatsKVStore
     /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
     ValueTask DeleteAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Delete an entry from the bucket
+    /// </summary>
+    /// <param name="key">Key of the entry</param>
+    /// <param name="opts">Delete options</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
+    /// <returns>A NatsResult object representing success or an error.</returns>
+    /// <remarks>
+    /// Use this method to avoid exceptions
+    /// </remarks>
+    ValueTask<NatsResult> TryDeleteAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Purge an entry from the bucket
     /// </summary>
@@ -65,6 +120,18 @@ public interface INatsKVStore
     /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
     ValueTask PurgeAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Tries to purge an entry from the bucket
+    /// </summary>
+    /// <param name="key">Key of the entry</param>
+    /// <param name="opts">Delete options</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
+    /// <returns>A NatsResult object representing success or an error.</returns>
+    /// <remarks>
+    /// Use this method to avoid exceptions
+    /// </remarks>
+    ValueTask<NatsResult> TryPurgeAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Get an entry from the bucket using the key
     /// </summary>