diff --git a/src/Dapr.Client/BulkStateItem.cs b/src/Dapr.Client/BulkStateItem.cs index fb717e1a8..5b30ddf21 100644 --- a/src/Dapr.Client/BulkStateItem.cs +++ b/src/Dapr.Client/BulkStateItem.cs @@ -49,4 +49,42 @@ public BulkStateItem(string key, string value, string etag) /// public string ETag { get; } } + + /// + /// Represents a state object returned from a bulk get state operation where the value has + /// been deserialized to the specified type. + /// + public readonly struct BulkStateItem + { + /// + /// Initializes a new instance of the class. + /// + /// The state key. + /// The typed value. + /// The ETag. + /// + /// Application code should not need to create instances of . + /// + public BulkStateItem(string key, TValue value, string etag) + { + this.Key = key; + this.Value = value; + this.ETag = etag; + } + + /// + /// Gets the state key. + /// + public string Key { get; } + + /// + /// Gets the deserialized value of the indicated type. + /// + public TValue Value { get; } + + /// + /// Get the ETag. + /// + public string ETag { get; } + } } diff --git a/src/Dapr.Client/DaprClient.cs b/src/Dapr.Client/DaprClient.cs index 7f094804f..21777105b 100644 --- a/src/Dapr.Client/DaprClient.cs +++ b/src/Dapr.Client/DaprClient.cs @@ -717,6 +717,20 @@ public abstract Task InvokeMethodGrpcAsync( /// A that will return the list of values when the operation has completed. public abstract Task> GetBulkStateAsync(string storeName, IReadOnlyList keys, int? parallelism, IReadOnlyDictionary metadata = default, CancellationToken cancellationToken = default); + /// + /// Gets a list of deserialized values associated with the from the Dapr state store. This overload should be used + /// if you expect the values of all the retrieved items to match the shape of the indicated . If you expect that + /// the values may differ in type from one another, do not specify the type parameter and instead use the original method + /// so the serialized string values will be returned instead. + /// + /// The name of state store to read from. + /// The list of keys to get values for. + /// The number of concurrent get operations the Dapr runtime will issue to the state store. a value equal to or smaller than 0 means max parallelism. + /// A collection of metadata key-value pairs that will be provided to the state store. The valid metadata keys and values are determined by the type of state store used. + /// A that can be used to cancel the operation. + /// A that will return the list of deserialized values when the operation has completed. + public abstract Task>> GetBulkStateAsync(string storeName, IReadOnlyList keys, int? parallelism, IReadOnlyDictionary metadata = default, CancellationToken cancellationToken = default); + /// /// Saves a list of to the Dapr state store. /// diff --git a/src/Dapr.Client/DaprClientGrpc.cs b/src/Dapr.Client/DaprClientGrpc.cs index 9c99b9eee..f856b87e6 100644 --- a/src/Dapr.Client/DaprClientGrpc.cs +++ b/src/Dapr.Client/DaprClientGrpc.cs @@ -601,7 +601,50 @@ public override async Task InvokeMethodGrpcAsync #region State Apis + /// public override async Task> GetBulkStateAsync(string storeName, IReadOnlyList keys, int? parallelism, IReadOnlyDictionary metadata = default, CancellationToken cancellationToken = default) + { + var rawBulkState = await GetBulkStateRawAsync(storeName, keys, parallelism, metadata, cancellationToken); + + var bulkResponse = new List(); + foreach (var item in rawBulkState) + { + bulkResponse.Add(new BulkStateItem(item.Key, item.Value.ToStringUtf8(), item.Etag)); + } + + return bulkResponse; + } + + /// + public override async Task>> GetBulkStateAsync( + string storeName, + IReadOnlyList keys, + int? parallelism, + IReadOnlyDictionary metadata = default, + CancellationToken cancellationToken = default) + { + var rawBulkState = await GetBulkStateRawAsync(storeName, keys, parallelism, metadata, cancellationToken); + + var bulkResponse = new List>(); + foreach (var item in rawBulkState) + { + var deserializedValue = TypeConverters.FromJsonByteString(item.Value, this.JsonSerializerOptions); + bulkResponse.Add(new BulkStateItem(item.Key, deserializedValue, item.Etag)); + } + + return bulkResponse; + } + + /// + /// Retrieves the bulk state data, but rather than deserializing the values, leaves the specific handling + /// to the public callers of this method to avoid duplicate deserialization. + /// + private async Task> GetBulkStateRawAsync( + string storeName, + IReadOnlyList keys, + int? parallelism, + IReadOnlyDictionary metadata = default, + CancellationToken cancellationToken = default) { ArgumentVerifier.ThrowIfNullOrEmpty(storeName, nameof(storeName)); if (keys.Count == 0) @@ -609,7 +652,7 @@ public override async Task> GetBulkStateAsync(strin var envelope = new Autogenerated.GetBulkStateRequest() { - StoreName = storeName, + StoreName = storeName, Parallelism = parallelism ?? default }; @@ -632,18 +675,20 @@ public override async Task> GetBulkStateAsync(strin } catch (RpcException ex) { - throw new DaprException("State operation failed: the Dapr endpoint indicated a failure. See InnerException for details.", ex); + throw new DaprException( + "State operation failed: the Dapr endpoint indicated a failure. See InnerException for details.", + ex); } - var bulkResponse = new List(); + var bulkResponse = new List<(string Key, string Etag, ByteString Value)>(); foreach (var item in response.Items) { - bulkResponse.Add(new BulkStateItem(item.Key, item.Data.ToStringUtf8(), item.Etag)); + bulkResponse.Add((item.Key, item.Etag, item.Data)); } return bulkResponse; } - + /// public override async Task GetStateAsync( string storeName, diff --git a/test/Dapr.Client.Test/StateApiTest.cs b/test/Dapr.Client.Test/StateApiTest.cs index cfa664663..2595fb006 100644 --- a/test/Dapr.Client.Test/StateApiTest.cs +++ b/test/Dapr.Client.Test/StateApiTest.cs @@ -75,6 +75,30 @@ public async Task GetBulkStateAsync_CanReadState() state.Should().HaveCount(1); } + [Fact] + public async Task GetBulkStateAsync_CanReadDeserializedState() + { + await using var client = TestClient.CreateForDaprClient(); + + var key = "test"; + var request = await client.CaptureGrpcRequestAsync(async daprClient => + { + return await daprClient.GetBulkStateAsync("testStore", new List() {key}, null); + }); + + // Create Response & Respond + const string size = "small"; + const string color = "yellow"; + var data = new Widget() {Size = size, Color = color}; + var envelope = MakeGetBulkStateResponse(key, data); + var state = await request.CompleteWithMessageAsync(envelope); + + // Get response and validate + state.Should().HaveCount(1); + state[0].Value.Size.Should().Match(size); + state[0].Value.Color.Should().Match(color); + } + [Fact] public async Task GetBulkStateAsync_WrapsRpcException() {