mirror of https://github.com/abpframework/abp.git
Browse Source
- IBlobPipelineContributor pipeline; encryption runs innermost with an authenticated end check - Adapt FileSystem/AWS to forward-only streams; fail early on unsupported AES-GCMpull/25836/head
maliming
2 weeks ago
61 changed files with 5106 additions and 251 deletions
@ -0,0 +1,122 @@ |
|||||
|
```json |
||||
|
//[doc-seo] |
||||
|
{ |
||||
|
"Description": "Learn how to transform BLOB content transparently (compression, validation, watermarking...) with pipeline contributors in ABP Framework." |
||||
|
} |
||||
|
``` |
||||
|
|
||||
|
# BLOB Content Pipeline |
||||
|
|
||||
|
The BLOB Storing system can pass the BLOB content through a **pipeline of contributors** while it is saved and read. A contributor transforms the content stream transparently, on top of the configured [storage provider](../blob-storing): compression, watermarking, content validation or any other stream transformation can be implemented without changing the storage provider or the application code that works with `IBlobContainer`. |
||||
|
|
||||
|
> Read the [BLOB Storing document](../blob-storing) to understand how to use the BLOB storing system. The pipeline is part of the [Volo.Abp.BlobStoring](https://www.nuget.org/packages/Volo.Abp.BlobStoring) package; no additional package is needed. |
||||
|
|
||||
|
## Creating a Pipeline Contributor |
||||
|
|
||||
|
A pipeline contributor implements the `IBlobPipelineContributor` interface. The following example compresses the BLOBs with GZip: |
||||
|
|
||||
|
````csharp |
||||
|
public class GZipBlobPipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
public async Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
var compressedStream = new MemoryStream(); |
||||
|
try |
||||
|
{ |
||||
|
using (var gzipStream = new GZipStream(compressedStream, CompressionLevel.Fastest, leaveOpen: true)) |
||||
|
{ |
||||
|
await context.BlobStream.CopyToAsync(gzipStream, context.CancellationToken); |
||||
|
} |
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
// A stream is only tracked for disposal by the pipeline once it is assigned |
||||
|
// to context.BlobStream, so dispose it here if the eager work fails first |
||||
|
compressedStream.Dispose(); |
||||
|
throw; |
||||
|
} |
||||
|
|
||||
|
compressedStream.Position = 0; |
||||
|
context.BlobStream = compressedStream; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
context.BlobStream = new GZipStream(context.BlobStream, CompressionMode.Decompress); |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
} |
||||
|
```` |
||||
|
|
||||
|
* `OnSavingAsync` is called before the BLOB reaches the storage provider. Replace `context.BlobStream` with the transformed content; it is also allowed to materialize the content eagerly, like the example does (a lazily transforming, read-only wrapper keeps the memory usage constant instead, which is preferable for large BLOBs). |
||||
|
* `OnGettingAsync` is called after the BLOB was read from the storage provider, in the reverse direction of `OnSavingAsync`. |
||||
|
* The `BlobPipelineContext` also provides the normalized container/BLOB names, the container configuration, the tenant id and a scoped `ServiceProvider`. Contributors are resolved from the [dependency injection](../../fundamentals/dependency-injection.md) system (register them like any other service, for example with `ITransientDependency`). While saving, the scope stays alive until the save operation completes; while getting, until the stream returned to the caller is disposed. |
||||
|
|
||||
|
### The Stream Ownership Contract |
||||
|
|
||||
|
* If a stream (or the DI scope) fails to dispose **after** the storage provider already saved the BLOB, `SaveAsync` still throws that cleanup error even though the data is committed — a retry with the default `overrideExisting: false` would then get a `BlobAlreadyExistsException`. |
||||
|
* **While saving**, do not dispose the stream you received (notice the `leaveOpen: true` in the example): every stream you assign to `context.BlobStream` is disposed after the save, while the original stream stays owned by the caller. A stream is only tracked from the moment it is assigned, so if you create a stream and then do work that may fail (like the eager copy above) before assigning it, dispose it yourself on the failure path. |
||||
|
* **While getting**, the stream you set must dispose the stream you received when it is disposed (a `GZipStream` already does that by default), because the composed stream is returned to the caller as a whole. |
||||
|
|
||||
|
## Configuring Containers |
||||
|
|
||||
|
Contributors are configured **per container**, like the other container options: |
||||
|
|
||||
|
````csharp |
||||
|
Configure<AbpBlobStoringOptions>(options => |
||||
|
{ |
||||
|
options.Containers.Configure<ProfilePictureContainer>(container => |
||||
|
{ |
||||
|
container.PipelineContributors.Add<GZipBlobPipelineContributor>(); |
||||
|
}); |
||||
|
}); |
||||
|
```` |
||||
|
|
||||
|
Configuring the default container applies the contributor to all containers; a named container can add its own contributors on top of them: |
||||
|
|
||||
|
````csharp |
||||
|
Configure<AbpBlobStoringOptions>(options => |
||||
|
{ |
||||
|
options.Containers.ConfigureDefault(container => |
||||
|
{ |
||||
|
container.PipelineContributors.Add<GZipBlobPipelineContributor>(); |
||||
|
}); |
||||
|
|
||||
|
options.Containers.Configure<ProfilePictureContainer>(container => |
||||
|
{ |
||||
|
// Runs after the GZip contributor while saving |
||||
|
container.PipelineContributors.Add<WatermarkPipelineContributor>(); |
||||
|
}); |
||||
|
}); |
||||
|
```` |
||||
|
|
||||
|
Think of the composition as **global stages plus container stages**: the contributors of the default container run first while saving, then the own ones of the container (each contributor type runs once). The inherited contributors are kept even when a container overrides its storage provider; set `InheritPipelineContributors` to `false` on a container to opt out of the global stages completely: |
||||
|
|
||||
|
````csharp |
||||
|
options.Containers.Configure<PublicPictureContainer>(container => |
||||
|
{ |
||||
|
container.InheritPipelineContributors = false; |
||||
|
}); |
||||
|
```` |
||||
|
|
||||
|
## Execution Order and Encryption |
||||
|
|
||||
|
* While **saving**, the contributors run in the configuration order, and the built-in [encryption](./encryption.md) always runs **after** them (immediately before the storage provider). |
||||
|
* While **getting**, the decryption runs first and the contributors run in the **reverse** order. |
||||
|
|
||||
|
So, contributors always work on the plain content, a compressing contributor always compresses before the encryption (encrypted data can not be compressed), and the stored form is always ciphertext when the encryption is enabled. |
||||
|
|
||||
|
## Behavioral Notes |
||||
|
|
||||
|
* The stream returned for a container with contributors is generally read-only and non-seekable, and its `Length` is only available when the transformation can provide it. See the behavioral notes of the [BLOB Encryption document](./encryption.md) — the same stream semantics apply to the pipeline. |
||||
|
* When a contributor changes the content size lazily, the final length is unknown to the storage provider; providers that require the object size before uploading need an eagerly materialized (or length-aware) stream. |
||||
|
* Some storage providers consume the stream **synchronously** (like the Aliyun provider); they require contributor streams that also support synchronous reads, exactly like they do without the pipeline. |
||||
|
* Containers without contributors are not affected at all. |
||||
|
|
||||
|
> **A contributor that transforms the content is part of the persisted data format.** A BLOB is only readable with the same transforming contributors, in the same order, it was saved with: adding, removing or re-ordering **transforming** contributors on a container that already has BLOBs makes the existing content fail to be read (or, for transformations without an own format check, silently return wrong content). A **metadata-only** contributor that neither consumes nor replaces `context.BlobStream` does not change the stored format, so it can be added to a container with existing BLOBs. A contributor that reads the content to validate it must return a pass-through wrapper (it still counts as consuming the stream); not replacing the stream after reading it would leave an empty/truncated stream for the provider. To change transforming contributors, migrate by reading the BLOBs **with the old configuration** and exporting the plain content, applying the change, and then writing the content back; re-saving in place under the old configuration does not change the stored form. |
||||
|
|
||||
|
## See Also |
||||
|
|
||||
|
* [BLOB Storing](../blob-storing) |
||||
|
* [BLOB Encryption](./encryption.md) |
||||
|
* [Creating a custom BLOB storage provider](./custom-provider.md) |
||||
@ -0,0 +1,3 @@ |
|||||
|
using System.Runtime.CompilerServices; |
||||
|
|
||||
|
[assembly: InternalsVisibleTo("Volo.Abp.BlobStoring.Aws.Tests")] |
||||
@ -0,0 +1,90 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Aws; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The unseekable multipart path of the AWS SDK ignores AutoCloseStream and
|
||||
|
/// disposes its input; this wrapper protects the ownership of the wrapped stream.
|
||||
|
/// </summary>
|
||||
|
internal sealed class LeaveOpenStreamWrapper : Stream |
||||
|
{ |
||||
|
private readonly Stream _inner; |
||||
|
|
||||
|
public LeaveOpenStreamWrapper(Stream inner) |
||||
|
{ |
||||
|
_inner = inner; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead => _inner.CanRead; |
||||
|
public override bool CanSeek => _inner.CanSeek; |
||||
|
public override bool CanWrite => false; |
||||
|
|
||||
|
// The SDK computes the optional content length from Length/Position, but only
|
||||
|
// handles NotSupportedException; translate an IOException of a probe, so an
|
||||
|
// unknown length stays "unknown" instead of failing the upload
|
||||
|
public override long Length |
||||
|
{ |
||||
|
get |
||||
|
{ |
||||
|
try |
||||
|
{ |
||||
|
return _inner.Length; |
||||
|
} |
||||
|
catch (IOException ex) |
||||
|
{ |
||||
|
throw new NotSupportedException("The length of the stream is not available!", ex); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get |
||||
|
{ |
||||
|
try |
||||
|
{ |
||||
|
return _inner.Position; |
||||
|
} |
||||
|
catch (IOException ex) |
||||
|
{ |
||||
|
throw new NotSupportedException("The position of the stream is not available!", ex); |
||||
|
} |
||||
|
} |
||||
|
set => _inner.Position = value; |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) => _inner.Read(buffer, offset, count); |
||||
|
|
||||
|
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) |
||||
|
{ |
||||
|
#if NETSTANDARD2_0
|
||||
|
return _inner.ReadAsync(buffer, offset, count, cancellationToken); |
||||
|
#else
|
||||
|
// The SDK reads over this (old) overload; dispatch it over the modern one,
|
||||
|
// so a source that only implements ReadAsync(Memory<byte>) keeps working
|
||||
|
return _inner.ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask(); |
||||
|
#endif
|
||||
|
} |
||||
|
|
||||
|
#if !NETSTANDARD2_0
|
||||
|
public override int Read(Span<byte> buffer) => _inner.Read(buffer); |
||||
|
|
||||
|
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) |
||||
|
{ |
||||
|
return _inner.ReadAsync(buffer, cancellationToken); |
||||
|
} |
||||
|
#endif
|
||||
|
|
||||
|
public override long Seek(long offset, SeekOrigin origin) => _inner.Seek(offset, origin); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
// Disposing the wrapper must not dispose the wrapped stream
|
||||
|
} |
||||
@ -0,0 +1,8 @@ |
|||||
|
namespace Volo.Abp.BlobStoring; |
||||
|
|
||||
|
public static class BlobEncryptionConfigurationNames |
||||
|
{ |
||||
|
public const string Enabled = "BlobEncryption.Enabled"; |
||||
|
public const string PassPhrase = "BlobEncryption.PassPhrase"; |
||||
|
public const string AllowLegacyPlainText = "BlobEncryption.AllowLegacyPlainText"; |
||||
|
} |
||||
@ -0,0 +1,51 @@ |
|||||
|
using System; |
||||
|
using JetBrains.Annotations; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The identity of the BLOB an encryption key is resolved for. It lets a custom
|
||||
|
/// <see cref="IBlobEncryptionKeyProvider"/> select the key by the container, the
|
||||
|
/// BLOB name or the tenant — not only by the container configuration.
|
||||
|
/// </summary>
|
||||
|
public class BlobEncryptionKeyContext |
||||
|
{ |
||||
|
/// <summary>
|
||||
|
/// The configuration of the container the BLOB belongs to (with the container
|
||||
|
/// passphrase, if one was set with <c>UseEncryption</c>).
|
||||
|
/// </summary>
|
||||
|
[NotNull] |
||||
|
public BlobContainerConfiguration Configuration { get; } |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The normalized container name.
|
||||
|
/// </summary>
|
||||
|
[NotNull] |
||||
|
public string ContainerName { get; } |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The normalized BLOB name.
|
||||
|
/// </summary>
|
||||
|
[NotNull] |
||||
|
public string BlobName { get; } |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The tenant of the BLOB operation (null for the host or a shared container).
|
||||
|
/// </summary>
|
||||
|
public Guid? TenantId { get; } |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Creates the context; the names are expected in their normalized form.
|
||||
|
/// </summary>
|
||||
|
public BlobEncryptionKeyContext( |
||||
|
[NotNull] BlobContainerConfiguration configuration, |
||||
|
[NotNull] string containerName, |
||||
|
[NotNull] string blobName, |
||||
|
Guid? tenantId) |
||||
|
{ |
||||
|
Configuration = Check.NotNull(configuration, nameof(configuration)); |
||||
|
ContainerName = Check.NotNullOrWhiteSpace(containerName, nameof(containerName)); |
||||
|
BlobName = Check.NotNullOrWhiteSpace(blobName, nameof(blobName)); |
||||
|
TenantId = tenantId; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,121 @@ |
|||||
|
using System; |
||||
|
using System.Collections.Generic; |
||||
|
using System.IO; |
||||
|
using System.Threading; |
||||
|
using JetBrains.Annotations; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The context an <see cref="IBlobPipelineContributor"/> works on. A contributor
|
||||
|
/// transforms the content by replacing <see cref="BlobStream"/> with a wrapper;
|
||||
|
/// see <see cref="IBlobPipelineContributor"/> for the stream ownership contract.
|
||||
|
/// </summary>
|
||||
|
public class BlobPipelineContext : IServiceProviderAccessor |
||||
|
{ |
||||
|
/// <summary>
|
||||
|
/// The scoped service provider of the pipeline. While saving, the scope stays
|
||||
|
/// alive until the save operation completes; while getting, until the stream
|
||||
|
/// returned to the caller is disposed — so lazily transforming wrappers can
|
||||
|
/// keep using their scoped services.
|
||||
|
/// </summary>
|
||||
|
[NotNull] |
||||
|
public IServiceProvider ServiceProvider { get; } |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The normalized container name.
|
||||
|
/// </summary>
|
||||
|
[NotNull] |
||||
|
public string ContainerName { get; } |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The normalized BLOB name.
|
||||
|
/// </summary>
|
||||
|
[NotNull] |
||||
|
public string BlobName { get; } |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The configuration of the container the BLOB belongs to.
|
||||
|
/// </summary>
|
||||
|
[NotNull] |
||||
|
public BlobContainerConfiguration Configuration { get; } |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The tenant of the BLOB operation (null for the host or a shared container).
|
||||
|
/// </summary>
|
||||
|
public Guid? TenantId { get; } |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The cancellation token of the BLOB operation. Pass it to any I/O the contributor
|
||||
|
/// performs while <see cref="IBlobPipelineContributor.OnSavingAsync"/> /
|
||||
|
/// <see cref="IBlobPipelineContributor.OnGettingAsync"/> runs. A lazy read wrapper
|
||||
|
/// returned from <c>OnGettingAsync</c> must instead honor the token passed to each of
|
||||
|
/// its own <c>Read</c>/<c>ReadAsync</c> calls (this token is captured once at
|
||||
|
/// <c>GetAsync</c> time and is not updated per read).
|
||||
|
/// </summary>
|
||||
|
public CancellationToken CancellationToken { get; } |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// The content stream. Replace it with a (typically lazily transforming,
|
||||
|
/// read-only) wrapper to transform the content.
|
||||
|
/// </summary>
|
||||
|
[NotNull] |
||||
|
public Stream BlobStream { |
||||
|
get => _blobStream; |
||||
|
set |
||||
|
{ |
||||
|
_blobStream = Check.NotNull(value, nameof(value)); |
||||
|
TrackCreatedStream(value); |
||||
|
} |
||||
|
} |
||||
|
private Stream _blobStream; |
||||
|
|
||||
|
private readonly Stream _initialStream; |
||||
|
|
||||
|
// While saving, every stream the pipeline creates is collected here (at
|
||||
|
// assignment, so intermediate replacements within one contributor call are
|
||||
|
// not lost) to be disposed after the save; the initial (caller-owned)
|
||||
|
// stream is never collected. Null while getting.
|
||||
|
internal List<Stream>? CreatedStreams { get; set; } |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Creates the context; the names are expected in their normalized form and
|
||||
|
/// <paramref name="blobStream"/> is the initial (untransformed) content.
|
||||
|
/// </summary>
|
||||
|
public BlobPipelineContext( |
||||
|
[NotNull] IServiceProvider serviceProvider, |
||||
|
[NotNull] string containerName, |
||||
|
[NotNull] string blobName, |
||||
|
[NotNull] BlobContainerConfiguration configuration, |
||||
|
Guid? tenantId, |
||||
|
[NotNull] Stream blobStream, |
||||
|
CancellationToken cancellationToken = default) |
||||
|
{ |
||||
|
ServiceProvider = Check.NotNull(serviceProvider, nameof(serviceProvider)); |
||||
|
ContainerName = Check.NotNullOrWhiteSpace(containerName, nameof(containerName)); |
||||
|
BlobName = Check.NotNullOrWhiteSpace(blobName, nameof(blobName)); |
||||
|
Configuration = Check.NotNull(configuration, nameof(configuration)); |
||||
|
TenantId = tenantId; |
||||
|
_initialStream = _blobStream = Check.NotNull(blobStream, nameof(blobStream)); |
||||
|
CancellationToken = cancellationToken; |
||||
|
} |
||||
|
|
||||
|
private void TrackCreatedStream(Stream stream) |
||||
|
{ |
||||
|
if (CreatedStreams == null || ReferenceEquals(stream, _initialStream)) |
||||
|
{ |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
foreach (var existingStream in CreatedStreams) |
||||
|
{ |
||||
|
if (ReferenceEquals(existingStream, stream)) |
||||
|
{ |
||||
|
return; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
CreatedStreams.Add(stream); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,488 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
using Microsoft.Extensions.DependencyInjection; |
||||
|
using Volo.Abp.MultiTenancy; |
||||
|
using Volo.Abp.Threading; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Keeps the service scope and the tenant context of the pipeline contributors
|
||||
|
/// until the returned (lazily transforming) stream is disposed: every member the
|
||||
|
/// wrappers may run work in — including the disposal of the scope itself — executes
|
||||
|
/// in the tenant the BLOB belongs to, not in the ambient tenant of the caller.
|
||||
|
/// </summary>
|
||||
|
internal sealed class BlobPipelineScopeStream : Stream |
||||
|
{ |
||||
|
private readonly Stream _inner; |
||||
|
private readonly AsyncServiceScope _scope; |
||||
|
private readonly ICurrentTenant _currentTenant; |
||||
|
private readonly Guid? _tenantId; |
||||
|
private readonly IBlobAuthenticatedEndStream? _authenticatedEndSource; |
||||
|
private bool _authenticatedEndChecked; |
||||
|
private bool _faulted; |
||||
|
private bool _disposed; |
||||
|
|
||||
|
public BlobPipelineScopeStream( |
||||
|
Stream inner, |
||||
|
AsyncServiceScope scope, |
||||
|
ICurrentTenant currentTenant, |
||||
|
Guid? tenantId, |
||||
|
IBlobAuthenticatedEndStream? authenticatedEndSource = null) |
||||
|
{ |
||||
|
_inner = inner; |
||||
|
_scope = scope; |
||||
|
_currentTenant = currentTenant; |
||||
|
_tenantId = tenantId; |
||||
|
// When encryption is enabled, this is the innermost decrypting stream. Its
|
||||
|
// terminal record is verified when this composed stream reaches EOF, so a
|
||||
|
// contributor that stops before the content ends can not hide a truncation.
|
||||
|
_authenticatedEndSource = authenticatedEndSource; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead |
||||
|
{ |
||||
|
get |
||||
|
{ |
||||
|
if (_disposed) |
||||
|
{ |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
return _inner.CanRead; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private void EnsureNotDisposed() |
||||
|
{ |
||||
|
if (_disposed) |
||||
|
{ |
||||
|
throw new ObjectDisposedException(GetType().FullName); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override bool CanSeek |
||||
|
{ |
||||
|
get |
||||
|
{ |
||||
|
if (_disposed) |
||||
|
{ |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
return _inner.CanSeek; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override bool CanWrite => false; |
||||
|
|
||||
|
public override bool CanTimeout |
||||
|
{ |
||||
|
get |
||||
|
{ |
||||
|
if (_disposed) |
||||
|
{ |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
return _inner.CanTimeout; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override int ReadTimeout |
||||
|
{ |
||||
|
get |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
return _inner.ReadTimeout; |
||||
|
} |
||||
|
} |
||||
|
set |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
_inner.ReadTimeout = value; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override long Length |
||||
|
{ |
||||
|
get |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
return _inner.Length; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
return _inner.Position; |
||||
|
} |
||||
|
} |
||||
|
set |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
_inner.Position = value; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
_inner.Flush(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override async Task FlushAsync(CancellationToken cancellationToken) |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
await _inner.FlushAsync(cancellationToken); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
EnsureNotFaulted(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
int read; |
||||
|
try |
||||
|
{ |
||||
|
read = _inner.Read(buffer, offset, count); |
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
// A failed read faults permanently, so a retry layer can not silently continue
|
||||
|
// from a position where an inner contributor already consumed bytes
|
||||
|
_faulted = true; |
||||
|
throw; |
||||
|
} |
||||
|
|
||||
|
VerifyAuthenticatedEndIfNeeded(read, count); |
||||
|
return read; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
EnsureNotFaulted(); |
||||
|
// A token cancelled before any I/O leaves the stream untouched (and healthy for a
|
||||
|
// retry); once a read has started, any failure faults it permanently
|
||||
|
cancellationToken.ThrowIfCancellationRequested(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
int read; |
||||
|
try |
||||
|
{ |
||||
|
#if NETSTANDARD2_0
|
||||
|
read = await _inner.ReadAsync(buffer, offset, count, cancellationToken); |
||||
|
#else
|
||||
|
// Dispatch over the modern overload, so a wrapper that only implements
|
||||
|
// ReadAsync(Memory<byte>) also works for callers of the old overload
|
||||
|
read = await _inner.ReadAsync(buffer.AsMemory(offset, count), cancellationToken); |
||||
|
#endif
|
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
_faulted = true; |
||||
|
throw; |
||||
|
} |
||||
|
|
||||
|
await VerifyAuthenticatedEndIfNeededAsync(read, count, cancellationToken); |
||||
|
return read; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
// Runs when the composed stream reaches EOF on a real (non-zero-count) read. A
|
||||
|
// legitimate partial read (stopping early and disposing) never reaches EOF, so it
|
||||
|
// is not affected. A failed check faults the stream permanently, so a read-retry
|
||||
|
// layer can not swallow the integrity error and later see a normal EOF.
|
||||
|
private void VerifyAuthenticatedEndIfNeeded(int read, int count) |
||||
|
{ |
||||
|
if (read != 0 || count == 0 || _authenticatedEndChecked || _authenticatedEndSource == null) |
||||
|
{ |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
try |
||||
|
{ |
||||
|
_authenticatedEndSource.EnsureReadToAuthenticatedEnd(); |
||||
|
_authenticatedEndChecked = true; |
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
// The check ran and failed on integrity: mark it done and fault permanently
|
||||
|
_authenticatedEndChecked = true; |
||||
|
_faulted = true; |
||||
|
throw; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private async ValueTask VerifyAuthenticatedEndIfNeededAsync(int read, int count, CancellationToken cancellationToken) |
||||
|
{ |
||||
|
if (read != 0 || count == 0 || _authenticatedEndChecked || _authenticatedEndSource == null) |
||||
|
{ |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
// A token cancelled before the check runs leaves it un-run and the stream healthy, so
|
||||
|
// a retry with a live token can still verify the end
|
||||
|
cancellationToken.ThrowIfCancellationRequested(); |
||||
|
try |
||||
|
{ |
||||
|
await _authenticatedEndSource.EnsureReadToAuthenticatedEndAsync(cancellationToken); |
||||
|
_authenticatedEndChecked = true; |
||||
|
} |
||||
|
catch (OperationCanceledException) |
||||
|
{ |
||||
|
// The decrypting stream owns the consumption state and faults itself on a mid-read
|
||||
|
// cancellation; a cancellation it lets through without faulting (for example the
|
||||
|
// token trips in the gap after the check above) leaves it healthy, so the outer
|
||||
|
// must not fault either — a retry with a live token can still verify the end
|
||||
|
throw; |
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
// A real integrity failure is permanent, so a read-retry layer can not swallow it
|
||||
|
// and later see a normal EOF
|
||||
|
_authenticatedEndChecked = true; |
||||
|
_faulted = true; |
||||
|
throw; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private void EnsureNotFaulted() |
||||
|
{ |
||||
|
if (_faulted) |
||||
|
{ |
||||
|
throw new AbpException("The stream can not be read anymore, because a previous read operation has failed!"); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
#if !NETSTANDARD2_0
|
||||
|
// Forwarded so a wrapper that only implements the modern overloads is not
|
||||
|
// degraded to the byte[] fallback of the base class
|
||||
|
public override int Read(Span<byte> buffer) |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
EnsureNotFaulted(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
int read; |
||||
|
try |
||||
|
{ |
||||
|
read = _inner.Read(buffer); |
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
_faulted = true; |
||||
|
throw; |
||||
|
} |
||||
|
|
||||
|
VerifyAuthenticatedEndIfNeeded(read, buffer.Length); |
||||
|
return read; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
EnsureNotFaulted(); |
||||
|
cancellationToken.ThrowIfCancellationRequested(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
int read; |
||||
|
try |
||||
|
{ |
||||
|
read = await _inner.ReadAsync(buffer, cancellationToken); |
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
_faulted = true; |
||||
|
throw; |
||||
|
} |
||||
|
|
||||
|
await VerifyAuthenticatedEndIfNeededAsync(read, buffer.Length, cancellationToken); |
||||
|
return read; |
||||
|
} |
||||
|
} |
||||
|
#endif
|
||||
|
|
||||
|
public override long Seek(long offset, SeekOrigin origin) |
||||
|
{ |
||||
|
EnsureNotDisposed(); |
||||
|
using (_currentTenant.Change(_tenantId)) |
||||
|
{ |
||||
|
return _inner.Seek(offset, origin); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override void SetLength(long value) |
||||
|
{ |
||||
|
throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Write(byte[] buffer, int offset, int count) |
||||
|
{ |
||||
|
throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
protected override void Dispose(bool disposing) |
||||
|
{ |
||||
|
if (disposing && !_disposed) |
||||
|
{ |
||||
|
_disposed = true; |
||||
|
|
||||
|
// The tenant context is best-effort: failing to enter it must not skip the
|
||||
|
// resource release (which would leak the provider stream, the scope and the
|
||||
|
// derived key), and a later dispose can not recover it since _disposed is set
|
||||
|
IDisposable? tenantChange = null; |
||||
|
try |
||||
|
{ |
||||
|
tenantChange = _currentTenant.Change(_tenantId); |
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
// ignored: release the resources below without the tenant context
|
||||
|
} |
||||
|
|
||||
|
try |
||||
|
{ |
||||
|
DisposeInnerAndScope(); |
||||
|
} |
||||
|
finally |
||||
|
{ |
||||
|
tenantChange?.Dispose(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
base.Dispose(disposing); |
||||
|
} |
||||
|
|
||||
|
private void DisposeInnerAndScope() |
||||
|
{ |
||||
|
try |
||||
|
{ |
||||
|
#if NETSTANDARD2_0
|
||||
|
// Stream has no DisposeAsync on netstandard2.0, but the inner stream
|
||||
|
// may still implement IAsyncDisposable for its async-only cleanup
|
||||
|
if (_inner is IAsyncDisposable innerAsyncDisposable) |
||||
|
{ |
||||
|
AsyncHelper.RunSync(() => innerAsyncDisposable.DisposeAsync().AsTask()); |
||||
|
} |
||||
|
else |
||||
|
{ |
||||
|
_inner.Dispose(); |
||||
|
} |
||||
|
#else
|
||||
|
// Also covers wrappers that only implement DisposeAsync
|
||||
|
AsyncHelper.RunSync(() => _inner.DisposeAsync().AsTask()); |
||||
|
#endif
|
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
// The stream failure is the root cause; a scope dispose
|
||||
|
// failure on top of it must not replace it
|
||||
|
try |
||||
|
{ |
||||
|
AsyncHelper.RunSync(() => _scope.DisposeAsync().AsTask()); |
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
// ignored
|
||||
|
} |
||||
|
|
||||
|
throw; |
||||
|
} |
||||
|
|
||||
|
// A synchronous scope dispose throws when a scoped service only
|
||||
|
// implements IAsyncDisposable, so always release it asynchronously
|
||||
|
AsyncHelper.RunSync(() => _scope.DisposeAsync().AsTask()); |
||||
|
} |
||||
|
|
||||
|
#if !NETSTANDARD2_0
|
||||
|
public override async ValueTask DisposeAsync() |
||||
|
{ |
||||
|
if (!_disposed) |
||||
|
{ |
||||
|
_disposed = true; |
||||
|
|
||||
|
IDisposable? tenantChange = null; |
||||
|
try |
||||
|
{ |
||||
|
tenantChange = _currentTenant.Change(_tenantId); |
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
// ignored: release the resources below without the tenant context
|
||||
|
} |
||||
|
|
||||
|
try |
||||
|
{ |
||||
|
await DisposeInnerAndScopeAsync(); |
||||
|
} |
||||
|
finally |
||||
|
{ |
||||
|
tenantChange?.Dispose(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
await base.DisposeAsync(); |
||||
|
} |
||||
|
|
||||
|
private async ValueTask DisposeInnerAndScopeAsync() |
||||
|
{ |
||||
|
try |
||||
|
{ |
||||
|
await _inner.DisposeAsync(); |
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
try |
||||
|
{ |
||||
|
await _scope.DisposeAsync(); |
||||
|
} |
||||
|
catch |
||||
|
{ |
||||
|
// ignored: the stream failure is the root cause
|
||||
|
} |
||||
|
|
||||
|
throw; |
||||
|
} |
||||
|
|
||||
|
await _scope.DisposeAsync(); |
||||
|
} |
||||
|
#endif
|
||||
|
} |
||||
@ -0,0 +1,26 @@ |
|||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Implemented by a read stream (like the decrypting stream) that can verify it was
|
||||
|
/// read to an authenticated end. The content pipeline calls it when the composed
|
||||
|
/// stream returned by <c>GetAsync</c> reaches EOF, so a contributor stopping before
|
||||
|
/// the end can not hide a truncation. A stream that wraps such a stream (for example
|
||||
|
/// a custom <c>CreateDecryptingStreamAsync</c> override) should implement this
|
||||
|
/// interface too and forward the calls to the wrapped stream, or the end verification
|
||||
|
/// is skipped for pipeline reads.
|
||||
|
/// </summary>
|
||||
|
public interface IBlobAuthenticatedEndStream |
||||
|
{ |
||||
|
/// <summary>
|
||||
|
/// Throws if the stream has not been consumed up to its authenticated end.
|
||||
|
/// </summary>
|
||||
|
void EnsureReadToAuthenticatedEnd(); |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Throws if the stream has not been consumed up to its authenticated end.
|
||||
|
/// </summary>
|
||||
|
ValueTask EnsureReadToAuthenticatedEndAsync(CancellationToken cancellationToken = default); |
||||
|
} |
||||
@ -0,0 +1,41 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
using JetBrains.Annotations; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Encrypts and decrypts the BLOB content stream (authenticated, chunked AES-256-GCM).
|
||||
|
/// Replace this service to change the encryption format or algorithm; the built-in
|
||||
|
/// <see cref="BlobEncryptionCodec"/> implements version 1 of the format.
|
||||
|
/// </summary>
|
||||
|
public interface IBlobEncryptionCodec |
||||
|
{ |
||||
|
/// <summary>
|
||||
|
/// Wraps <paramref name="plainStream"/> in a read-only stream that encrypts the
|
||||
|
/// content while it is read. The container and BLOB names are expected in their
|
||||
|
/// normalized form.
|
||||
|
/// </summary>
|
||||
|
Task<Stream> CreateEncryptingStreamAsync( |
||||
|
[NotNull] BlobContainerConfiguration configuration, |
||||
|
[NotNull] string containerName, |
||||
|
[NotNull] string blobName, |
||||
|
Guid? tenantId, |
||||
|
[NotNull] Stream plainStream, |
||||
|
CancellationToken cancellationToken = default); |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Wraps <paramref name="cipherStream"/> in a read-only stream that decrypts the
|
||||
|
/// content while it is read. The container and BLOB names are expected in their
|
||||
|
/// normalized form.
|
||||
|
/// </summary>
|
||||
|
Task<Stream> CreateDecryptingStreamAsync( |
||||
|
[NotNull] BlobContainerConfiguration configuration, |
||||
|
[NotNull] string containerName, |
||||
|
[NotNull] string blobName, |
||||
|
Guid? tenantId, |
||||
|
[NotNull] Stream cipherStream, |
||||
|
CancellationToken cancellationToken = default); |
||||
|
} |
||||
@ -0,0 +1,41 @@ |
|||||
|
using System.Threading.Tasks; |
||||
|
using JetBrains.Annotations; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Transforms the BLOB content stream (compression, watermarking, validation...)
|
||||
|
/// while it is saved and read. Contributors are configured per container with
|
||||
|
/// <see cref="BlobContainerConfiguration.PipelineContributors"/> and run in the
|
||||
|
/// configuration order while saving and in the reverse order while reading.
|
||||
|
/// The built-in encryption always runs after the contributors while saving (and
|
||||
|
/// before them while reading), so contributors always work on the plain content.
|
||||
|
/// </summary>
|
||||
|
public interface IBlobPipelineContributor |
||||
|
{ |
||||
|
/// <summary>
|
||||
|
/// Transform the content by replacing <see cref="BlobPipelineContext.BlobStream"/>:
|
||||
|
/// with a lazily transforming read-only wrapper (best for large content), or with an
|
||||
|
/// eagerly materialized stream. A replacement must leave the stream it received open:
|
||||
|
/// every stream <b>assigned</b> to <see cref="BlobPipelineContext.BlobStream"/> is disposed
|
||||
|
/// after the save, while the original stream stays owned by the caller. A stream is only
|
||||
|
/// tracked from the moment it is assigned, so if you create a stream and then do work
|
||||
|
/// that may fail before assigning it, dispose it yourself on the failure path.
|
||||
|
/// <para>
|
||||
|
/// Not replacing the stream is only valid for a contributor that does not consume the
|
||||
|
/// content (for example a metadata check). A contributor that reads the content to
|
||||
|
/// validate it must return a pass-through wrapper that validates the bytes as they
|
||||
|
/// flow (or an eagerly materialized replacement) — reading the content without
|
||||
|
/// replacing the stream would leave an empty/truncated stream for the provider.
|
||||
|
/// </para>
|
||||
|
/// </summary>
|
||||
|
Task OnSavingAsync([NotNull] BlobPipelineContext context); |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Reverse the save-time transformation by replacing
|
||||
|
/// <see cref="BlobPipelineContext.BlobStream"/> the same way. Here a replacement
|
||||
|
/// must dispose the stream it received when it is disposed, since the composed
|
||||
|
/// stream is returned to the caller as a whole.
|
||||
|
/// </summary>
|
||||
|
Task OnGettingAsync([NotNull] BlobPipelineContext context); |
||||
|
} |
||||
@ -0,0 +1,104 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Threading.Tasks; |
||||
|
using Shouldly; |
||||
|
using Xunit; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Aws; |
||||
|
|
||||
|
public class AwsBlobProviderUploadDecision_Tests |
||||
|
{ |
||||
|
private readonly ExposedAwsBlobProvider _provider = new ExposedAwsBlobProvider(); |
||||
|
|
||||
|
[Fact] |
||||
|
public void Should_Keep_The_Plain_PutObject_Behavior_For_Untransformed_Containers() |
||||
|
{ |
||||
|
// A non-seekable stream of a container without encryption/pipeline
|
||||
|
// must be uploaded exactly like before
|
||||
|
var args = CreateArgs(new BlobContainerConfiguration(), new NonSeekableStream()); |
||||
|
|
||||
|
_provider.RequiresRetrySafeUploadPublic(args).ShouldBeFalse(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public void Should_Use_The_Retry_Safe_Upload_For_An_Encrypted_Container() |
||||
|
{ |
||||
|
var configuration = new BlobContainerConfiguration().UseEncryption("test-passphrase"); |
||||
|
var args = CreateArgs(configuration, new NonSeekableStream()); |
||||
|
|
||||
|
_provider.RequiresRetrySafeUploadPublic(args).ShouldBeTrue(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public void Should_Use_The_Retry_Safe_Upload_For_A_Container_With_PipelineContributors() |
||||
|
{ |
||||
|
var configuration = new BlobContainerConfiguration(); |
||||
|
configuration.PipelineContributors.Add<FakePipelineContributor>(); |
||||
|
var args = CreateArgs(configuration, new NonSeekableStream()); |
||||
|
|
||||
|
_provider.RequiresRetrySafeUploadPublic(args).ShouldBeTrue(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public void Should_Not_Use_The_Retry_Safe_Upload_For_A_Seekable_Stream() |
||||
|
{ |
||||
|
var configuration = new BlobContainerConfiguration().UseEncryption("test-passphrase"); |
||||
|
var args = CreateArgs(configuration, new MemoryStream()); |
||||
|
|
||||
|
_provider.RequiresRetrySafeUploadPublic(args).ShouldBeFalse(); |
||||
|
} |
||||
|
|
||||
|
private static BlobProviderSaveArgs CreateArgs(BlobContainerConfiguration configuration, Stream stream) |
||||
|
{ |
||||
|
return new BlobProviderSaveArgs("test-container", configuration, "test-blob", stream); |
||||
|
} |
||||
|
|
||||
|
private sealed class ExposedAwsBlobProvider : AwsBlobProvider |
||||
|
{ |
||||
|
public ExposedAwsBlobProvider() |
||||
|
: base(null!, null!, null!) |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public bool RequiresRetrySafeUploadPublic(BlobProviderSaveArgs args) |
||||
|
{ |
||||
|
return RequiresRetrySafeUpload(args); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private sealed class FakePipelineContributor : IBlobPipelineContributor |
||||
|
{ |
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private sealed class NonSeekableStream : Stream |
||||
|
{ |
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) => 0; |
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,84 @@ |
|||||
|
using System.IO; |
||||
|
using Amazon.S3.Model; |
||||
|
using Amazon.S3.Transfer; |
||||
|
using Shouldly; |
||||
|
using Xunit; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Aws; |
||||
|
|
||||
|
public class AwsBlobProviderUploadRequest_Tests |
||||
|
{ |
||||
|
private readonly ExposedAwsBlobProvider _provider = new ExposedAwsBlobProvider(); |
||||
|
|
||||
|
[Fact] |
||||
|
public void Should_Wrap_The_Stream_And_Disable_Auto_Close_For_A_Multipart_Request() |
||||
|
{ |
||||
|
// The SDK's non-seekable multipart path ignores AutoCloseStream and disposes the
|
||||
|
// input, so the source must be protected by the leave-open wrapper
|
||||
|
var source = new MemoryStream(); |
||||
|
var configuration = new AwsBlobProviderConfiguration(new BlobContainerConfiguration()) { DisablePayloadSigning = true }; |
||||
|
|
||||
|
var request = _provider.CreateMultipartUploadRequestPublic("bucket", "key", source, configuration); |
||||
|
|
||||
|
request.InputStream.ShouldBeOfType<LeaveOpenStreamWrapper>(); |
||||
|
request.AutoCloseStream.ShouldBeFalse(); |
||||
|
request.DisablePayloadSigning.ShouldBe(true); |
||||
|
request.BucketName.ShouldBe("bucket"); |
||||
|
request.Key.ShouldBe("key"); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public void Should_Keep_The_Source_Open_After_The_Multipart_Input_Is_Disposed() |
||||
|
{ |
||||
|
var source = new MemoryStream(); |
||||
|
var configuration = new AwsBlobProviderConfiguration(new BlobContainerConfiguration()); |
||||
|
|
||||
|
var request = _provider.CreateMultipartUploadRequestPublic("bucket", "key", source, configuration); |
||||
|
request.InputStream.Dispose(); // the SDK disposes the input on the multipart path
|
||||
|
|
||||
|
source.CanRead.ShouldBeTrue(); // the wrapper must have left the source open
|
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public void Should_Propagate_Disable_Payload_Signing_And_Keep_Ownership_For_A_Put_Object_Request() |
||||
|
{ |
||||
|
var source = new MemoryStream(); |
||||
|
var configuration = new AwsBlobProviderConfiguration(new BlobContainerConfiguration()) { DisablePayloadSigning = true }; |
||||
|
|
||||
|
var request = _provider.CreatePutObjectRequestPublic("bucket", "key", source, configuration); |
||||
|
|
||||
|
request.InputStream.ShouldBeSameAs(source); |
||||
|
request.AutoCloseStream.ShouldBeFalse(); |
||||
|
request.DisablePayloadSigning.ShouldBe(true); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public void Should_Keep_Disable_Payload_Signing_Off_By_Default_For_A_Put_Object_Request() |
||||
|
{ |
||||
|
var configuration = new AwsBlobProviderConfiguration(new BlobContainerConfiguration()); |
||||
|
|
||||
|
var request = _provider.CreatePutObjectRequestPublic("bucket", "key", new MemoryStream(), configuration); |
||||
|
|
||||
|
request.DisablePayloadSigning.ShouldBe(false); |
||||
|
} |
||||
|
|
||||
|
private sealed class ExposedAwsBlobProvider : AwsBlobProvider |
||||
|
{ |
||||
|
public ExposedAwsBlobProvider() |
||||
|
: base(null!, null!, null!) |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public TransferUtilityUploadRequest CreateMultipartUploadRequestPublic( |
||||
|
string containerName, string blobName, Stream blobStream, AwsBlobProviderConfiguration configuration) |
||||
|
{ |
||||
|
return CreateMultipartUploadRequest(containerName, blobName, blobStream, configuration); |
||||
|
} |
||||
|
|
||||
|
public PutObjectRequest CreatePutObjectRequestPublic( |
||||
|
string containerName, string blobName, Stream blobStream, AwsBlobProviderConfiguration configuration) |
||||
|
{ |
||||
|
return CreatePutObjectRequest(containerName, blobName, blobStream, configuration); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,54 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Linq; |
||||
|
using System.Threading.Tasks; |
||||
|
using Shouldly; |
||||
|
using Volo.Abp.BlobStoring.Fakes; |
||||
|
using Xunit; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Aws; |
||||
|
|
||||
|
public class LeaveOpenStreamWrapper_Tests |
||||
|
{ |
||||
|
[Fact] |
||||
|
public async Task Should_Bridge_The_Old_Read_Overload_To_The_Modern_One_And_Leave_The_Stream_Open() |
||||
|
{ |
||||
|
var content = new byte[1000]; |
||||
|
new Random(42).NextBytes(content); |
||||
|
using var inner = new FakeModernAsyncOnlyStream(new MemoryStream(content)); |
||||
|
var wrapper = new LeaveOpenStreamWrapper(inner); |
||||
|
|
||||
|
// The SDK reads over the old overload; the inner stream only supports the modern one
|
||||
|
var buffer = new byte[content.Length]; |
||||
|
var totalReadCount = 0; |
||||
|
while (totalReadCount < buffer.Length) |
||||
|
{ |
||||
|
var readCount = await wrapper.ReadAsync(buffer, totalReadCount, buffer.Length - totalReadCount, default); |
||||
|
if (readCount == 0) |
||||
|
{ |
||||
|
break; |
||||
|
} |
||||
|
|
||||
|
totalReadCount += readCount; |
||||
|
} |
||||
|
|
||||
|
totalReadCount.ShouldBe(content.Length); |
||||
|
buffer.SequenceEqual(content).ShouldBeTrue(); |
||||
|
|
||||
|
wrapper.Dispose(); |
||||
|
|
||||
|
// The wrapped stream stays open (a disposed one would throw here)
|
||||
|
(await inner.ReadAsync(new byte[1].AsMemory(), default)).ShouldBe(0); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public void Should_Translate_A_Failing_Length_Probe_For_The_Sdk() |
||||
|
{ |
||||
|
// The SDK only handles NotSupportedException while probing the content length
|
||||
|
using var inner = new FakeIoFailingLengthStream(new MemoryStream(new byte[10])); |
||||
|
var wrapper = new LeaveOpenStreamWrapper(inner); |
||||
|
|
||||
|
Should.Throw<NotSupportedException>(() => wrapper.Length); |
||||
|
Should.Throw<NotSupportedException>(() => wrapper.Position); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,618 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Text; |
||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
using Microsoft.Extensions.DependencyInjection; |
||||
|
using Shouldly; |
||||
|
using Volo.Abp.BlobStoring.Fakes; |
||||
|
using Xunit; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring; |
||||
|
|
||||
|
public class BlobContainerPipeline_Tests : AbpBlobStoringTestBase |
||||
|
{ |
||||
|
private readonly IBlobContainerFactory _blobContainerFactory; |
||||
|
private readonly FakeInMemoryBlobProvider _fakeProvider; |
||||
|
|
||||
|
public BlobContainerPipeline_Tests() |
||||
|
{ |
||||
|
_blobContainerFactory = GetRequiredService<IBlobContainerFactory>(); |
||||
|
_fakeProvider = GetRequiredService<FakeInMemoryBlobProvider>(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Transform_While_Saving_And_Restore_While_Getting() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-markers"); |
||||
|
var content = "pipeline content".GetBytes(); |
||||
|
using var source = new MemoryStream(content); |
||||
|
|
||||
|
await container.SaveAsync("markers-blob", source); |
||||
|
|
||||
|
// Contributors run in the configuration order while saving: A wraps first,
|
||||
|
// B wraps the result, so the stored form starts with the marker of B
|
||||
|
var rawBytes = _fakeProvider.GetRawBytesOrNull("pipeline-markers", "markers-blob"); |
||||
|
rawBytes.ShouldNotBeNull(); |
||||
|
Encoding.UTF8.GetString(rawBytes, 0, 4).ShouldBe("B>A>"); |
||||
|
|
||||
|
source.CanRead.ShouldBeTrue(); // The caller keeps the ownership of the original stream
|
||||
|
|
||||
|
(await container.GetAllBytesAsync("markers-blob")).ShouldBe(content); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Run_Contributors_On_The_Plain_Content_When_Encryption_Is_Enabled() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-encrypted"); |
||||
|
var content = "encrypted pipeline content".GetBytes(); |
||||
|
|
||||
|
await container.SaveAsync("encrypted-blob", content); |
||||
|
|
||||
|
// The encryption always runs after the contributors, so the stored form is ciphertext
|
||||
|
var rawBytes = _fakeProvider.GetRawBytesOrNull("pipeline-encrypted", "encrypted-blob"); |
||||
|
rawBytes.ShouldNotBeNull(); |
||||
|
Encoding.ASCII.GetString(rawBytes, 0, 4).ShouldBe("ABPE"); |
||||
|
|
||||
|
(await container.GetAllBytesAsync("encrypted-blob")).ShouldBe(content); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Keep_The_Contributor_Scope_Alive_Until_The_Returned_Stream_Is_Disposed() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-scoped"); |
||||
|
var content = "scoped pipeline content".GetBytes(); |
||||
|
|
||||
|
await container.SaveAsync("scoped-blob", content); |
||||
|
|
||||
|
var stream = await container.GetAsync("scoped-blob"); |
||||
|
|
||||
|
// The scoped service is used lazily here, after GetAsync already returned
|
||||
|
using var result = new MemoryStream(); |
||||
|
await stream.CopyToAsync(result); |
||||
|
result.ToArray().ShouldBe(content); |
||||
|
|
||||
|
var disposedCountBefore = FakeScopedMarkerService.DisposedCount; |
||||
|
stream.Dispose(); |
||||
|
FakeScopedMarkerService.DisposedCount.ShouldBe(disposedCountBefore + 1); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Dispose_The_Provider_Stream_When_Reading_The_Configuration_Fails() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("get-bad-encryption-config"); |
||||
|
|
||||
|
// Save through a plain (well-configured) container to the same provider key,
|
||||
|
// then read through the mis-configured one so the get fails after the provider
|
||||
|
// stream was obtained
|
||||
|
_fakeProvider.SetRawBytes("get-bad-encryption-config", "config-fail-blob", "content".GetBytes()); |
||||
|
|
||||
|
await Assert.ThrowsAnyAsync<Exception>(async () => |
||||
|
{ |
||||
|
await container.GetAsync("config-fail-blob"); |
||||
|
}); |
||||
|
|
||||
|
_fakeProvider.LastServedStream.ShouldNotBeNull(); |
||||
|
_fakeProvider.LastServedStream!.Disposed.ShouldBeTrue(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Dispose_The_Provider_Stream_When_A_Get_Contributor_Fails() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-failing-get"); |
||||
|
|
||||
|
await container.SaveAsync("failing-blob", "failing content".GetBytes()); |
||||
|
|
||||
|
await Assert.ThrowsAsync<InvalidOperationException>(async () => |
||||
|
{ |
||||
|
await container.GetAsync("failing-blob"); |
||||
|
}); |
||||
|
|
||||
|
_fakeProvider.LastServedStream.ShouldNotBeNull(); |
||||
|
_fakeProvider.LastServedStream!.Disposed.ShouldBeTrue(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Dispose_The_Stream_Of_A_Contributor_That_Fails_After_Replacing_It() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-set-throw-save"); |
||||
|
|
||||
|
await Assert.ThrowsAsync<InvalidOperationException>(async () => |
||||
|
{ |
||||
|
await container.SaveAsync("set-throw-blob", "content".GetBytes()); |
||||
|
}); |
||||
|
|
||||
|
FakeSetThenThrowPipelineContributor.LastCreatedStream.ShouldNotBeNull(); |
||||
|
FakeSetThenThrowPipelineContributor.LastCreatedStream!.Disposed.ShouldBeTrue(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Dispose_The_Whole_Chain_When_A_Later_Get_Contributor_Fails() |
||||
|
{ |
||||
|
// The first contributor already wrapped the provider stream when the second one fails
|
||||
|
var container = _blobContainerFactory.Create("pipeline-partial-get"); |
||||
|
|
||||
|
await container.SaveAsync("partial-get-blob", "partial content".GetBytes()); |
||||
|
|
||||
|
await Assert.ThrowsAsync<InvalidOperationException>(async () => |
||||
|
{ |
||||
|
await container.GetAsync("partial-get-blob"); |
||||
|
}); |
||||
|
|
||||
|
_fakeProvider.LastServedStream.ShouldNotBeNull(); |
||||
|
_fakeProvider.LastServedStream!.Disposed.ShouldBeTrue(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Release_The_Remaining_Streams_And_The_Scope_When_A_Dispose_Fails() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-dispose-throw"); |
||||
|
var disposedCountBefore = FakeScopedMarkerService.DisposedCount; |
||||
|
|
||||
|
// The save itself succeeds; the injected failure surfaces from the cleanup
|
||||
|
var exception = await Assert.ThrowsAsync<IOException>(async () => |
||||
|
{ |
||||
|
await container.SaveAsync("dispose-throw-blob", "content".GetBytes()); |
||||
|
}); |
||||
|
|
||||
|
exception.Message.ShouldContain("Injected dispose failure"); |
||||
|
FakeScopedMarkerService.DisposedCount.ShouldBe(disposedCountBefore + 1); // The scope was still released
|
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Keep_The_Container_Tenant_Context_While_The_Returned_Stream_Is_Read() |
||||
|
{ |
||||
|
// A tenant reads a shared (IsMultiTenant = false) container: the lazy
|
||||
|
// transformation must still run in the host context of the container
|
||||
|
var container = _blobContainerFactory.Create("pipeline-shared-tenant"); |
||||
|
var currentTenant = GetRequiredService<Volo.Abp.MultiTenancy.ICurrentTenant>(); |
||||
|
var content = "shared tenant content".GetBytes(); |
||||
|
|
||||
|
using (currentTenant.Change(Guid.NewGuid())) |
||||
|
{ |
||||
|
await container.SaveAsync("shared-tenant-blob", content); |
||||
|
|
||||
|
using var stream = await container.GetAsync("shared-tenant-blob"); |
||||
|
using var result = new MemoryStream(); |
||||
|
await stream.CopyToAsync(result); // The wrapper asserts the tenant context here
|
||||
|
|
||||
|
result.ToArray().ShouldBe(content); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Not_Degrade_A_Modern_Async_Only_Wrapper_Stream() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-modern-async"); |
||||
|
var content = "modern async content".GetBytes(); |
||||
|
|
||||
|
await container.SaveAsync("modern-async-blob", content); |
||||
|
|
||||
|
using var stream = await container.GetAsync("modern-async-blob"); |
||||
|
using var result = new MemoryStream(); |
||||
|
await stream.CopyToAsync(result); // Uses ReadAsync(Memory<byte>) on modern runtimes
|
||||
|
|
||||
|
result.ToArray().ShouldBe(content); |
||||
|
|
||||
|
// Callers of the old overload must get the same bridging
|
||||
|
using var oldOverloadStream = await container.GetAsync("modern-async-blob"); |
||||
|
var buffer = new byte[content.Length]; |
||||
|
var totalReadCount = 0; |
||||
|
while (totalReadCount < buffer.Length) |
||||
|
{ |
||||
|
var readCount = await oldOverloadStream.ReadAsync(buffer, totalReadCount, buffer.Length - totalReadCount, default); |
||||
|
if (readCount == 0) |
||||
|
{ |
||||
|
break; |
||||
|
} |
||||
|
|
||||
|
totalReadCount += readCount; |
||||
|
} |
||||
|
|
||||
|
buffer.ShouldBe(content); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Dispose_The_Contributor_Scope_In_The_Container_Tenant_Context() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-shared-tenant"); |
||||
|
var currentTenant = GetRequiredService<Volo.Abp.MultiTenancy.ICurrentTenant>(); |
||||
|
var content = "scope dispose tenant content".GetBytes(); |
||||
|
|
||||
|
using (currentTenant.Change(Guid.NewGuid())) |
||||
|
{ |
||||
|
await container.SaveAsync("scope-dispose-tenant-blob", content, overrideExisting: true); |
||||
|
|
||||
|
var stream = await container.GetAsync("scope-dispose-tenant-blob"); |
||||
|
using (var result = new MemoryStream()) |
||||
|
{ |
||||
|
await stream.CopyToAsync(result); |
||||
|
} |
||||
|
|
||||
|
FakeTenantRecordingScopedService.Reset(); |
||||
|
stream.Dispose(); // Synchronous dispose from the tenant context
|
||||
|
|
||||
|
FakeTenantRecordingScopedService.HasRecordedDispose.ShouldBeTrue(); |
||||
|
FakeTenantRecordingScopedService.LastDisposeTenantId.ShouldBeNull(); // The container is shared (host)
|
||||
|
|
||||
|
var asyncStream = await container.GetAsync("scope-dispose-tenant-blob"); |
||||
|
using (var result = new MemoryStream()) |
||||
|
{ |
||||
|
await asyncStream.CopyToAsync(result); |
||||
|
} |
||||
|
|
||||
|
FakeTenantRecordingScopedService.Reset(); |
||||
|
await asyncStream.DisposeAsync(); // Asynchronous dispose from the tenant context
|
||||
|
|
||||
|
FakeTenantRecordingScopedService.HasRecordedDispose.ShouldBeTrue(); |
||||
|
FakeTenantRecordingScopedService.LastDisposeTenantId.ShouldBeNull(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Not_Dispose_The_Original_Stream_When_A_Contributor_Sets_It_Back() |
||||
|
{ |
||||
|
// A wraps the original, then the second contributor sets the original back:
|
||||
|
// the pipeline must not treat the caller-owned stream as its own
|
||||
|
var container = _blobContainerFactory.Create("pipeline-unwrap"); |
||||
|
var content = "unwrap content".GetBytes(); |
||||
|
using var source = new MemoryStream(content); |
||||
|
FakeOriginalRestoringPipelineContributor.RestoreTo = source; |
||||
|
try |
||||
|
{ |
||||
|
await container.SaveAsync("unwrap-blob", source); |
||||
|
} |
||||
|
finally |
||||
|
{ |
||||
|
FakeOriginalRestoringPipelineContributor.RestoreTo = null; |
||||
|
} |
||||
|
|
||||
|
source.CanRead.ShouldBeTrue(); // The caller keeps the ownership of the original stream
|
||||
|
_fakeProvider.GetRawBytesOrNull("pipeline-unwrap", "unwrap-blob").ShouldBe(content); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Release_A_Pipeline_Stream_That_Only_Cleans_Up_In_DisposeAsync() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-async-dispose"); |
||||
|
|
||||
|
await container.SaveAsync("async-dispose-blob", "async dispose content".GetBytes()); |
||||
|
|
||||
|
FakeAsyncDisposePipelineContributor.LastSaveStream.ShouldNotBeNull(); |
||||
|
FakeAsyncDisposePipelineContributor.LastSaveStream!.AsyncDisposed.ShouldBeTrue(); |
||||
|
|
||||
|
// The intermediate stream created within the same contributor call is disposed too
|
||||
|
FakeAsyncDisposePipelineContributor.IntermediateSaveStream.ShouldNotBeNull(); |
||||
|
FakeAsyncDisposePipelineContributor.IntermediateSaveStream!.AsyncDisposed.ShouldBeTrue(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public void Should_Bridge_A_Synchronous_Dispose_To_The_Async_Cleanup_Of_The_Cipher_Stream() |
||||
|
{ |
||||
|
// The decrypting stream owns the provider (cipher) stream; a synchronous
|
||||
|
// Dispose of it must still run the async-only cleanup of that stream
|
||||
|
var cipherStream = new AsyncOnlyDisposeStream(); |
||||
|
var decryptingStream = new ChunkedDecryptingReadStream( |
||||
|
cipherStream, new byte[16], new byte[32], new byte[8], 64 * 1024); |
||||
|
|
||||
|
decryptingStream.Dispose(); |
||||
|
|
||||
|
cipherStream.AsyncDisposed.ShouldBeTrue(); |
||||
|
} |
||||
|
|
||||
|
private sealed class AsyncOnlyDisposeStream : Stream |
||||
|
{ |
||||
|
public bool AsyncDisposed { get; private set; } |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) => 0; |
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
// The cleanup only happens asynchronously; a synchronous Dispose does nothing,
|
||||
|
// so the test fails if the decrypting stream does not bridge to DisposeAsync
|
||||
|
public override ValueTask DisposeAsync() |
||||
|
{ |
||||
|
AsyncDisposed = true; |
||||
|
return default; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Bridge_A_Synchronous_Dispose_To_The_Async_Cleanup_Of_A_Get_Wrapper() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-async-dispose"); |
||||
|
var content = "sync dispose bridge content".GetBytes(); |
||||
|
|
||||
|
await container.SaveAsync("sync-bridge-blob", content, overrideExisting: true); |
||||
|
|
||||
|
using (var stream = await container.GetAsync("sync-bridge-blob")) |
||||
|
{ |
||||
|
using var result = new MemoryStream(); |
||||
|
await stream.CopyToAsync(result); |
||||
|
result.ToArray().ShouldBe(content); |
||||
|
} // The synchronous using must still trigger the async-only cleanup
|
||||
|
|
||||
|
FakeAsyncDisposePipelineContributor.LastGetStream.ShouldNotBeNull(); |
||||
|
FakeAsyncDisposePipelineContributor.LastGetStream!.AsyncDisposed.ShouldBeTrue(); |
||||
|
_fakeProvider.LastServedStream.ShouldNotBeNull(); |
||||
|
_fakeProvider.LastServedStream!.Disposed.ShouldBeTrue(); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Forward_The_Timeout_Capability_Of_The_Wrapped_Stream() |
||||
|
{ |
||||
|
var currentTenant = GetRequiredService<Volo.Abp.MultiTenancy.ICurrentTenant>(); |
||||
|
var scope = GetRequiredService<IServiceProvider>().CreateAsyncScope(); |
||||
|
var inner = new TimeoutCapableStream(); |
||||
|
await using var stream = new BlobPipelineScopeStream(inner, scope, currentTenant, null); |
||||
|
|
||||
|
stream.CanTimeout.ShouldBeTrue(); |
||||
|
stream.ReadTimeout = 1234; |
||||
|
stream.ReadTimeout.ShouldBe(1234); |
||||
|
} |
||||
|
|
||||
|
private sealed class TimeoutCapableStream : MemoryStream |
||||
|
{ |
||||
|
private int _readTimeout = -1; |
||||
|
|
||||
|
public override bool CanTimeout => true; |
||||
|
|
||||
|
public override int ReadTimeout |
||||
|
{ |
||||
|
get => _readTimeout; |
||||
|
set => _readTimeout = value; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Not_Fault_The_Composed_Stream_When_The_End_Source_Cancels_Without_Faulting() |
||||
|
{ |
||||
|
// The decrypting stream lets a cancellation before any I/O through while staying
|
||||
|
// healthy (so a retry can still verify). The composed outer stream must mirror that:
|
||||
|
// it must not permanently fault on such a cancellation, or the retry can never verify
|
||||
|
var currentTenant = GetRequiredService<Volo.Abp.MultiTenancy.ICurrentTenant>(); |
||||
|
var scope = GetRequiredService<IServiceProvider>().CreateAsyncScope(); |
||||
|
var endSource = new CancelOnceHealthyAuthenticatedEndStream(); |
||||
|
await using var stream = new BlobPipelineScopeStream(endSource, scope, currentTenant, null, endSource); |
||||
|
|
||||
|
var buffer = new byte[16]; |
||||
|
|
||||
|
// The first read reaches EOF and runs the end check, which cancels while staying healthy
|
||||
|
await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => |
||||
|
{ |
||||
|
await stream.ReadAsync(buffer, 0, buffer.Length); |
||||
|
}); |
||||
|
|
||||
|
// A retry must still be able to run (and pass) the end check, not hit a faulted stream
|
||||
|
var read = await stream.ReadAsync(buffer, 0, buffer.Length); |
||||
|
read.ShouldBe(0); |
||||
|
endSource.EndCheckAttempts.ShouldBe(2); |
||||
|
} |
||||
|
|
||||
|
private sealed class CancelOnceHealthyAuthenticatedEndStream : Stream, IBlobAuthenticatedEndStream |
||||
|
{ |
||||
|
public int EndCheckAttempts { get; private set; } |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) => 0; |
||||
|
|
||||
|
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) |
||||
|
=> Task.FromResult(0); |
||||
|
|
||||
|
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) |
||||
|
=> new ValueTask<int>(0); |
||||
|
|
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
public void EnsureReadToAuthenticatedEnd() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public ValueTask EnsureReadToAuthenticatedEndAsync(CancellationToken cancellationToken = default) |
||||
|
{ |
||||
|
EndCheckAttempts++; |
||||
|
if (EndCheckAttempts == 1) |
||||
|
{ |
||||
|
// Cancelled before any I/O: the source stays healthy, exactly like the real
|
||||
|
// decrypting stream when the token trips just after the outer's own check
|
||||
|
throw new OperationCanceledException(); |
||||
|
} |
||||
|
|
||||
|
return default; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Fault_The_Composed_Stream_When_An_Inner_Read_Fails_After_Consuming() |
||||
|
{ |
||||
|
// An inner contributor consumes a byte from the stream below it and then fails: the
|
||||
|
// composed stream must fault so a read-retry layer can not silently continue from the
|
||||
|
// consumed position and hand the caller content that is missing that byte
|
||||
|
var currentTenant = GetRequiredService<Volo.Abp.MultiTenancy.ICurrentTenant>(); |
||||
|
var scope = GetRequiredService<IServiceProvider>().CreateAsyncScope(); |
||||
|
var underlying = new MemoryStream(new byte[] { 1, 2, 3, 4 }); |
||||
|
var inner = new ConsumeThenThrowStream(underlying); |
||||
|
await using var stream = new BlobPipelineScopeStream(inner, scope, currentTenant, null); |
||||
|
|
||||
|
var buffer = new byte[16]; |
||||
|
await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => |
||||
|
{ |
||||
|
await stream.ReadAsync(buffer, 0, buffer.Length); |
||||
|
}); |
||||
|
|
||||
|
// The failed read must have faulted the stream permanently
|
||||
|
var retry = await Assert.ThrowsAsync<AbpException>(async () => |
||||
|
{ |
||||
|
await stream.ReadAsync(buffer, 0, buffer.Length); |
||||
|
}); |
||||
|
retry.Message.ShouldContain("a previous read operation has failed"); |
||||
|
} |
||||
|
|
||||
|
private sealed class ConsumeThenThrowStream : Stream |
||||
|
{ |
||||
|
private readonly Stream _inner; |
||||
|
private bool _thrown; |
||||
|
|
||||
|
public ConsumeThenThrowStream(Stream inner) |
||||
|
{ |
||||
|
_inner = inner; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) |
||||
|
{ |
||||
|
if (!_thrown) |
||||
|
{ |
||||
|
_thrown = true; |
||||
|
// Consume one byte from the stream below, then fail without handing it up
|
||||
|
_ = _inner.Read(new byte[1], 0, 1); |
||||
|
throw new OperationCanceledException(); |
||||
|
} |
||||
|
|
||||
|
return _inner.ReadAsync(buffer, cancellationToken); |
||||
|
} |
||||
|
|
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
protected override void Dispose(bool disposing) |
||||
|
{ |
||||
|
if (disposing) |
||||
|
{ |
||||
|
_inner.Dispose(); |
||||
|
} |
||||
|
|
||||
|
base.Dispose(disposing); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Verify_The_Authenticated_End_Through_A_Contributor_That_Stops_Early() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-encrypted-earlystop"); |
||||
|
var content = new byte[100_000]; |
||||
|
new Random(42).NextBytes(content); |
||||
|
|
||||
|
await container.SaveAsync("earlystop-blob", content, overrideExisting: true); |
||||
|
|
||||
|
// The normal round trip works: reading to the end verifies the terminal record
|
||||
|
(await container.GetAllBytesAsync("earlystop-blob")).ShouldBe(content); |
||||
|
|
||||
|
// Strip the 20-byte terminal record from the stored ciphertext
|
||||
|
var raw = _fakeProvider.GetRawBytesOrNull("pipeline-encrypted-earlystop", "earlystop-blob"); |
||||
|
raw.ShouldNotBeNull(); |
||||
|
var truncated = new byte[raw!.Length - 20]; |
||||
|
Array.Copy(raw, truncated, truncated.Length); |
||||
|
_fakeProvider.SetRawBytes("pipeline-encrypted-earlystop", "earlystop-blob", truncated); |
||||
|
|
||||
|
// Even though the contributor stops at its own declared length, reading the
|
||||
|
// composed stream to EOF must now fail because the terminal record is gone
|
||||
|
using var stream = await container.GetAsync("earlystop-blob"); |
||||
|
var buffer = new byte[content.Length + 1024]; |
||||
|
|
||||
|
await Assert.ThrowsAsync<AbpException>(async () => |
||||
|
{ |
||||
|
await ReadAllAsync(stream, buffer); |
||||
|
}); |
||||
|
|
||||
|
// The failure must be permanent: reading again must not swallow it and return
|
||||
|
// a normal EOF (which a read-retry layer would treat as a complete read)
|
||||
|
await Assert.ThrowsAsync<AbpException>(async () => |
||||
|
{ |
||||
|
await stream.ReadAsync(buffer, 0, buffer.Length); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private static async Task ReadAllAsync(Stream stream, byte[] buffer) |
||||
|
{ |
||||
|
int read; |
||||
|
var offset = 0; |
||||
|
while ((read = await stream.ReadAsync(buffer, offset, buffer.Length - offset)) > 0) |
||||
|
{ |
||||
|
offset += read; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public void Should_Not_Forward_Reads_After_The_Scope_Stream_Is_Disposed() |
||||
|
{ |
||||
|
// After dispose the contributor scope is gone; reads must not reach the inner
|
||||
|
// stream (a use-after-scope), and CanRead must be consistent with that
|
||||
|
var currentTenant = GetRequiredService<Volo.Abp.MultiTenancy.ICurrentTenant>(); |
||||
|
var scope = GetRequiredService<IServiceProvider>().CreateAsyncScope(); |
||||
|
var inner = new MemoryStream(new byte[10]); |
||||
|
var stream = new BlobPipelineScopeStream(inner, scope, currentTenant, null); |
||||
|
|
||||
|
stream.Dispose(); |
||||
|
|
||||
|
stream.CanRead.ShouldBeFalse(); |
||||
|
Should.Throw<ObjectDisposedException>(() => stream.Read(new byte[1], 0, 1)); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Support_A_Synchronous_Dispose_With_An_Async_Only_Scoped_Service() |
||||
|
{ |
||||
|
var container = _blobContainerFactory.Create("pipeline-async-scoped"); |
||||
|
var content = "async scoped content".GetBytes(); |
||||
|
|
||||
|
await container.SaveAsync("async-scoped-blob", content); |
||||
|
|
||||
|
var stream = await container.GetAsync("async-scoped-blob"); |
||||
|
using var result = new MemoryStream(); |
||||
|
await stream.CopyToAsync(result); |
||||
|
result.ToArray().ShouldBe(content); |
||||
|
|
||||
|
var asyncDisposedCountBefore = FakeAsyncOnlyDisposableService.AsyncDisposedCount; |
||||
|
stream.Dispose(); // Must not throw although the scoped service is async-only disposable
|
||||
|
FakeAsyncOnlyDisposableService.AsyncDisposedCount.ShouldBe(asyncDisposedCountBefore + 1); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,11 @@ |
|||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
public class FakeAPipelineContributor : FakeMarkerPipelineContributorBase, ITransientDependency |
||||
|
{ |
||||
|
public FakeAPipelineContributor() |
||||
|
: base("A>") |
||||
|
{ |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,89 @@ |
|||||
|
#nullable enable |
||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Wraps the content with a stream whose cleanup only happens in DisposeAsync,
|
||||
|
/// so tests can verify the pipeline releases its streams asynchronously — on the
|
||||
|
/// save side and also when the caller disposes the returned stream synchronously.
|
||||
|
/// </summary>
|
||||
|
public class FakeAsyncDisposePipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
public static AsyncDisposeOnlyStream? LastSaveStream { get; private set; } |
||||
|
|
||||
|
public static AsyncDisposeOnlyStream? LastGetStream { get; private set; } |
||||
|
|
||||
|
public static AsyncDisposeOnlyStream? IntermediateSaveStream { get; private set; } |
||||
|
|
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
// Two replacements in one call: the intermediate stream must also be collected
|
||||
|
context.BlobStream = IntermediateSaveStream = new AsyncDisposeOnlyStream(context.BlobStream, ownsInner: false); |
||||
|
context.BlobStream = LastSaveStream = new AsyncDisposeOnlyStream(context.BlobStream, ownsInner: false); |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
// The get-side contract: the wrapper owns the received stream
|
||||
|
context.BlobStream = LastGetStream = new AsyncDisposeOnlyStream(context.BlobStream, ownsInner: true); |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public sealed class AsyncDisposeOnlyStream : Stream |
||||
|
{ |
||||
|
private readonly Stream _inner; |
||||
|
private readonly bool _ownsInner; |
||||
|
|
||||
|
public bool AsyncDisposed { get; private set; } |
||||
|
|
||||
|
public AsyncDisposeOnlyStream(Stream inner, bool ownsInner) |
||||
|
{ |
||||
|
_inner = inner; |
||||
|
_ownsInner = ownsInner; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) => _inner.Read(buffer, offset, count); |
||||
|
|
||||
|
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) |
||||
|
{ |
||||
|
return _inner.ReadAsync(buffer, cancellationToken); |
||||
|
} |
||||
|
|
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
// The cleanup only happens asynchronously; the synchronous Dispose is a no-op
|
||||
|
public override async ValueTask DisposeAsync() |
||||
|
{ |
||||
|
AsyncDisposed = true; |
||||
|
if (_ownsInner) |
||||
|
{ |
||||
|
await _inner.DisposeAsync(); |
||||
|
} |
||||
|
|
||||
|
await base.DisposeAsync(); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,23 @@ |
|||||
|
using System; |
||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// A scoped service that only implements <see cref="IAsyncDisposable"/>: a synchronous
|
||||
|
/// dispose of the owning scope would throw for such a service.
|
||||
|
/// </summary>
|
||||
|
public class FakeAsyncOnlyDisposableService : IScopedDependency, IAsyncDisposable |
||||
|
{ |
||||
|
private static int _asyncDisposedCount; |
||||
|
|
||||
|
public static int AsyncDisposedCount => Volatile.Read(ref _asyncDisposedCount); |
||||
|
|
||||
|
public ValueTask DisposeAsync() |
||||
|
{ |
||||
|
Interlocked.Increment(ref _asyncDisposedCount); |
||||
|
return default; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,28 @@ |
|||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Holds an async-only disposable scoped service without transforming the content.
|
||||
|
/// </summary>
|
||||
|
public class FakeAsyncScopedPipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
// ReSharper disable once NotAccessedField.Local
|
||||
|
private readonly FakeAsyncOnlyDisposableService _service; |
||||
|
|
||||
|
public FakeAsyncScopedPipelineContributor(FakeAsyncOnlyDisposableService service) |
||||
|
{ |
||||
|
_service = service; |
||||
|
} |
||||
|
|
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,11 @@ |
|||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
public class FakeBPipelineContributor : FakeMarkerPipelineContributorBase, ITransientDependency |
||||
|
{ |
||||
|
public FakeBPipelineContributor() |
||||
|
: base("B>") |
||||
|
{ |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,59 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Wraps the stream with a pass-through wrapper that fails on Dispose, so tests
|
||||
|
/// can verify the best-effort cleanup of the pipeline.
|
||||
|
/// </summary>
|
||||
|
public class FakeDisposeThrowingPipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
context.BlobStream = new DisposeThrowingStream(context.BlobStream); |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
private sealed class DisposeThrowingStream : Stream |
||||
|
{ |
||||
|
private readonly Stream _inner; |
||||
|
|
||||
|
public DisposeThrowingStream(Stream inner) |
||||
|
{ |
||||
|
_inner = inner; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) => _inner.Read(buffer, offset, count); |
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
protected override void Dispose(bool disposing) |
||||
|
{ |
||||
|
throw new IOException("Injected dispose failure!"); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,120 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// On saving, prepends the 4-byte content length; on getting, returns a wrapper that
|
||||
|
/// stops at that declared length — reproducing a contributor that reaches its own EOF
|
||||
|
/// before the decryption stream's authenticated end.
|
||||
|
/// </summary>
|
||||
|
public class FakeEarlyStopPipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
public async Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
using var content = new MemoryStream(); |
||||
|
await context.BlobStream.CopyToAsync(content, 81920, context.CancellationToken); |
||||
|
|
||||
|
var output = new MemoryStream(); |
||||
|
var lengthPrefix = BitConverter.GetBytes((int)content.Length); |
||||
|
output.Write(lengthPrefix, 0, lengthPrefix.Length); |
||||
|
content.Position = 0; |
||||
|
await content.CopyToAsync(output, 81920, context.CancellationToken); |
||||
|
output.Position = 0; |
||||
|
context.BlobStream = output; |
||||
|
} |
||||
|
|
||||
|
public async Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
var lengthPrefix = new byte[4]; |
||||
|
await ReadExactlyAsync(context.BlobStream, lengthPrefix, context.CancellationToken); |
||||
|
var contentLength = BitConverter.ToInt32(lengthPrefix, 0); |
||||
|
context.BlobStream = new LengthLimitedStream(context.BlobStream, contentLength); |
||||
|
} |
||||
|
|
||||
|
private static async Task ReadExactlyAsync(Stream stream, byte[] buffer, CancellationToken cancellationToken) |
||||
|
{ |
||||
|
var total = 0; |
||||
|
while (total < buffer.Length) |
||||
|
{ |
||||
|
var read = await stream.ReadAsync(buffer.AsMemory(total, buffer.Length - total), cancellationToken); |
||||
|
if (read == 0) |
||||
|
{ |
||||
|
throw new AbpException("Unexpected end of stream while reading the length prefix!"); |
||||
|
} |
||||
|
|
||||
|
total += read; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private sealed class LengthLimitedStream : Stream |
||||
|
{ |
||||
|
private readonly Stream _inner; |
||||
|
private long _remaining; |
||||
|
|
||||
|
public LengthLimitedStream(Stream inner, long length) |
||||
|
{ |
||||
|
_inner = inner; |
||||
|
_remaining = length; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
// Stops at the declared length without reading the rest of the inner stream
|
||||
|
public override int Read(byte[] buffer, int offset, int count) |
||||
|
{ |
||||
|
if (_remaining <= 0) |
||||
|
{ |
||||
|
return 0; |
||||
|
} |
||||
|
|
||||
|
var toRead = (int)Math.Min(count, _remaining); |
||||
|
var read = _inner.Read(buffer, offset, toRead); |
||||
|
_remaining -= read; |
||||
|
return read; |
||||
|
} |
||||
|
|
||||
|
public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) |
||||
|
{ |
||||
|
if (_remaining <= 0) |
||||
|
{ |
||||
|
return 0; |
||||
|
} |
||||
|
|
||||
|
var toRead = (int)Math.Min(buffer.Length, _remaining); |
||||
|
var read = await _inner.ReadAsync(buffer.Slice(0, toRead), cancellationToken); |
||||
|
_remaining -= read; |
||||
|
return read; |
||||
|
} |
||||
|
|
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
protected override void Dispose(bool disposing) |
||||
|
{ |
||||
|
if (disposing) |
||||
|
{ |
||||
|
_inner.Dispose(); |
||||
|
} |
||||
|
|
||||
|
base.Dispose(disposing); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,18 @@ |
|||||
|
using System; |
||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
public class FakeFailingGetPipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
throw new InvalidOperationException("This contributor can not read content back!"); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,48 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// A readable, non-seekable stream whose length/position probes fail with an
|
||||
|
/// <see cref="IOException"/> — a legal stream shape the optional probes must tolerate.
|
||||
|
/// </summary>
|
||||
|
public sealed class FakeIoFailingLengthStream : Stream |
||||
|
{ |
||||
|
private readonly Stream _inner; |
||||
|
|
||||
|
public FakeIoFailingLengthStream(Stream inner) |
||||
|
{ |
||||
|
_inner = inner; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new IOException("The length is not available!"); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new IOException("The position is not available!"); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) => _inner.Read(buffer, offset, count); |
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
protected override void Dispose(bool disposing) |
||||
|
{ |
||||
|
if (disposing) |
||||
|
{ |
||||
|
_inner.Dispose(); |
||||
|
} |
||||
|
|
||||
|
base.Dispose(disposing); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,156 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Text; |
||||
|
using System.Threading.Tasks; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Prepends a marker while saving and verifies/strips it while getting,
|
||||
|
/// so tests can observe the raw stored form and the execution order.
|
||||
|
/// </summary>
|
||||
|
public abstract class FakeMarkerPipelineContributorBase : IBlobPipelineContributor |
||||
|
{ |
||||
|
private readonly byte[] _marker; |
||||
|
|
||||
|
protected FakeMarkerPipelineContributorBase(string marker) |
||||
|
{ |
||||
|
_marker = Encoding.UTF8.GetBytes(marker); |
||||
|
} |
||||
|
|
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
context.BlobStream = new MarkerPrependingStream(_marker, context.BlobStream); |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
context.BlobStream = new MarkerStrippingStream(_marker, context.BlobStream); |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
private sealed class MarkerPrependingStream : Stream |
||||
|
{ |
||||
|
private readonly byte[] _marker; |
||||
|
private readonly Stream _inner; |
||||
|
private int _markerPosition; |
||||
|
|
||||
|
public MarkerPrependingStream(byte[] marker, Stream inner) |
||||
|
{ |
||||
|
_marker = marker; |
||||
|
_inner = inner; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) |
||||
|
{ |
||||
|
if (_markerPosition < _marker.Length) |
||||
|
{ |
||||
|
var toCopy = Math.Min(count, _marker.Length - _markerPosition); |
||||
|
Array.Copy(_marker, _markerPosition, buffer, offset, toCopy); |
||||
|
_markerPosition += toCopy; |
||||
|
return toCopy; |
||||
|
} |
||||
|
|
||||
|
return _inner.Read(buffer, offset, count); |
||||
|
} |
||||
|
|
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
// The save-side contract: the wrapper leaves the received stream open
|
||||
|
} |
||||
|
|
||||
|
private sealed class MarkerStrippingStream : Stream |
||||
|
{ |
||||
|
private readonly byte[] _marker; |
||||
|
private readonly Stream _inner; |
||||
|
private bool _markerConsumed; |
||||
|
|
||||
|
public MarkerStrippingStream(byte[] marker, Stream inner) |
||||
|
{ |
||||
|
_marker = marker; |
||||
|
_inner = inner; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) |
||||
|
{ |
||||
|
ConsumeMarker(); |
||||
|
return _inner.Read(buffer, offset, count); |
||||
|
} |
||||
|
|
||||
|
private void ConsumeMarker() |
||||
|
{ |
||||
|
if (_markerConsumed) |
||||
|
{ |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
_markerConsumed = true; |
||||
|
|
||||
|
var markerBytes = new byte[_marker.Length]; |
||||
|
var readCount = 0; |
||||
|
while (readCount < markerBytes.Length) |
||||
|
{ |
||||
|
var read = _inner.Read(markerBytes, readCount, markerBytes.Length - readCount); |
||||
|
if (read <= 0) |
||||
|
{ |
||||
|
break; |
||||
|
} |
||||
|
|
||||
|
readCount += read; |
||||
|
} |
||||
|
|
||||
|
if (readCount != _marker.Length || !((ReadOnlySpan<byte>)markerBytes).SequenceEqual(_marker)) |
||||
|
{ |
||||
|
throw new AbpException($"The expected content marker '{Encoding.UTF8.GetString(_marker)}' was not found!"); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
protected override void Dispose(bool disposing) |
||||
|
{ |
||||
|
// The get-side contract: the wrapper owns the received stream
|
||||
|
if (disposing) |
||||
|
{ |
||||
|
_inner.Dispose(); |
||||
|
} |
||||
|
|
||||
|
base.Dispose(disposing); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,59 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Threading; |
||||
|
using System.Threading.Tasks; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// A stream that only supports the modern async read overload; the synchronous
|
||||
|
/// read throws, like some modern network/response streams.
|
||||
|
/// </summary>
|
||||
|
public sealed class FakeModernAsyncOnlyStream : Stream |
||||
|
{ |
||||
|
private readonly Stream _inner; |
||||
|
|
||||
|
public FakeModernAsyncOnlyStream(Stream inner) |
||||
|
{ |
||||
|
_inner = inner; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) |
||||
|
{ |
||||
|
throw new InvalidOperationException("This stream only supports the modern async read overload!"); |
||||
|
} |
||||
|
|
||||
|
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) |
||||
|
{ |
||||
|
return _inner.ReadAsync(buffer, cancellationToken); |
||||
|
} |
||||
|
|
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
protected override void Dispose(bool disposing) |
||||
|
{ |
||||
|
if (disposing) |
||||
|
{ |
||||
|
_inner.Dispose(); |
||||
|
} |
||||
|
|
||||
|
base.Dispose(disposing); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,23 @@ |
|||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Wraps the content with a stream that only supports the modern async read
|
||||
|
/// overload, so tests can verify the pipeline does not degrade such a stream
|
||||
|
/// to the synchronous byte[] fallback of the Stream base class.
|
||||
|
/// </summary>
|
||||
|
public class FakeModernAsyncPipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
context.BlobStream = new FakeModernAsyncOnlyStream(context.BlobStream); |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,30 @@ |
|||||
|
#nullable enable |
||||
|
using System.IO; |
||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Sets the content stream back to a stream chosen by the test (the caller's
|
||||
|
/// original), so tests can verify the pipeline never treats it as its own.
|
||||
|
/// </summary>
|
||||
|
public class FakeOriginalRestoringPipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
public static Stream? RestoreTo { get; set; } |
||||
|
|
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
if (RestoreTo != null) |
||||
|
{ |
||||
|
context.BlobStream = RestoreTo; |
||||
|
} |
||||
|
|
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,37 @@ |
|||||
|
using System; |
||||
|
using System.Threading; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// A scoped service used by <see cref="FakeScopedXorPipelineContributor"/> to prove
|
||||
|
/// that the contributor scope stays alive while the returned stream is being read.
|
||||
|
/// </summary>
|
||||
|
public class FakeScopedMarkerService : IScopedDependency, IDisposable |
||||
|
{ |
||||
|
private static int _disposedCount; |
||||
|
|
||||
|
public static int DisposedCount => Volatile.Read(ref _disposedCount); |
||||
|
|
||||
|
private bool _disposed; |
||||
|
|
||||
|
public byte Transform(byte value) |
||||
|
{ |
||||
|
if (_disposed) |
||||
|
{ |
||||
|
throw new ObjectDisposedException(nameof(FakeScopedMarkerService)); |
||||
|
} |
||||
|
|
||||
|
return (byte)(value ^ 0x5A); |
||||
|
} |
||||
|
|
||||
|
public void Dispose() |
||||
|
{ |
||||
|
if (!_disposed) |
||||
|
{ |
||||
|
_disposed = true; |
||||
|
Interlocked.Increment(ref _disposedCount); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,86 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// XOR-transforms the content through a scoped service that is used lazily,
|
||||
|
/// while the stream is being read.
|
||||
|
/// </summary>
|
||||
|
public class FakeScopedXorPipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
private readonly FakeScopedMarkerService _markerService; |
||||
|
|
||||
|
public FakeScopedXorPipelineContributor(FakeScopedMarkerService markerService) |
||||
|
{ |
||||
|
_markerService = markerService; |
||||
|
} |
||||
|
|
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
context.BlobStream = new XorStream(_markerService, context.BlobStream, ownsInner: false); |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
context.BlobStream = new XorStream(_markerService, context.BlobStream, ownsInner: true); |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
private sealed class XorStream : Stream |
||||
|
{ |
||||
|
private readonly FakeScopedMarkerService _markerService; |
||||
|
private readonly Stream _inner; |
||||
|
private readonly bool _ownsInner; |
||||
|
|
||||
|
public XorStream(FakeScopedMarkerService markerService, Stream inner, bool ownsInner) |
||||
|
{ |
||||
|
_markerService = markerService; |
||||
|
_inner = inner; |
||||
|
_ownsInner = ownsInner; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) |
||||
|
{ |
||||
|
var readCount = _inner.Read(buffer, offset, count); |
||||
|
for (var i = 0; i < readCount; i++) |
||||
|
{ |
||||
|
buffer[offset + i] = _markerService.Transform(buffer[offset + i]); |
||||
|
} |
||||
|
|
||||
|
return readCount; |
||||
|
} |
||||
|
|
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
protected override void Dispose(bool disposing) |
||||
|
{ |
||||
|
if (disposing && _ownsInner) |
||||
|
{ |
||||
|
_inner.Dispose(); |
||||
|
} |
||||
|
|
||||
|
base.Dispose(disposing); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,65 @@ |
|||||
|
#nullable enable |
||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Replaces the stream and then fails, so tests can verify that the
|
||||
|
/// already-created stream is not leaked.
|
||||
|
/// </summary>
|
||||
|
public class FakeSetThenThrowPipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
public static TrackableStream? LastCreatedStream { get; private set; } |
||||
|
|
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
context.BlobStream = LastCreatedStream = new TrackableStream(context.BlobStream); |
||||
|
throw new InvalidOperationException("This contributor fails after replacing the stream!"); |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public sealed class TrackableStream : Stream |
||||
|
{ |
||||
|
private readonly Stream _inner; |
||||
|
|
||||
|
public bool Disposed { get; private set; } |
||||
|
|
||||
|
public TrackableStream(Stream inner) |
||||
|
{ |
||||
|
_inner = inner; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead => true; |
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) => _inner.Read(buffer, offset, count); |
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
protected override void Dispose(bool disposing) |
||||
|
{ |
||||
|
Disposed = true; |
||||
|
base.Dispose(disposing); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,100 @@ |
|||||
|
using System; |
||||
|
using System.IO; |
||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
using Volo.Abp.MultiTenancy; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Asserts, lazily while the content is read, that the ambient tenant is the
|
||||
|
/// tenant the BLOB operation belongs to.
|
||||
|
/// </summary>
|
||||
|
public class FakeTenantAssertingPipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
private readonly ICurrentTenant _currentTenant; |
||||
|
|
||||
|
public FakeTenantAssertingPipelineContributor(ICurrentTenant currentTenant) |
||||
|
{ |
||||
|
_currentTenant = currentTenant; |
||||
|
} |
||||
|
|
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
context.BlobStream = new TenantAssertingStream(context.BlobStream, _currentTenant, context.TenantId); |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
private sealed class TenantAssertingStream : Stream |
||||
|
{ |
||||
|
private readonly Stream _inner; |
||||
|
private readonly ICurrentTenant _currentTenant; |
||||
|
private readonly Guid? _expectedTenantId; |
||||
|
|
||||
|
public TenantAssertingStream(Stream inner, ICurrentTenant currentTenant, Guid? expectedTenantId) |
||||
|
{ |
||||
|
_inner = inner; |
||||
|
_currentTenant = currentTenant; |
||||
|
_expectedTenantId = expectedTenantId; |
||||
|
} |
||||
|
|
||||
|
public override bool CanRead |
||||
|
{ |
||||
|
get |
||||
|
{ |
||||
|
// Stream.CopyToAsync reads CanRead before the first ReadAsync call
|
||||
|
AssertTenant(); |
||||
|
return true; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override bool CanSeek => false; |
||||
|
public override bool CanWrite => false; |
||||
|
public override long Length => throw new NotSupportedException(); |
||||
|
|
||||
|
public override long Position |
||||
|
{ |
||||
|
get => throw new NotSupportedException(); |
||||
|
set => throw new NotSupportedException(); |
||||
|
} |
||||
|
|
||||
|
public override void Flush() |
||||
|
{ |
||||
|
} |
||||
|
|
||||
|
public override int Read(byte[] buffer, int offset, int count) |
||||
|
{ |
||||
|
AssertTenant(); |
||||
|
return _inner.Read(buffer, offset, count); |
||||
|
} |
||||
|
|
||||
|
private void AssertTenant() |
||||
|
{ |
||||
|
if (_currentTenant.Id != _expectedTenantId) |
||||
|
{ |
||||
|
throw new AbpException( |
||||
|
$"The lazy transformation ran in the tenant '{_currentTenant.Id?.ToString() ?? "host"}' " + |
||||
|
$"instead of the tenant of the BLOB operation ('{_expectedTenantId?.ToString() ?? "host"}')!"); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); |
||||
|
public override void SetLength(long value) => throw new NotSupportedException(); |
||||
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); |
||||
|
|
||||
|
protected override void Dispose(bool disposing) |
||||
|
{ |
||||
|
if (disposing) |
||||
|
{ |
||||
|
_inner.Dispose(); |
||||
|
} |
||||
|
|
||||
|
base.Dispose(disposing); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,29 @@ |
|||||
|
using System.Threading.Tasks; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Holds a <see cref="FakeTenantRecordingScopedService"/> in the contributor scope
|
||||
|
/// without transforming the content.
|
||||
|
/// </summary>
|
||||
|
public class FakeTenantRecordingScopedPipelineContributor : IBlobPipelineContributor, ITransientDependency |
||||
|
{ |
||||
|
// ReSharper disable once NotAccessedField.Local
|
||||
|
private readonly FakeTenantRecordingScopedService _service; |
||||
|
|
||||
|
public FakeTenantRecordingScopedPipelineContributor(FakeTenantRecordingScopedService service) |
||||
|
{ |
||||
|
_service = service; |
||||
|
} |
||||
|
|
||||
|
public Task OnSavingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
|
||||
|
public Task OnGettingAsync(BlobPipelineContext context) |
||||
|
{ |
||||
|
return Task.CompletedTask; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,35 @@ |
|||||
|
using System; |
||||
|
using Volo.Abp.DependencyInjection; |
||||
|
using Volo.Abp.MultiTenancy; |
||||
|
|
||||
|
namespace Volo.Abp.BlobStoring.Fakes; |
||||
|
|
||||
|
/// <summary>
|
||||
|
/// Records the ambient tenant at the moment the owning scope disposes it, so tests
|
||||
|
/// can verify the scope is released in the tenant of the BLOB operation.
|
||||
|
/// </summary>
|
||||
|
public class FakeTenantRecordingScopedService : IScopedDependency, IDisposable |
||||
|
{ |
||||
|
public static Guid? LastDisposeTenantId { get; private set; } |
||||
|
|
||||
|
public static bool HasRecordedDispose { get; private set; } |
||||
|
|
||||
|
private readonly ICurrentTenant _currentTenant; |
||||
|
|
||||
|
public FakeTenantRecordingScopedService(ICurrentTenant currentTenant) |
||||
|
{ |
||||
|
_currentTenant = currentTenant; |
||||
|
} |
||||
|
|
||||
|
public static void Reset() |
||||
|
{ |
||||
|
LastDisposeTenantId = null; |
||||
|
HasRecordedDispose = false; |
||||
|
} |
||||
|
|
||||
|
public void Dispose() |
||||
|
{ |
||||
|
LastDisposeTenantId = _currentTenant.Id; |
||||
|
HasRecordedDispose = true; |
||||
|
} |
||||
|
} |
||||
Loading…
Reference in new issue