dapr
/
dotnet-sdk
mirror of https://github.com/dapr/dotnet-sdk.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into release-1.16 

# Conflicts:
#	Directory.Packages.props
#	examples/Client/Cryptography/Examples/EncryptDecryptFileStreamExample.cs
#	src/Dapr.Actors/Runtime/ActorManager.cs
#	src/Dapr.Client/DaprClient.cs
#	src/Dapr.Client/DaprClientGrpc.cs
#	test/Dapr.Actors.Test/Runtime/ActorManagerTests.cs
#	test/Dapr.Client.Test/CryptographyApiTest.cs
This commit is contained in:
Whit Waldo 2025-06-21 04:57:28 -05:00
parent 5e8e88de77 435d9fa6d0
commit 31c98c12fd
12 changed files with 700 additions and 168 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
daprdocs/content/en/dotnet-sdk-docs/dotnet-development/dotnet-development-dapr-aspire.md
View File

@ -66,12 +66,18 @@ dotnet new web MyApp
```
Next we'll configure the AppHost project to add the necessary package to support local Dapr development. Navigate
into the AppHost directory with the following and install the `Aspire.Hosting.Dapr` package from NuGet into the project.
into the AppHost directory with the following and install the `CommunityToolkit.Aspire.Hosting.Dapr` package from NuGet into the project.
We'll also add a reference to our `MyApp` project so we can reference it during the registration process.
{{% alert color="primary" %}}
This package was previously called `Aspire.Hosting.Dapr`, which has been [marked as deprecated](https://www.nuget.org/packages/Aspire.Hosting.Dapr).
{{% /alert %}}
```sh
cd aspiredemo.AppHost
dotnet add package Aspire.Hosting.Dapr
dotnet add package CommunityToolkit.Aspire.Hosting.Dapr
dotnet add reference ../MyApp/
```

72
examples/Client/Cryptography/Examples/EncryptDecryptFileStreamExample.cs Normal file
View File

@ -0,0 +1,72 @@
// ------------------------------------------------------------------------
// Copyright 2023 The Dapr Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
// http://www.apache.org/licenses/LICENSE-2.0
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------------
using System.Buffers;
using Dapr.Client;
#pragma warning disable CS0618 // Type or member is obsolete
namespace Cryptography.Examples
{
internal class EncryptDecryptFileStreamExample(string componentName, string keyName) : Example
{
public override string DisplayName => "Use Cryptography to encrypt and decrypt a file";
public override async Task RunAsync(CancellationToken cancellationToken)
{
using var client = new DaprClientBuilder().Build();
// The name of the file we're using as an example
const string fileName = "file.txt";
Console.WriteLine("Original file contents:");
foreach (var line in await File.ReadAllLinesAsync(fileName, cancellationToken))
{
Console.WriteLine(line);
}
//Encrypt from a file stream and buffer the resulting bytes to an in-memory buffer
await using var encryptFs = new FileStream(fileName, FileMode.Open);
var bufferedEncryptedBytes = new ArrayBufferWriter<byte>();
await foreach (var bytes in (client.EncryptAsync(componentName, encryptFs, keyName,
new EncryptionOptions(KeyWrapAlgorithm.Rsa), cancellationToken)))
{
bufferedEncryptedBytes.Write(bytes.Span);
}
Console.WriteLine("Encrypted bytes:");
Console.WriteLine(Convert.ToBase64String(bufferedEncryptedBytes.WrittenMemory.ToArray()));
//We'll write to a temporary file via a FileStream
var tempDecryptedFile = Path.GetTempFileName();
await using var decryptFs = new FileStream(tempDecryptedFile, FileMode.Create);
//We'll stream the decrypted bytes from a MemoryStream into the above temporary file
await using var encryptedMs = new MemoryStream(bufferedEncryptedBytes.WrittenMemory.ToArray());
await foreach (var result in (client.DecryptAsync(componentName, encryptedMs, keyName,
cancellationToken)))
{
decryptFs.Write(result.Span);
}
decryptFs.Close();
//Let's confirm the value as written to the file
var decryptedValue = await File.ReadAllTextAsync(tempDecryptedFile, cancellationToken);
Console.WriteLine("Decrypted value: ");
Console.WriteLine(decryptedValue);
//And some cleanup to delete our temp file
File.Delete(tempDecryptedFile);
}
}
}

5
src/Dapr.Actors/Extensions/DurationExtensions.cs
View File

@ -68,13 +68,12 @@ internal static class DurationExtensions
if (period.StartsWith(MonthlyPrefixPeriod))
{
var dateTime = DateTime.UtcNow;
return dateTime.AddMonths(1) - dateTime;
return TimeSpan.FromDays(30);
}
if (period.StartsWith(MidnightPrefixPeriod))
{
return new TimeSpan();
return TimeSpan.Zero;
}
if (period.StartsWith(WeeklyPrefixPeriod))

76
src/Dapr.Actors/Runtime/ActorManager.cs
View File

@ -220,11 +220,11 @@ internal sealed class ActorManager
}
}
internal async Task FireTimerAsync(ActorId actorId, Stream requestBodyStream, CancellationToken cancellationToken = default)
{
#pragma warning disable 0618
var timerData = await JsonSerializer.DeserializeAsync<TimerInfo>(requestBodyStream);
#pragma warning restore 0618
internal async Task FireTimerAsync(ActorId actorId, Stream requestBodyStream, CancellationToken cancellationToken = default)
{
#pragma warning disable 0618
var timerData = await DeserializeAsync(requestBodyStream);
#pragma warning restore 0618
// Create a Func to be invoked by common method.
async Task<byte[]> RequestFunc(Actor actor, CancellationToken ct)
@ -243,10 +243,66 @@ internal sealed class ActorManager
await this.DispatchInternalAsync(actorId, this.timerMethodContext, RequestFunc, cancellationToken);
}
internal async Task ActivateActorAsync(ActorId actorId)
{
// An actor is activated by "Dapr" runtime when a call is to be made for an actor.
var state = await this.CreateActorAsync(actorId);
#pragma warning disable 0618
internal static async Task<TimerInfo> DeserializeAsync(Stream stream)
{
var json = await JsonSerializer.DeserializeAsync<JsonElement>(stream);
if (json.ValueKind == JsonValueKind.Null)
{
return null;
}
var setAnyProperties = false; // Used to determine if anything was actually deserialized
var dueTime = TimeSpan.Zero;
var callback = "";
var period = TimeSpan.Zero;
var data = Array.Empty<byte>();
TimeSpan? ttl = null;
if (json.TryGetProperty("callback", out var callbackProperty))
{
setAnyProperties = true;
callback = callbackProperty.GetString();
}
if (json.TryGetProperty("dueTime", out var dueTimeProperty))
{
setAnyProperties = true;
var dueTimeString = dueTimeProperty.GetString();
dueTime = ConverterUtils.ConvertTimeSpanFromDaprFormat(dueTimeString);
}
if (json.TryGetProperty("period", out var periodProperty))
{
setAnyProperties = true;
var periodString = periodProperty.GetString();
(period, _) = ConverterUtils.ConvertTimeSpanValueFromISO8601Format(periodString);
}
if (json.TryGetProperty("data", out var dataProperty) && dataProperty.ValueKind != JsonValueKind.Null)
{
setAnyProperties = true;
data = dataProperty.GetBytesFromBase64();
}
if (json.TryGetProperty("ttl", out var ttlProperty))
{
setAnyProperties = true;
var ttlString = ttlProperty.GetString();
ttl = ConverterUtils.ConvertTimeSpanFromDaprFormat(ttlString);
}
if (!setAnyProperties)
{
return null; //No properties were ever deserialized, so return null instead of default values
}
return new TimerInfo(callback, data, dueTime, period, ttl);
}
#pragma warning restore 0618
internal async Task ActivateActorAsync(ActorId actorId)
{
// An actor is activated by "Dapr" runtime when a call is to be made for an actor.
var state = await this.CreateActorAsync(actorId);
try
{
@ -397,4 +453,4 @@ internal sealed class ActorManager
return new Tuple<string, byte[]>(string.Empty, responseMsgBodyBytes);
}
}
}

4
src/Dapr.Actors/Runtime/ConverterUtils.cs
View File

@ -49,7 +49,7 @@ internal static class ConverterUtils
int msIndex = spanOfValue.IndexOf("ms");
// handle days from hours.
var hoursSpan = spanOfValue.Slice(0, hIndex);
var hoursSpan = spanOfValue[..hIndex];
var hours = int.Parse(hoursSpan);
var days = hours / 24;
hours %= 24;
@ -103,7 +103,7 @@ internal static class ConverterUtils
builder.Append($"{value.Days}D");
}
builder.Append("T");
builder.Append('T');
if(value.Hours > 0)
{

153
src/Dapr.Client/Crypto/DecryptionStreamProcessor.cs Normal file
View File

@ -0,0 +1,153 @@
// ------------------------------------------------------------------------
// Copyright 2025 The Dapr Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
// http://www.apache.org/licenses/LICENSE-2.0
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------------
#nullable enable
using System;
using System.Collections.Generic;
using System.IO;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Google.Protobuf;
using Grpc.Core;
using Autogenerated = Dapr.Client.Autogen.Grpc.v1;
#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
namespace Dapr.Client.Crypto;
/// <summary>
/// Provides the implementation to decrypt a stream of plaintext data with the Dapr runtime.
/// </summary>
internal sealed class DecryptionStreamProcessor : IDisposable
{
private bool disposed;
private readonly Channel<ReadOnlyMemory<byte>> outputChannel = Channel.CreateUnbounded<ReadOnlyMemory<byte>>();
/// <summary>
/// Surfaces any exceptions encountered while asynchronously processing the inbound and outbound streams.
/// </summary>
internal event EventHandler<Exception>? OnException;
/// <summary>
/// Sends the provided bytes in chunks to the sidecar for the encryption operation.
/// </summary>
/// <param name="inputStream">The stream containing the bytes to decrypt.</param>
/// <param name="call">The call to make to the sidecar to process the encryption operation.</param>
/// <param name="streamingBlockSizeInBytes">The size, in bytes, of the streaming blocks.</param>
/// <param name="options">The decryption options.</param>
/// <param name="cancellationToken">Token used to cancel the ongoing request.</param>
public async Task ProcessStreamAsync(
Stream inputStream,
AsyncDuplexStreamingCall<Autogenerated.DecryptRequest, Autogenerated.DecryptResponse> call,
int streamingBlockSizeInBytes,
Autogenerated.DecryptRequestOptions options,
CancellationToken cancellationToken)
{
//Read from the input stream and write to the gRPC call
_ = Task.Run(async () =>
{
try
{
await using var bufferedStream = new BufferedStream(inputStream, streamingBlockSizeInBytes);
var buffer = new byte[streamingBlockSizeInBytes];
int bytesRead;
ulong sequenceNumber = 0;
while ((bytesRead = await bufferedStream.ReadAsync(buffer, cancellationToken)) > 0)
{
var request = new Autogenerated.DecryptRequest
{
Payload = new Autogenerated.StreamPayload
{
Data = ByteString.CopyFrom(buffer, 0, bytesRead), Seq = sequenceNumber
}
};
//Only include the options in the first message
if (sequenceNumber == 0)
{
request.Options = options;
}
await call.RequestStream.WriteAsync(request, cancellationToken);
//Increment the sequence number
sequenceNumber++;
}
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
// Expected cancellation exception
}
catch (Exception ex)
{
OnException?.Invoke(this, ex);
}
finally
{
await call.RequestStream.CompleteAsync();
}
}, cancellationToken);
//Start reading from the gRPC call and writing to the output channel
_ = Task.Run(async () =>
{
try
{
await foreach (var response in call.ResponseStream.ReadAllAsync(cancellationToken))
{
await outputChannel.Writer.WriteAsync(response.Payload.Data.Memory, cancellationToken);
}
}
catch (Exception ex)
{
OnException?.Invoke(this, ex);
}
finally
{
outputChannel.Writer.Complete();
}
}, cancellationToken);
}
/// <summary>
/// Retrieves the processed bytes from the operation from the sidecar and
/// returns as an enumerable stream.
/// </summary>
public async IAsyncEnumerable<ReadOnlyMemory<byte>> GetProcessedDataAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (var data in outputChannel.Reader.ReadAllAsync(cancellationToken))
{
yield return data;
}
}
public void Dispose()
{
Dispose(true);
}
private void Dispose(bool disposing)
{
if (!disposed)
{
if (disposing)
{
outputChannel.Writer.TryComplete();
}
disposed = true;
}
}
}

154
src/Dapr.Client/Crypto/EncryptionStreamProcessor.cs Normal file
View File

@ -0,0 +1,154 @@
// ------------------------------------------------------------------------
// Copyright 2025 The Dapr Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
// http://www.apache.org/licenses/LICENSE-2.0
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------------
#nullable enable
using System;
using System.Collections.Generic;
using System.IO;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Google.Protobuf;
using Grpc.Core;
using Autogenerated = Dapr.Client.Autogen.Grpc.v1;
#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
namespace Dapr.Client.Crypto;
/// <summary>
/// Provides the implementation to encrypt a stream of plaintext data with the Dapr runtime.
/// </summary>
internal sealed class EncryptionStreamProcessor : IDisposable
{
private bool disposed;
private readonly Channel<ReadOnlyMemory<byte>> outputChannel = Channel.CreateUnbounded<ReadOnlyMemory<byte>>();
/// <summary>
/// Surfaces any exceptions encountered while asynchronously processing the inbound and outbound streams.
/// </summary>
internal event EventHandler<Exception>? OnException;
/// <summary>
/// Sends the provided bytes in chunks to the sidecar for the encryption operation.
/// </summary>
/// <param name="inputStream">The stream containing the bytes to encrypt.</param>
/// <param name="call">The call to make to the sidecar to process the encryption operation.</param>
/// <param name="options">The encryption options.</param>
/// <param name="streamingBlockSizeInBytes">The size, in bytes, of the streaming blocks.</param>
/// <param name="cancellationToken">Token used to cancel the ongoing request.</param>
public async Task ProcessStreamAsync(
Stream inputStream,
AsyncDuplexStreamingCall<Autogenerated.EncryptRequest, Autogenerated.EncryptResponse> call,
Autogenerated.EncryptRequestOptions options,
int streamingBlockSizeInBytes,
CancellationToken cancellationToken)
{
//Read from the input stream and write to the gRPC call
_ = Task.Run(async () =>
{
try
{
await using var bufferedStream = new BufferedStream(inputStream, streamingBlockSizeInBytes);
var buffer = new byte[streamingBlockSizeInBytes];
int bytesRead;
ulong sequenceNumber = 0;
while ((bytesRead = await bufferedStream.ReadAsync(buffer, cancellationToken)) > 0)
{
var request = new Autogenerated.EncryptRequest
{
Payload = new Autogenerated.StreamPayload
{
Data = ByteString.CopyFrom(buffer, 0, bytesRead), Seq = sequenceNumber
}
};
//Only include the options in the first message
if (sequenceNumber == 0)
{
request.Options = options;
}
await call.RequestStream.WriteAsync(request, cancellationToken);
//Increment the sequence number
sequenceNumber++;
}
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
// Expected cancellation exception
}
catch (Exception ex)
{
OnException?.Invoke(this, ex);
}
finally
{
await call.RequestStream.CompleteAsync();
}
}, cancellationToken);
//Start reading from the gRPC call and writing to the output channel
_ = Task.Run(async () =>
{
try
{
await foreach (var response in call.ResponseStream.ReadAllAsync(cancellationToken))
{
await outputChannel.Writer.WriteAsync(response.Payload.Data.Memory, cancellationToken);
}
}
catch (Exception ex)
{
OnException?.Invoke(this, ex);
}
finally
{
outputChannel.Writer.Complete();
}
}, cancellationToken);
}
/// <summary>
/// Retrieves the processed bytes from the operation from the sidecar and
/// returns as an enumerable stream.
/// </summary>
public async IAsyncEnumerable<ReadOnlyMemory<byte>> GetProcessedDataAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (var data in outputChannel.Reader.ReadAllAsync(cancellationToken))
{
yield return data;
}
}
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
private void Dispose(bool disposing)
{
if (!disposed)
{
if (disposing)
{
outputChannel.Writer.TryComplete();
}
disposed = true;
}
}
}

182
src/Dapr.Client/DaprClientGrpc.cs
View File

@ -11,10 +11,9 @@
// limitations under the License.
// ------------------------------------------------------------------------
using System.Diagnostics.CodeAnalysis;
namespace Dapr.Client;
using Crypto;
using System;
using System.Buffers;
using System.Collections.Generic;
@ -30,6 +29,7 @@ using Google.Protobuf;
using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using Grpc.Net.Client;
using System.Diagnostics.CodeAnalysis;
using Autogenerated = Dapr.Client.Autogen.Grpc.v1;
/// <summary>
@ -1675,11 +1675,10 @@ internal class DaprClientGrpc : DaprClient
{
using var memoryStream = plaintextBytes.CreateMemoryStream(true);
var encryptionResult =
await EncryptAsync(vaultResourceName, memoryStream, keyName, encryptionOptions, cancellationToken);
var encryptionResult = EncryptAsync(vaultResourceName, memoryStream, keyName, encryptionOptions, cancellationToken);
var bufferedResult = new ArrayBufferWriter<byte>();
await foreach (var item in encryptionResult.WithCancellation(cancellationToken))
await foreach (var item in encryptionResult)
{
bufferedResult.Write(item.Span);
}
@ -1691,12 +1690,15 @@ internal class DaprClientGrpc : DaprClient
[Experimental("DAPR_CRYPTOGRAPHY", UrlFormat = "https://docs.dapr.io/developing-applications/building-blocks/cryptography/cryptography-overview/")]
public override async Task<IAsyncEnumerable<ReadOnlyMemory<byte>>> EncryptAsync(string vaultResourceName,
Stream plaintextStream,
string keyName, EncryptionOptions encryptionOptions, CancellationToken cancellationToken = default)
string keyName, EncryptionOptions encryptionOptions,
[EnumeratorCancellation] CancellationToken cancellationToken = default)
{
ArgumentVerifier.ThrowIfNullOrEmpty(vaultResourceName, nameof(vaultResourceName));
ArgumentVerifier.ThrowIfNullOrEmpty(keyName, nameof(keyName));
ArgumentVerifier.ThrowIfNull(plaintextStream, nameof(plaintextStream));
ArgumentVerifier.ThrowIfNull(encryptionOptions, nameof(encryptionOptions));
EventHandler<Exception> exceptionHandler = (_, ex) => throw ex;
var shouldOmitDecryptionKeyName =
string.IsNullOrWhiteSpace(encryptionOptions
@ -1719,72 +1721,24 @@ internal class DaprClientGrpc : DaprClient
}
var options = CreateCallOptions(headers: null, cancellationToken);
var duplexStream = client.EncryptAlpha1(options);
var duplexStream = Client.EncryptAlpha1(options);
//Run both operations at the same time, but return the output of the streaming values coming from the operation
var receiveResult = Task.FromResult(RetrieveEncryptedStreamAsync(duplexStream, cancellationToken));
return await Task.WhenAll(
//Stream the plaintext data to the sidecar in chunks
SendPlaintextStreamAsync(plaintextStream, encryptionOptions.StreamingBlockSizeInBytes,
duplexStream, encryptRequestOptions, cancellationToken),
//At the same time, retrieve the encrypted response from the sidecar
receiveResult).ContinueWith(_ => receiveResult.Result, cancellationToken);
}
/// <summary>
/// Sends the plaintext bytes in chunks to the sidecar to be encrypted.
/// </summary>
private static async Task SendPlaintextStreamAsync(Stream plaintextStream,
int streamingBlockSizeInBytes,
AsyncDuplexStreamingCall<Autogenerated.EncryptRequest, Autogenerated.EncryptResponse> duplexStream,
Autogenerated.EncryptRequestOptions encryptRequestOptions,
CancellationToken cancellationToken)
{
//Start with passing the metadata about the encryption request itself in the first message
await duplexStream.RequestStream.WriteAsync(
new Autogenerated.EncryptRequest { Options = encryptRequestOptions }, cancellationToken);
//Send the plaintext bytes in blocks in subsequent messages
await using (var bufferedStream = new BufferedStream(plaintextStream, streamingBlockSizeInBytes))
using var streamProcessor = new EncryptionStreamProcessor();
try
{
var buffer = new byte[streamingBlockSizeInBytes];
int bytesRead;
ulong sequenceNumber = 0;
streamProcessor.OnException += exceptionHandler;
await streamProcessor.ProcessStreamAsync(plaintextStream, duplexStream, encryptRequestOptions,
encryptionOptions.StreamingBlockSizeInBytes,
cancellationToken);
while ((bytesRead =
await bufferedStream.ReadAsync(buffer.AsMemory(0, streamingBlockSizeInBytes),
cancellationToken)) !=
0)
await foreach (var value in streamProcessor.GetProcessedDataAsync(cancellationToken))
{
await duplexStream.RequestStream.WriteAsync(
new Autogenerated.EncryptRequest
{
Payload = new Autogenerated.StreamPayload
{
Data = ByteString.CopyFrom(buffer, 0, bytesRead), Seq = sequenceNumber
}
}, cancellationToken);
//Increment the sequence number
sequenceNumber++;
yield return value;
}
}
//Send the completion message
await duplexStream.RequestStream.CompleteAsync();
}
/// <summary>
/// Retrieves the encrypted bytes from the encryption operation on the sidecar and returns as an enumerable stream.
/// </summary>
private static async IAsyncEnumerable<ReadOnlyMemory<byte>> RetrieveEncryptedStreamAsync(
AsyncDuplexStreamingCall<Autogenerated.EncryptRequest, Autogenerated.EncryptResponse> duplexStream,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (var encryptResponse in duplexStream.ResponseStream.ReadAllAsync(cancellationToken)
.ConfigureAwait(false))
finally
{
yield return encryptResponse.Payload.Data.Memory;
streamProcessor.OnException -= exceptionHandler;
}
}
@ -1792,31 +1746,42 @@ internal class DaprClientGrpc : DaprClient
[Experimental("DAPR_CRYPTOGRAPHY", UrlFormat = "https://docs.dapr.io/developing-applications/building-blocks/cryptography/cryptography-overview/")]
public override async Task<IAsyncEnumerable<ReadOnlyMemory<byte>>> DecryptAsync(string vaultResourceName,
Stream ciphertextStream, string keyName,
DecryptionOptions decryptionOptions, CancellationToken cancellationToken = default)
DecryptionOptions decryptionOptions,
[EnumeratorCancellation] CancellationToken cancellationToken = default)
{
ArgumentVerifier.ThrowIfNullOrEmpty(vaultResourceName, nameof(vaultResourceName));
ArgumentVerifier.ThrowIfNullOrEmpty(keyName, nameof(keyName));
ArgumentVerifier.ThrowIfNull(ciphertextStream, nameof(ciphertextStream));
ArgumentVerifier.ThrowIfNull(decryptionOptions, nameof(decryptionOptions));
decryptionOptions ??= new DecryptionOptions();
EventHandler<Exception> exceptionHandler = (_, ex) => throw ex;
var decryptRequestOptions = new Autogenerated.DecryptRequestOptions
{
ComponentName = vaultResourceName, KeyName = keyName
ComponentName = vaultResourceName,
KeyName = keyName
};
var options = CreateCallOptions(headers: null, cancellationToken);
var duplexStream = client.DecryptAlpha1(options);
//Run both operations at the same time, but return the output of the streaming values coming from the operation
var receiveResult = Task.FromResult(RetrieveDecryptedStreamAsync(duplexStream, cancellationToken));
return await Task.WhenAll(
//Stream the ciphertext data to the sidecar in chunks
SendCiphertextStreamAsync(ciphertextStream, decryptionOptions.StreamingBlockSizeInBytes,
duplexStream, decryptRequestOptions, cancellationToken),
//At the same time, retrieve the decrypted response from the sidecar
receiveResult)
//Return only the result of the `RetrieveEncryptedStreamAsync` method
.ContinueWith(_ => receiveResult.Result, cancellationToken);
using var streamProcessor = new DecryptionStreamProcessor();
try
{
streamProcessor.OnException += exceptionHandler;
await streamProcessor.ProcessStreamAsync(ciphertextStream, duplexStream, decryptionOptions.StreamingBlockSizeInBytes,
decryptRequestOptions,
cancellationToken);
await foreach (var value in streamProcessor.GetProcessedDataAsync(cancellationToken))
{
yield return value;
}
}
finally
{
streamProcessor.OnException -= exceptionHandler;
}
}
/// <inheritdoc />
@ -1826,62 +1791,6 @@ internal class DaprClientGrpc : DaprClient
DecryptAsync(vaultResourceName, ciphertextStream, keyName, new DecryptionOptions(),
cancellationToken);
/// <summary>
/// Sends the ciphertext bytes in chunks to the sidecar to be decrypted.
/// </summary>
private static async Task SendCiphertextStreamAsync(Stream ciphertextStream,
int streamingBlockSizeInBytes,
AsyncDuplexStreamingCall<Autogenerated.DecryptRequest, Autogenerated.DecryptResponse> duplexStream,
Autogenerated.DecryptRequestOptions decryptRequestOptions,
CancellationToken cancellationToken)
{
//Start with passing the metadata about the decryption request itself in the first message
await duplexStream.RequestStream.WriteAsync(
new Autogenerated.DecryptRequest { Options = decryptRequestOptions }, cancellationToken);
//Send the ciphertext bytes in blocks in subsequent messages
await using (var bufferedStream = new BufferedStream(ciphertextStream, streamingBlockSizeInBytes))
{
var buffer = new byte[streamingBlockSizeInBytes];
int bytesRead;
ulong sequenceNumber = 0;
while ((bytesRead =
await bufferedStream.ReadAsync(buffer.AsMemory(0, streamingBlockSizeInBytes),
cancellationToken)) != 0)
{
await duplexStream.RequestStream.WriteAsync(
new Autogenerated.DecryptRequest
{
Payload = new Autogenerated.StreamPayload
{
Data = ByteString.CopyFrom(buffer, 0, bytesRead), Seq = sequenceNumber
}
}, cancellationToken);
//Increment the sequence number
sequenceNumber++;
}
}
//Send the completion message
await duplexStream.RequestStream.CompleteAsync();
}
/// <summary>
/// Retrieves the decrypted bytes from the decryption operation on the sidecar and returns as an enumerable stream.
/// </summary>
private static async IAsyncEnumerable<ReadOnlyMemory<byte>> RetrieveDecryptedStreamAsync(
AsyncDuplexStreamingCall<Autogenerated.DecryptRequest, Autogenerated.DecryptResponse> duplexStream,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (var decryptResponse in duplexStream.ResponseStream.ReadAllAsync(cancellationToken)
.ConfigureAwait(false))
{
yield return decryptResponse.Payload.Data.Memory;
}
}
/// <inheritdoc />
[Experimental("DAPR_CRYPTOGRAPHY", UrlFormat = "https://docs.dapr.io/developing-applications/building-blocks/cryptography/cryptography-overview/")]
public override async Task<ReadOnlyMemory<byte>> DecryptAsync(string vaultResourceName,
@ -1890,11 +1799,10 @@ internal class DaprClientGrpc : DaprClient
{
using var memoryStream = ciphertextBytes.CreateMemoryStream(true);
var decryptionResult =
await DecryptAsync(vaultResourceName, memoryStream, keyName, decryptionOptions, cancellationToken);
var decryptionResult = DecryptAsync(vaultResourceName, memoryStream, keyName, decryptionOptions, cancellationToken);
var bufferedResult = new ArrayBufferWriter<byte>();
await foreach (var item in decryptionResult.WithCancellation(cancellationToken))
await foreach (var item in decryptionResult)
{
bufferedResult.Write(item.Span);
}

35
src/Dapr.Jobs/DaprJobsGrpcClient.cs
View File

@ -170,17 +170,7 @@ internal sealed class DaprJobsGrpcClient(Autogenerated.Dapr.DaprClient client, H
var envelope = new Autogenerated.GetJobRequest { Name = jobName };
var grpcCallOptions = DaprClientUtilities.ConfigureGrpcCallOptions(typeof(DaprJobsClient).Assembly, this.DaprApiToken, cancellationToken);
var response = await Client.GetJobAlpha1Async(envelope, grpcCallOptions);
var schedule = DateTime.TryParse(response.Job.DueTime, out var dueTime)
? DaprJobSchedule.FromDateTime(dueTime)
: new DaprJobSchedule(response.Job.Schedule);
return new DaprJobDetails(schedule)
{
DueTime = !string.IsNullOrWhiteSpace(response.Job.DueTime) ? DateTime.Parse(response.Job.DueTime) : null,
Ttl = !string.IsNullOrWhiteSpace(response.Job.Ttl) ? DateTime.Parse(response.Job.Ttl) : null,
RepeatCount = (int?)response.Job.Repeats,
Payload = response.Job.Data.ToByteArray()
};
return DeserializeJobResponse(response);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
@ -199,6 +189,29 @@ internal sealed class DaprJobsGrpcClient(Autogenerated.Dapr.DaprClient client, H
throw new DaprException("Get job operation failed: the Dapr endpoint did not return the expected value.");
}
/// <summary>
/// Testable method for performing job response deserialization.
/// </summary>
/// <remarks>
/// This is exposed strictly for testing purposes.
/// </remarks>
/// <param name="response">The job response to deserialize.</param>
/// <returns>The deserialized job response.</returns>
internal static DaprJobDetails DeserializeJobResponse(Autogenerated.GetJobResponse response)
{
var schedule = DateTime.TryParse(response.Job.DueTime, out var dueTime)
? DaprJobSchedule.FromDateTime(dueTime)
: new DaprJobSchedule(response.Job.Schedule);
return new DaprJobDetails(schedule)
{
DueTime = !string.IsNullOrWhiteSpace(response.Job.DueTime) ? DateTime.Parse(response.Job.DueTime) : null,
Ttl = !string.IsNullOrWhiteSpace(response.Job.Ttl) ? DateTime.Parse(response.Job.Ttl) : null,
RepeatCount = (int?)response.Job.Repeats ?? 0,
Payload = response.Job.Data?.ToByteArray() ?? null
};
}
/// <summary>
/// Deletes the specified job.
/// </summary>

158
test/Dapr.Actors.Test/Runtime/ActorManagerTests.cs
View File

@ -12,6 +12,8 @@
// ------------------------------------------------------------------------
using System;
using System.IO;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Dapr.Actors.Client;
@ -175,6 +177,160 @@ public sealed class ActorManagerTests
Assert.Equal(1, activator.DeleteCallCount);
}
[Fact]
public async Task DeserializeTimer_Period_Iso8601_Time()
{
const string timerJson = "{\"callback\": \"TimerCallback\", \"period\": \"0h0m7s10ms\"}";
await using var stream = new MemoryStream(Encoding.UTF8.GetBytes(timerJson));
var result = await ActorManager.DeserializeAsync(stream);
Assert.Equal("TimerCallback", result.Callback);
Assert.Equal(Array.Empty<byte>(), result.Data);
Assert.Null(result.Ttl);
Assert.Equal(TimeSpan.Zero, result.DueTime);
Assert.Equal(TimeSpan.FromSeconds(7).Add(TimeSpan.FromMilliseconds(10)), result.Period);
}
[Fact]
public async Task DeserializeTimer_Period_DaprFormat_Every()
{
const string timerJson = "{\"callback\": \"TimerCallback\", \"period\": \"@every 15s\"}";
await using var stream = new MemoryStream(Encoding.UTF8.GetBytes(timerJson));
var result = await ActorManager.DeserializeAsync(stream);
Assert.Equal("TimerCallback", result.Callback);
Assert.Equal(Array.Empty<byte>(), result.Data);
Assert.Null(result.Ttl);
Assert.Equal(TimeSpan.Zero, result.DueTime);
Assert.Equal(TimeSpan.FromSeconds(15), result.Period);
}
[Fact]
public async Task DeserializeTimer_Period_DaprFormat_Every2()
{
const string timerJson = "{\"callback\": \"TimerCallback\", \"period\": \"@every 3h2m15s\"}";
await using var stream = new MemoryStream(Encoding.UTF8.GetBytes(timerJson));
var result = await ActorManager.DeserializeAsync(stream);
Assert.Equal("TimerCallback", result.Callback);
Assert.Equal(Array.Empty<byte>(), result.Data);
Assert.Null(result.Ttl);
Assert.Equal(TimeSpan.Zero, result.DueTime);
Assert.Equal(TimeSpan.FromHours(3).Add(TimeSpan.FromMinutes(2)).Add(TimeSpan.FromSeconds(15)), result.Period);
}
[Fact]
public async Task DeserializeTimer_Period_DaprFormat_Monthly()
{
const string timerJson = "{\"callback\": \"TimerCallback\", \"period\": \"@monthly\"}";
await using var stream = new MemoryStream(Encoding.UTF8.GetBytes(timerJson));
var result = await ActorManager.DeserializeAsync(stream);
Assert.Equal("TimerCallback", result.Callback);
Assert.Equal(Array.Empty<byte>(), result.Data);
Assert.Null(result.Ttl);
Assert.Equal(TimeSpan.Zero, result.DueTime);
Assert.Equal(TimeSpan.FromDays(30), result.Period);
}
[Fact]
public async Task DeserializeTimer_Period_DaprFormat_Weekly()
{
const string timerJson = "{\"callback\": \"TimerCallback\", \"period\": \"@weekly\"}";
await using var stream = new MemoryStream(Encoding.UTF8.GetBytes(timerJson));
var result = await ActorManager.DeserializeAsync(stream);
Assert.Equal("TimerCallback", result.Callback);
Assert.Equal(Array.Empty<byte>(), result.Data);
Assert.Null(result.Ttl);
Assert.Equal(TimeSpan.Zero, result.DueTime);
Assert.Equal(TimeSpan.FromDays(7), result.Period);
}
[Fact]
public async Task DeserializeTimer_Period_DaprFormat_Daily()
{
const string timerJson = "{\"callback\": \"TimerCallback\", \"period\": \"@daily\"}";
await using var stream = new MemoryStream(Encoding.UTF8.GetBytes(timerJson));
var result = await ActorManager.DeserializeAsync(stream);
Assert.Equal("TimerCallback", result.Callback);
Assert.Equal(Array.Empty<byte>(), result.Data);
Assert.Null(result.Ttl);
Assert.Equal(TimeSpan.Zero, result.DueTime);
Assert.Equal(TimeSpan.FromDays(1), result.Period);
}
[Fact]
public async Task DeserializeTimer_Period_DaprFormat_Hourly()
{
const string timerJson = "{\"callback\": \"TimerCallback\", \"period\": \"@hourly\"}";
await using var stream = new MemoryStream(Encoding.UTF8.GetBytes(timerJson));
var result = await ActorManager.DeserializeAsync(stream);
Assert.Equal("TimerCallback", result.Callback);
Assert.Equal(Array.Empty<byte>(), result.Data);
Assert.Null(result.Ttl);
Assert.Equal(TimeSpan.Zero, result.DueTime);
Assert.Equal(TimeSpan.FromHours(1), result.Period);
}
[Fact]
public async Task DeserializeTimer_DueTime_DaprFormat_Hourly()
{
const string timerJson = "{\"callback\": \"TimerCallback\", \"dueTime\": \"@hourly\"}";
await using var stream = new MemoryStream(Encoding.UTF8.GetBytes(timerJson));
var result = await ActorManager.DeserializeAsync(stream);
Assert.Equal("TimerCallback", result.Callback);
Assert.Equal(Array.Empty<byte>(), result.Data);
Assert.Null(result.Ttl);
Assert.Equal(TimeSpan.FromHours(1), result.DueTime);
Assert.Equal(TimeSpan.Zero, result.Period);
}
[Fact]
public async Task DeserializeTimer_DueTime_Iso8601Times()
{
const string timerJson = "{\"callback\": \"TimerCallback\", \"dueTime\": \"0h0m7s10ms\"}";
await using var stream = new MemoryStream(Encoding.UTF8.GetBytes(timerJson));
var result = await ActorManager.DeserializeAsync(stream);
Assert.Equal("TimerCallback", result.Callback);
Assert.Equal(Array.Empty<byte>(), result.Data);
Assert.Null(result.Ttl);
Assert.Equal(TimeSpan.Zero, result.Period);
Assert.Equal(TimeSpan.FromSeconds(7).Add(TimeSpan.FromMilliseconds(10)), result.DueTime);
}
[Fact]
public async Task DeserializeTimer_Ttl_DaprFormat_Hourly()
{
const string timerJson = "{\"callback\": \"TimerCallback\", \"ttl\": \"@hourly\"}";
await using var stream = new MemoryStream(Encoding.UTF8.GetBytes(timerJson));
var result = await ActorManager.DeserializeAsync(stream);
Assert.Equal("TimerCallback", result.Callback);
Assert.Equal(Array.Empty<byte>(), result.Data);
Assert.Equal(TimeSpan.Zero, result.DueTime);
Assert.Equal(TimeSpan.Zero, result.Period);
Assert.Equal(TimeSpan.FromHours(1), result.Ttl);
}
[Fact]
public async Task DeserializeTimer_Ttl_Iso8601Times()
{
const string timerJson = "{\"callback\": \"TimerCallback\", \"ttl\": \"0h0m7s10ms\"}";
await using var stream = new MemoryStream(Encoding.UTF8.GetBytes(timerJson));
var result = await ActorManager.DeserializeAsync(stream);
Assert.Equal("TimerCallback", result.Callback);
Assert.Equal(Array.Empty<byte>(), result.Data);
Assert.Equal(TimeSpan.Zero, result.DueTime);
Assert.Equal(TimeSpan.Zero, result.Period);
Assert.Equal(TimeSpan.FromSeconds(7).Add(TimeSpan.FromMilliseconds(10)), result.Ttl);
}
private interface ITestActor : IActor { }
private class TestActor : Actor, ITestActor, IDisposable
@ -255,4 +411,4 @@ public sealed class ActorManagerTests
return base.DeleteAsync(state);
}
}
}
}

3
test/Dapr.Client.Test/CryptographyApiTest.cs
View File

@ -1,5 +1,4 @@
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Xunit;
@ -91,4 +90,4 @@ public class CryptographyApiTest
await Assert.ThrowsAsync<ArgumentException>(async () => await client.DecryptAsync("myVault",
new MemoryStream(), keyName, new DecryptionOptions(), CancellationToken.None));
}
}
}

16
test/Dapr.Jobs.Test/DaprJobsGrpcClientTests.cs
View File

@ -13,6 +13,7 @@
using System;
using System.Net.Http;
using Dapr.Client.Autogen.Grpc.v1;
using Dapr.Jobs.Models;
using Moq;
@ -166,6 +167,21 @@ public sealed class DaprJobsGrpcClientTests
});
#pragma warning restore CS0618 // Type or member is obsolete
}
[Fact]
public void ShouldDeserialize_EveryExpression()
{
const string scheduleText = "@every 1m";
var response = new GetJobResponse { Job = new Job { Name = "test", Schedule = scheduleText } };
var schedule = DaprJobSchedule.FromExpression(scheduleText);
var jobDetails = DaprJobsGrpcClient.DeserializeJobResponse(response);
Assert.Null(jobDetails.Payload);
Assert.Equal(0, jobDetails.RepeatCount);
Assert.Null(jobDetails.Ttl);
Assert.Null(jobDetails.DueTime);
Assert.Equal(jobDetails.Schedule.ExpressionValue, schedule.ExpressionValue);
}
private sealed record TestPayload(string Name, string Color);
}