Compare commits
4 Commits
3.9.0
...
jrm/blobs-poc
| Author | SHA1 | Date | |
|---|---|---|---|
| a560f7f159 | |||
| 94d2a01880 | |||
| b0da4510bf | |||
| 96392d0d2f |
@@ -33,6 +33,9 @@ jobs:
|
||||
|
||||
- name: 🔨 Unit Tests
|
||||
run: dotnet test ${{ env.Solution }} --configuration Release --filter "Category!=Integration" --no-build --no-restore --verbosity=normal /p:AltCover=true /p:AltCoverAttributeFilter=ExcludeFromCodeCoverage
|
||||
|
||||
- name: 🎁 Pack
|
||||
run: dotnet pack ${{ env.Solution }} --configuration Release --no-build
|
||||
|
||||
- name: Upload coverage reports to Codecov with GitHub Action
|
||||
uses: codecov/codecov-action@v5
|
||||
|
||||
@@ -42,6 +42,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "workflows", "workflows", "{
|
||||
ProjectSection(SolutionItems) = preProject
|
||||
.github\workflows\pr.yml = .github\workflows\pr.yml
|
||||
.github\workflows\release.yml = .github\workflows\release.yml
|
||||
.github\workflows\integration-test.yml = .github\workflows\integration-test.yml
|
||||
EndProjectSection
|
||||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Speckle.Sdk.Tests.Performance", "tests\Speckle.Sdk.Tests.Performance\Speckle.Sdk.Tests.Performance.csproj", "{870E3396-E6F7-43AE-B120-E651FA4F46BD}"
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
using System.Threading.Channels;
|
||||
|
||||
namespace Speckle.Sdk.Dependencies;
|
||||
|
||||
internal sealed class BroadcastChannel<T>
|
||||
{
|
||||
private readonly List<Channel<T>> _subscribers = [];
|
||||
|
||||
public ChannelReader<T> Subscribe()
|
||||
{
|
||||
var channel = Channel.CreateUnbounded<T>(new UnboundedChannelOptions() { SingleReader = true });
|
||||
_subscribers.Add(channel);
|
||||
return channel.Reader;
|
||||
}
|
||||
|
||||
public async Task WriteAsync(T item, CancellationToken cancellationToken)
|
||||
{
|
||||
foreach (var sub in _subscribers)
|
||||
{
|
||||
await sub.Writer.WriteAsync(item, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
public bool IsReadingCompleted()
|
||||
{
|
||||
return _subscribers.All(x => x.Reader.Completion.IsCompleted);
|
||||
}
|
||||
|
||||
public void CompleteWriters()
|
||||
{
|
||||
foreach (var sub in _subscribers)
|
||||
{
|
||||
sub.Writer.Complete();
|
||||
}
|
||||
}
|
||||
|
||||
public async Task CompleteReaders()
|
||||
{
|
||||
await Task.WhenAll(_subscribers.Select(x => x.Reader.Completion)).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
@@ -6,28 +6,23 @@ namespace Speckle.Sdk.Serialisation.V2.Send;
|
||||
public sealed class Batch<T> : IMemoryOwner<T>
|
||||
where T : IHasByteSize
|
||||
{
|
||||
private static readonly Pool<List<T>> _pool = Pools.CreateListPool<T>();
|
||||
#pragma warning disable IDE0032
|
||||
private readonly List<T> _items = _pool.Get();
|
||||
private int _batchByteSize;
|
||||
#pragma warning restore IDE0032
|
||||
private static readonly Pool<List<T>> s_pool = Pools.CreateListPool<T>();
|
||||
public List<T> Items { get; } = s_pool.Get();
|
||||
public int BatchByteSize { get; private set; }
|
||||
|
||||
public void Add(T item)
|
||||
{
|
||||
_items.Add(item);
|
||||
_batchByteSize += item.ByteSize;
|
||||
Items.Add(item);
|
||||
BatchByteSize += item.ByteSize;
|
||||
}
|
||||
|
||||
public void TrimExcess()
|
||||
{
|
||||
_items.TrimExcess();
|
||||
_batchByteSize = _items.Sum(x => x.ByteSize);
|
||||
Items.TrimExcess();
|
||||
BatchByteSize = Items.Sum(x => x.ByteSize);
|
||||
}
|
||||
|
||||
public int BatchByteSize => _batchByteSize;
|
||||
public List<T> Items => _items;
|
||||
public void Dispose() => s_pool.Return(Items);
|
||||
|
||||
public void Dispose() => _pool.Return(_items);
|
||||
|
||||
public Memory<T> Memory => new(_items.ToArray());
|
||||
public Memory<T> Memory => new(Items.ToArray());
|
||||
}
|
||||
|
||||
@@ -1,74 +1,134 @@
|
||||
using System.Buffers;
|
||||
using System.Threading.Channels;
|
||||
using Open.ChannelExtensions;
|
||||
using Speckle.Sdk.Serialisation.V2.Send;
|
||||
|
||||
namespace Speckle.Sdk.Dependencies.Serialization;
|
||||
|
||||
public abstract class ChannelSaver<T>
|
||||
where T : IHasByteSize
|
||||
public abstract class ChannelSaver<TItem, TBlobItem>
|
||||
where TItem : IHasByteSize
|
||||
where TBlobItem : IHasByteSize, TItem
|
||||
{
|
||||
private const int SEND_CAPACITY = 10000;
|
||||
private const int HTTP_SEND_CHUNK_SIZE = 25_000_000; //bytes
|
||||
private const int BLOB_SEND_CHUNK_SIZE = 10; //count
|
||||
private static readonly TimeSpan HTTP_BATCH_TIMEOUT = TimeSpan.FromSeconds(2);
|
||||
private const int MAX_PARALLELISM_HTTP = 4;
|
||||
private const int HTTP_CAPACITY = 500;
|
||||
private const int MAX_CACHE_WRITE_PARALLELISM = 1;
|
||||
private const int MAX_CACHE_BATCH = 1000;
|
||||
|
||||
private readonly Channel<T> _checkCacheChannel = Channel.CreateBounded<T>(
|
||||
new BoundedChannelOptions(SEND_CAPACITY)
|
||||
{
|
||||
AllowSynchronousContinuations = true,
|
||||
Capacity = SEND_CAPACITY,
|
||||
SingleWriter = false,
|
||||
SingleReader = false,
|
||||
FullMode = BoundedChannelFullMode.Wait,
|
||||
},
|
||||
_ => throw new NotImplementedException("Dropping items not supported.")
|
||||
);
|
||||
private readonly BroadcastChannel<TItem> _broadcastChannel = new();
|
||||
|
||||
public Task Start(
|
||||
public async Task Start(
|
||||
int? maxParallelism,
|
||||
int? httpBatchSize,
|
||||
int? blobSendCache,
|
||||
int? cacheBatchSize,
|
||||
CancellationToken cancellationToken
|
||||
) =>
|
||||
_checkCacheChannel
|
||||
.Reader.BatchByByteSize(httpBatchSize ?? HTTP_SEND_CHUNK_SIZE)
|
||||
.WithTimeout(HTTP_BATCH_TIMEOUT)
|
||||
.PipeAsync(
|
||||
maxParallelism ?? MAX_PARALLELISM_HTTP,
|
||||
async x => await SendToServer(x).ConfigureAwait(false),
|
||||
HTTP_CAPACITY,
|
||||
false,
|
||||
)
|
||||
{
|
||||
maxParallelism ??= MAX_PARALLELISM_HTTP;
|
||||
httpBatchSize ??= HTTP_SEND_CHUNK_SIZE;
|
||||
blobSendCache ??= BLOB_SEND_CHUNK_SIZE;
|
||||
cacheBatchSize ??= MAX_CACHE_BATCH;
|
||||
await StartInternal(
|
||||
maxParallelism.Value,
|
||||
httpBatchSize.Value,
|
||||
blobSendCache.Value,
|
||||
cacheBatchSize.Value,
|
||||
cancellationToken
|
||||
)
|
||||
.Join()
|
||||
.Batch(cacheBatchSize ?? MAX_CACHE_BATCH, singleReader: true)
|
||||
.WithTimeout(HTTP_BATCH_TIMEOUT)
|
||||
.ReadAllConcurrently(MAX_CACHE_WRITE_PARALLELISM, SaveToCache, cancellationToken)
|
||||
.ContinueWith(
|
||||
t =>
|
||||
{
|
||||
Exception? ex = t.Exception;
|
||||
if (ex is null && t.Status is TaskStatus.Canceled && !cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
ex = new OperationCanceledException();
|
||||
}
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
if (ex is not null)
|
||||
{
|
||||
RecordException(ex);
|
||||
}
|
||||
_checkCacheChannel.Writer.TryComplete(ex);
|
||||
},
|
||||
cancellationToken,
|
||||
TaskContinuationOptions.ExecuteSynchronously,
|
||||
TaskScheduler.Current
|
||||
private Task StartInternal(
|
||||
int maxParallelism,
|
||||
int httpBatchSize,
|
||||
int blobSendCache,
|
||||
int cacheBatchSize,
|
||||
CancellationToken cancellationToken
|
||||
)
|
||||
{
|
||||
Task serverSend = _broadcastChannel
|
||||
.Subscribe()
|
||||
.BatchByByteSize(httpBatchSize)
|
||||
.WithTimeout(HTTP_BATCH_TIMEOUT)
|
||||
.ReadAllConcurrentlyAsync(
|
||||
maxParallelism,
|
||||
async x => await SendToServer(x).ConfigureAwait(false),
|
||||
cancellationToken
|
||||
);
|
||||
|
||||
public async Task SaveAsync(T item, CancellationToken cancellationToken)
|
||||
Task writeCache = _broadcastChannel
|
||||
.Subscribe()
|
||||
.Batch(cacheBatchSize)
|
||||
.ReadAll(SaveToCache, true, cancellationToken: cancellationToken)
|
||||
.AsTask();
|
||||
|
||||
Task blobsCache = _broadcastChannel
|
||||
.Subscribe()
|
||||
.OfType<TItem, TBlobItem>()
|
||||
.BatchByByteSize(blobSendCache)
|
||||
.ReadAllAsync(
|
||||
async x => await SendBlobToServer(x).ConfigureAwait(false),
|
||||
true,
|
||||
cancellationToken: cancellationToken
|
||||
)
|
||||
.AsTask();
|
||||
|
||||
return Task.WhenAll(serverSend, writeCache, blobsCache);
|
||||
|
||||
// return _broadcastChannel
|
||||
// .Subscribe()
|
||||
// .BatchByByteSize(httpBatchSize ?? HTTP_SEND_CHUNK_SIZE)
|
||||
// .WithTimeout(HTTP_BATCH_TIMEOUT)
|
||||
// .PipeAsync(
|
||||
// maxParallelism ?? MAX_PARALLELISM_HTTP,
|
||||
// async x => await SendToServer(x).ConfigureAwait(false),
|
||||
// HTTP_CAPACITY,
|
||||
// false,
|
||||
// cancellationToken
|
||||
// )
|
||||
// .Join()
|
||||
// .Batch(cacheBatchSize ?? MAX_CACHE_BATCH, singleReader: true)
|
||||
// .WithTimeout(HTTP_BATCH_TIMEOUT)
|
||||
// .ReadAllConcurrently(MAX_CACHE_WRITE_PARALLELISM, SaveToCache, cancellationToken)
|
||||
// .ContinueWith(
|
||||
// t =>
|
||||
// {
|
||||
// Exception? ex = t.Exception;
|
||||
// if (ex is null && t.Status is TaskStatus.Canceled && !cancellationToken.IsCancellationRequested)
|
||||
// {
|
||||
// ex = new OperationCanceledException();
|
||||
// }
|
||||
//
|
||||
// if (ex is not null)
|
||||
// {
|
||||
// RecordException(ex);
|
||||
// }
|
||||
//
|
||||
// _checkCacheChannel.Writer.TryComplete(ex);
|
||||
// },
|
||||
// cancellationToken,
|
||||
// TaskContinuationOptions.ExecuteSynchronously,
|
||||
// TaskScheduler.Current
|
||||
// );
|
||||
}
|
||||
|
||||
private async ValueTask SendBlobToServer(IMemoryOwner<TBlobItem> batch)
|
||||
{
|
||||
try
|
||||
{
|
||||
await SendBlobToServerInternal((Batch<TBlobItem>)batch).ConfigureAwait(false);
|
||||
}
|
||||
#pragma warning disable CA1031
|
||||
catch (Exception ex)
|
||||
#pragma warning restore CA1031
|
||||
{
|
||||
RecordException(ex);
|
||||
}
|
||||
}
|
||||
|
||||
protected abstract Task SendBlobToServerInternal(Batch<TBlobItem> batch);
|
||||
|
||||
public async Task SaveAsync(TItem item, CancellationToken cancellationToken)
|
||||
{
|
||||
if (Exception is not null)
|
||||
{
|
||||
@@ -76,36 +136,34 @@ public abstract class ChannelSaver<T>
|
||||
}
|
||||
//can switch to check then try pattern when back pressure is needed or exceptions are too much
|
||||
//the trees don't need to respond to back pressure
|
||||
await _checkCacheChannel.Writer.WriteAsync(item, cancellationToken).ConfigureAwait(false);
|
||||
await _broadcastChannel.WriteAsync(item, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task<IMemoryOwner<T>> SendToServer(IMemoryOwner<T> batch)
|
||||
private async Task SendToServer(IMemoryOwner<TItem> batch)
|
||||
{
|
||||
try
|
||||
{
|
||||
await SendToServerInternal((Batch<T>)batch).ConfigureAwait(false);
|
||||
return batch;
|
||||
await SendToServerInternal((Batch<TItem>)batch).ConfigureAwait(false);
|
||||
}
|
||||
#pragma warning disable CA1031
|
||||
catch (Exception ex)
|
||||
#pragma warning restore CA1031
|
||||
{
|
||||
RecordException(ex);
|
||||
return batch;
|
||||
}
|
||||
}
|
||||
|
||||
protected abstract Task SendToServerInternal(Batch<T> batch);
|
||||
protected abstract Task SendToServerInternal(Batch<TItem> batch);
|
||||
|
||||
public abstract void SaveToCache(List<T> item);
|
||||
public abstract void SaveToCache(List<TItem> item);
|
||||
|
||||
public void DoneTraversing() => _checkCacheChannel.Writer.TryComplete();
|
||||
public void DoneTraversing() => _broadcastChannel.CompleteWriters();
|
||||
|
||||
public async Task DoneSaving()
|
||||
{
|
||||
if (!_checkCacheChannel.Reader.Completion.IsCompleted)
|
||||
if (!_broadcastChannel.IsReadingCompleted())
|
||||
{
|
||||
await _checkCacheChannel.Reader.Completion.ConfigureAwait(false);
|
||||
await _broadcastChannel.CompleteReaders().ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -114,6 +172,5 @@ public abstract class ChannelSaver<T>
|
||||
private void RecordException(Exception ex)
|
||||
{
|
||||
Exception = ex;
|
||||
_checkCacheChannel.Writer.TryComplete(ex);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using System.Diagnostics.Contracts;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text;
|
||||
#if NET6_0_OR_GREATER
|
||||
@@ -8,47 +9,58 @@ using System.Runtime.InteropServices;
|
||||
|
||||
namespace Speckle.Sdk.Common;
|
||||
|
||||
/// <summary>
|
||||
/// Helpers for hashing data to a hex string
|
||||
/// </summary>
|
||||
public static class Sha256
|
||||
{
|
||||
public const string DEFAULT_FORMAT = "x2";
|
||||
public const int HASH_SIZE_CHARS = 64; // SHA256.HashSizeInBytes * sizeof(char)
|
||||
#if NET6_0_OR_GREATER
|
||||
/// <param name="input">the value to hash</param>
|
||||
/// <param name="format"><c>"x2"</c> for lower case, <c>"X2"</c> for uppercase.</param>
|
||||
/// <param name="length">Desired length of the returned string. Must be 2 ≤ Length ≤ 64, and must be a multiple of 2</param>
|
||||
/// <returns><inheritdoc cref="GetString(string, string?, int)"/></returns>
|
||||
[Pure]
|
||||
public static string GetString(
|
||||
ReadOnlySpan<char> input,
|
||||
[StringSyntax(StringSyntaxAttribute.NumericFormat)] string? format = "x2",
|
||||
int length = SHA256.HashSizeInBytes * sizeof(char)
|
||||
)
|
||||
/// <param name="destination">Output hash; it must have <c>2 ≤ Length ≤ 64</c>, and must be a multiple of 2</param>
|
||||
/// <param name="formatUpperCase"><see langword="true"/> for upper case, false otherwise</param>
|
||||
public static void Hash(ReadOnlySpan<char> input, bool formatUpperCase, Span<char> destination)
|
||||
{
|
||||
ReadOnlySpan<byte> inputBytes = MemoryMarshal.AsBytes(input);
|
||||
Hash(inputBytes, formatUpperCase, destination);
|
||||
}
|
||||
|
||||
public static void Hash(ReadOnlySpan<byte> input, bool formatUpperCase, Span<char> destination)
|
||||
{
|
||||
Span<byte> hash = stackalloc byte[SHA256.HashSizeInBytes];
|
||||
SHA256.HashData(inputBytes, hash);
|
||||
SHA256.HashData(input, hash);
|
||||
|
||||
Span<char> output = stackalloc char[length];
|
||||
FormatHash(hash, formatUpperCase, destination);
|
||||
}
|
||||
|
||||
for (int i = 0, j = 0; j < length; i += sizeof(byte), j += sizeof(char))
|
||||
public static void Hash(Stream source, bool formatUpperCase, Span<char> destination)
|
||||
{
|
||||
Span<byte> hash = stackalloc byte[SHA256.HashSizeInBytes];
|
||||
SHA256.HashData(source, hash);
|
||||
|
||||
FormatHash(hash, formatUpperCase, destination);
|
||||
}
|
||||
|
||||
private static void FormatHash(ReadOnlySpan<byte> input, bool formatUpperCase, Span<char> output)
|
||||
{
|
||||
for (int i = 0, j = 0; j < output.Length; i += sizeof(byte), j += sizeof(char))
|
||||
{
|
||||
hash[i].TryFormat(output[j..], out _, format);
|
||||
input[i].TryFormat(output[j..], out _, formatUpperCase ? "X2" : "x2");
|
||||
}
|
||||
|
||||
return new string(output);
|
||||
}
|
||||
#endif
|
||||
|
||||
/// <param name="input">the value to hash</param>
|
||||
/// <param name="format"><c>"x2"</c> for lower case, <c>"X2"</c> for uppercase.</param>
|
||||
/// <param name="length">Desired length of the returned string</param>
|
||||
/// <param name="outputLengthChars">Desired length of the returned string</param>
|
||||
/// <returns>the hash string</returns>
|
||||
/// <exception cref="FormatException"><paramref name="format"/> is not a recognised numeric format</exception>
|
||||
/// <exception cref="ArgumentOutOfRangeException"><inheritdoc cref="StringBuilder.ToString(int, int)"/></exception>
|
||||
[Pure]
|
||||
public static string GetString(
|
||||
public static string Hash(
|
||||
string input,
|
||||
[StringSyntax(StringSyntaxAttribute.NumericFormat)] string? format = "x2",
|
||||
int length = 64
|
||||
[StringSyntax(StringSyntaxAttribute.NumericFormat)] string? format = DEFAULT_FORMAT,
|
||||
int outputLengthChars = HASH_SIZE_CHARS
|
||||
)
|
||||
{
|
||||
var inputBytes = Encoding.Unicode.GetBytes(input);
|
||||
@@ -59,12 +71,43 @@ public static class Sha256
|
||||
byte[] hash = sha256.ComputeHash(inputBytes);
|
||||
#endif
|
||||
|
||||
StringBuilder sb = new(64);
|
||||
StringBuilder sb = new(HASH_SIZE_CHARS);
|
||||
foreach (byte b in hash)
|
||||
{
|
||||
sb.Append(b.ToString(format));
|
||||
}
|
||||
|
||||
return sb.ToString(0, length);
|
||||
return sb.ToString(0, outputLengthChars);
|
||||
}
|
||||
|
||||
/// <inheritdoc cref="Hash(string, string?, int)"/>
|
||||
[Pure]
|
||||
public static string Hash(
|
||||
Stream input,
|
||||
[StringSyntax(StringSyntaxAttribute.NumericFormat)] string? format = DEFAULT_FORMAT,
|
||||
int outputLengthChars = HASH_SIZE_CHARS
|
||||
)
|
||||
{
|
||||
#if NET6_0_OR_GREATER
|
||||
byte[] hash = SHA256.HashData(input);
|
||||
#else
|
||||
using var sha256 = SHA256.Create();
|
||||
byte[] hash = sha256.ComputeHash(input);
|
||||
#endif
|
||||
|
||||
return FormatHash(hash, format, outputLengthChars);
|
||||
}
|
||||
|
||||
[Pure]
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
private static string FormatHash(byte[] hash, string? format, int outputLengthChars)
|
||||
{
|
||||
StringBuilder sb = new(HASH_SIZE_CHARS);
|
||||
foreach (byte b in hash)
|
||||
{
|
||||
sb.Append(b.ToString(format));
|
||||
}
|
||||
|
||||
return sb.ToString(0, outputLengthChars);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,38 +1,39 @@
|
||||
using System.Runtime.Serialization;
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using System.Runtime.Serialization;
|
||||
using Speckle.Newtonsoft.Json;
|
||||
|
||||
namespace Speckle.Sdk.Models;
|
||||
|
||||
[SpeckleType("Speckle.Core.Models.Blob")]
|
||||
public class Blob : Base
|
||||
public sealed class Blob : Base
|
||||
{
|
||||
[JsonIgnore]
|
||||
public static int LocalHashPrefixLength => 20;
|
||||
|
||||
private string _filePath;
|
||||
private string _hash;
|
||||
private string? _hash;
|
||||
private bool _isHashExpired = true;
|
||||
|
||||
public Blob() { }
|
||||
|
||||
[SetsRequiredMembers]
|
||||
public Blob(string filePath)
|
||||
{
|
||||
this.filePath = filePath;
|
||||
this.originalPath = filePath;
|
||||
}
|
||||
|
||||
public string filePath
|
||||
public required string filePath
|
||||
{
|
||||
get => _filePath;
|
||||
set
|
||||
{
|
||||
originalPath ??= value;
|
||||
|
||||
_filePath = value;
|
||||
_isHashExpired = true;
|
||||
}
|
||||
}
|
||||
public required string originalPath { get; set; }
|
||||
|
||||
public string originalPath { get; set; }
|
||||
[JsonIgnore]
|
||||
public FileInfo FileInfo => new(filePath);
|
||||
|
||||
/// <summary>
|
||||
/// For blobs, the id is the same as the file hash. Please note, when deserialising, the id will be set from the original hash generated on sending.
|
||||
@@ -45,9 +46,9 @@ public class Blob : Base
|
||||
|
||||
public string? GetFileHash()
|
||||
{
|
||||
if ((_isHashExpired || _hash == null) && filePath != null)
|
||||
if ((_isHashExpired || _hash == null))
|
||||
{
|
||||
_hash = HashUtility.HashFile(filePath);
|
||||
_hash = HashUtility.CalculateBlobHash(filePath);
|
||||
}
|
||||
|
||||
return _hash;
|
||||
|
||||
@@ -1,26 +1,39 @@
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using System.Security.Cryptography;
|
||||
using System.Diagnostics.Contracts;
|
||||
using Speckle.Sdk.Common;
|
||||
using Speckle.Sdk.Serialisation;
|
||||
|
||||
namespace Speckle.Sdk.Models;
|
||||
|
||||
/// <summary>
|
||||
/// Helper functions for calculating hash based Ids for Speckle core concepts
|
||||
/// </summary>
|
||||
public static class HashUtility
|
||||
{
|
||||
public enum HashingFunctions
|
||||
public const int HASH_LENGTH_CHARS = 32;
|
||||
|
||||
[Pure]
|
||||
public static Id ComputeObjectId(Json serialized)
|
||||
{
|
||||
SHA256,
|
||||
MD5,
|
||||
#if NET6_0_OR_GREATER
|
||||
Span<char> hash = stackalloc char[HASH_LENGTH_CHARS];
|
||||
Sha256.Hash(serialized.Value.AsSpan(), false, hash);
|
||||
return new Id(new string(hash));
|
||||
#else
|
||||
string hash = Sha256.Hash(serialized.Value, outputLengthChars: HashUtility.HASH_LENGTH_CHARS);
|
||||
return new Id(hash);
|
||||
#endif
|
||||
}
|
||||
|
||||
public const int HASH_LENGTH = 32;
|
||||
|
||||
[SuppressMessage("Security", "CA5351:Do Not Use Broken Cryptographic Algorithms")]
|
||||
public static string HashFile(string filePath, HashingFunctions func = HashingFunctions.SHA256)
|
||||
[Pure]
|
||||
public static string CalculateBlobHash(string filePath)
|
||||
{
|
||||
using HashAlgorithm hashAlgorithm = func == HashingFunctions.MD5 ? MD5.Create() : SHA256.Create();
|
||||
|
||||
using var stream = File.OpenRead(filePath);
|
||||
|
||||
var hash = hashAlgorithm.ComputeHash(stream);
|
||||
return BitConverter.ToString(hash, 0, HASH_LENGTH).Replace("-", "").ToLowerInvariant();
|
||||
#if NET6_0_OR_GREATER
|
||||
Span<char> hash = stackalloc char[HASH_LENGTH_CHARS];
|
||||
Sha256.Hash(stream, false, hash);
|
||||
return new(hash);
|
||||
#else
|
||||
return Sha256.Hash(stream, "x2", HASH_LENGTH_CHARS);
|
||||
#endif
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,19 +0,0 @@
|
||||
using System.Diagnostics.Contracts;
|
||||
using Speckle.Sdk.Common;
|
||||
using Speckle.Sdk.Models;
|
||||
|
||||
namespace Speckle.Sdk.Serialisation;
|
||||
|
||||
public static class IdGenerator
|
||||
{
|
||||
[Pure]
|
||||
public static Id ComputeId(Json serialized)
|
||||
{
|
||||
#if NET6_0_OR_GREATER
|
||||
string hash = Sha256.GetString(serialized.Value.AsSpan(), length: HashUtility.HASH_LENGTH);
|
||||
#else
|
||||
string hash = Sha256.GetString(serialized.Value, length: HashUtility.HASH_LENGTH);
|
||||
#endif
|
||||
return new Id(hash);
|
||||
}
|
||||
}
|
||||
@@ -358,7 +358,7 @@ public class SpeckleObjectSerializer
|
||||
if (writer is SerializerIdWriter serializerIdWriter)
|
||||
{
|
||||
(var json, writer) = serializerIdWriter.FinishIdWriter();
|
||||
id = IdGenerator.ComputeId(json);
|
||||
id = HashUtility.ComputeObjectId(json);
|
||||
}
|
||||
else
|
||||
{
|
||||
|
||||
@@ -1,12 +1,13 @@
|
||||
using System.Text;
|
||||
using Speckle.Sdk.Models;
|
||||
|
||||
namespace Speckle.Sdk.Serialisation.V2.Send;
|
||||
|
||||
public sealed record BaseItem(Id Id, Json Json, bool NeedsStorage, Dictionary<Id, int>? Closures) : IHasByteSize
|
||||
public record BaseItem(Id Id, Json Json, bool NeedsStorage, Dictionary<Id, int>? Closures) : IHasByteSize
|
||||
{
|
||||
public int ByteSize { get; } = Encoding.UTF8.GetByteCount(Json.Value);
|
||||
public virtual int ByteSize { get; } = Encoding.UTF8.GetByteCount(Json.Value);
|
||||
|
||||
public bool Equals(BaseItem? other)
|
||||
public virtual bool Equals(BaseItem? other)
|
||||
{
|
||||
if (other is null)
|
||||
{
|
||||
@@ -17,3 +18,10 @@ public sealed record BaseItem(Id Id, Json Json, bool NeedsStorage, Dictionary<Id
|
||||
|
||||
public override int GetHashCode() => Id.GetHashCode();
|
||||
}
|
||||
|
||||
public sealed record BlobItem(Id Id, Json Json, bool NeedsStorage, Dictionary<Id, int>? Closures, Blob Blob)
|
||||
: BaseItem(Id, Json, NeedsStorage, Closures)
|
||||
{
|
||||
public Blob Blob { get; } = Blob;
|
||||
public override int ByteSize { get; } = (int)Blob.FileInfo.Length;
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Speckle.Sdk.Common;
|
||||
using Speckle.Sdk.Dependencies;
|
||||
using Speckle.Sdk.Dependencies.Serialization;
|
||||
using Speckle.Sdk.SQLite;
|
||||
@@ -9,7 +10,13 @@ namespace Speckle.Sdk.Serialisation.V2.Send;
|
||||
public interface IObjectSaver : IDisposable
|
||||
{
|
||||
Exception? Exception { get; set; }
|
||||
Task Start(int? maxParallelism, int? httpBatchSize, int? cacheBatchSize, CancellationToken cancellationToken);
|
||||
Task Start(
|
||||
int? maxParallelism,
|
||||
int? httpBatchSize,
|
||||
int? blobBatchSize,
|
||||
int? cacheBatchSize,
|
||||
CancellationToken cancellationToken
|
||||
);
|
||||
void DoneTraversing();
|
||||
Task DoneSaving();
|
||||
Task SaveAsync(BaseItem item);
|
||||
@@ -19,14 +26,11 @@ public sealed class ObjectSaver(
|
||||
IProgress<ProgressArgs>? progress,
|
||||
ISqLiteJsonCacheManager sqLiteJsonCacheManager,
|
||||
IServerObjectManager serverObjectManager,
|
||||
IServerBlobManager? serverBlobManager,
|
||||
ILogger<ObjectSaver> logger,
|
||||
SerializeProcessOptions options,
|
||||
CancellationToken cancellationToken
|
||||
#pragma warning disable CS9107
|
||||
#pragma warning disable CA2254
|
||||
) : ChannelSaver<BaseItem>, IObjectSaver
|
||||
#pragma warning restore CA2254
|
||||
#pragma warning restore CS9107
|
||||
) : ChannelSaver<BaseItem, BlobItem>, IObjectSaver
|
||||
{
|
||||
private readonly CancellationTokenSource _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(
|
||||
cancellationToken
|
||||
@@ -40,6 +44,24 @@ public sealed class ObjectSaver(
|
||||
private long _objectsSerialized;
|
||||
private bool _disposed;
|
||||
|
||||
protected override async Task SendBlobToServerInternal(Batch<BlobItem> batch)
|
||||
{
|
||||
// Callers should either setup a blob manager, or not try and send blobs
|
||||
serverBlobManager.NotNull("No blob manager was setup to handle sending blobs");
|
||||
|
||||
var objectBatch = batch.Items.Distinct().Select(x => (x.Blob.id.NotNull(), x.Blob.filePath)).ToList();
|
||||
// var hasObjects = await serverBlobManager
|
||||
// .HasObjects(objectBatch.Select(x => x.Id.Value).Freeze(), _cancellationTokenSource.Token)
|
||||
// .ConfigureAwait(false);
|
||||
// objectBatch = batch.Items.Where(x => !hasObjects[x.Id.Value]).ToList();
|
||||
if (objectBatch.Count != 0)
|
||||
{
|
||||
// Interlocked.Add(ref _uploading, batch.Items.Count);
|
||||
// progress?.Report(new(ProgressEvent.UploadingObjects, _uploading, null));
|
||||
await serverBlobManager.UploadBlobs(objectBatch, progress, _cancellationTokenSource.Token).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
protected override async Task SendToServerInternal(Batch<BaseItem> batch)
|
||||
{
|
||||
if (IsCancelled())
|
||||
|
||||
@@ -343,7 +343,7 @@ public sealed class ObjectSerializer : IObjectSerializer
|
||||
if (writer is SerializerIdWriter serializerIdWriter)
|
||||
{
|
||||
(var json, writer) = serializerIdWriter.FinishIdWriter();
|
||||
id = IdGenerator.ComputeId(json);
|
||||
id = HashUtility.ComputeObjectId(json);
|
||||
}
|
||||
else
|
||||
{
|
||||
|
||||
@@ -17,6 +17,7 @@ public record SerializeProcessOptions(
|
||||
{
|
||||
public int? MaxHttpSendBatchSize { get; set; }
|
||||
public int? MaxCacheBatchSize { get; set; }
|
||||
public int? MaxBlobBatchSize { get; set; }
|
||||
public int? MaxParallelism { get; set; }
|
||||
}
|
||||
|
||||
@@ -109,6 +110,7 @@ public sealed class SerializeProcess(
|
||||
var channelTask = objectSaver.Start(
|
||||
options.MaxParallelism,
|
||||
options.MaxHttpSendBatchSize,
|
||||
options.MaxBlobBatchSize,
|
||||
options.MaxCacheBatchSize,
|
||||
_processSource.Token
|
||||
);
|
||||
|
||||
@@ -13,26 +13,36 @@ public class SerializeProcessFactory(
|
||||
IObjectSerializerFactory objectSerializerFactory,
|
||||
ISqLiteJsonCacheManagerFactory sqLiteJsonCacheManagerFactory,
|
||||
IServerObjectManagerFactory serverObjectManagerFactory,
|
||||
IServerBlobManagerFactory serverBlobManagerFactory,
|
||||
ILoggerFactory loggerFactory
|
||||
) : ISerializeProcessFactory
|
||||
{
|
||||
public ISerializeProcess CreateSerializeProcess(
|
||||
Uri url,
|
||||
string streamId,
|
||||
string projectId,
|
||||
string? authorizationToken,
|
||||
IProgress<ProgressArgs>? progress,
|
||||
CancellationToken cancellationToken,
|
||||
SerializeProcessOptions? options = null
|
||||
)
|
||||
{
|
||||
var sqLiteJsonCacheManager = sqLiteJsonCacheManagerFactory.CreateFromStream(streamId);
|
||||
var serverObjectManager = serverObjectManagerFactory.Create(url, streamId, authorizationToken);
|
||||
return CreateSerializeProcess(sqLiteJsonCacheManager, serverObjectManager, progress, cancellationToken, options);
|
||||
var sqLiteJsonCacheManager = sqLiteJsonCacheManagerFactory.CreateFromStream(projectId);
|
||||
var serverObjectManager = serverObjectManagerFactory.Create(url, projectId, authorizationToken);
|
||||
var serverBlobManager = serverBlobManagerFactory.Create(url, projectId, authorizationToken);
|
||||
return CreateSerializeProcess(
|
||||
sqLiteJsonCacheManager,
|
||||
serverObjectManager,
|
||||
serverBlobManager,
|
||||
progress,
|
||||
cancellationToken,
|
||||
options
|
||||
);
|
||||
}
|
||||
|
||||
public ISerializeProcess CreateSerializeProcess(
|
||||
ISqLiteJsonCacheManager sqLiteJsonCacheManager,
|
||||
IServerObjectManager serverObjectManager,
|
||||
IServerBlobManager? serverBlobManager,
|
||||
IProgress<ProgressArgs>? progress,
|
||||
CancellationToken cancellationToken,
|
||||
SerializeProcessOptions? options = null
|
||||
@@ -43,6 +53,7 @@ public class SerializeProcessFactory(
|
||||
progress,
|
||||
sqLiteJsonCacheManager,
|
||||
serverObjectManager,
|
||||
serverBlobManager,
|
||||
loggerFactory.CreateLogger<ObjectSaver>(),
|
||||
options ?? new SerializeProcessOptions(),
|
||||
cancellationToken
|
||||
@@ -68,6 +79,7 @@ public class SerializeProcessFactory(
|
||||
return CreateSerializeProcess(
|
||||
memoryJsonCacheManager,
|
||||
new MemoryServerObjectManager(objects),
|
||||
null!, //this would need a better solution
|
||||
progress,
|
||||
cancellationToken,
|
||||
options
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
using Speckle.InterfaceGenerator;
|
||||
using Speckle.Sdk.Helpers;
|
||||
|
||||
namespace Speckle.Sdk.Serialisation.V2;
|
||||
|
||||
[GenerateAutoInterface]
|
||||
public sealed class ServerBlobManagerFactory(ISpeckleHttp speckleHttp) : IServerBlobManagerFactory
|
||||
{
|
||||
public IServerBlobManager Create(
|
||||
Uri serverUrl,
|
||||
string projectId,
|
||||
string? authorizationToken,
|
||||
TimeSpan? timeout = null
|
||||
)
|
||||
{
|
||||
var client = speckleHttp.CreateHttpClient(authorizationToken: authorizationToken);
|
||||
client.BaseAddress = serverUrl;
|
||||
return new ServerBlobManager(client, projectId);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
using Speckle.InterfaceGenerator;
|
||||
using Speckle.Sdk.Transports;
|
||||
using Speckle.Sdk.Transports.ServerUtils;
|
||||
|
||||
namespace Speckle.Sdk.Serialisation.V2;
|
||||
|
||||
[GenerateAutoInterface(VisibilityModifier = "public")]
|
||||
internal sealed class ServerBlobManager(HttpClient authorizedClient, string projectId) : IServerBlobManager
|
||||
{
|
||||
public async Task UploadBlobs(
|
||||
IReadOnlyCollection<(string blobId, string filePath)> objects,
|
||||
IProgress<ProgressArgs>? progress,
|
||||
CancellationToken cancellationToken
|
||||
)
|
||||
{
|
||||
if (objects.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var multipartFormDataContent = new MultipartFormDataContent();
|
||||
foreach (var (id, filePath) in objects)
|
||||
{
|
||||
var fileName = Path.GetFileName(filePath);
|
||||
var stream = File.OpenRead(filePath);
|
||||
StreamContent fsc = new(stream);
|
||||
|
||||
multipartFormDataContent.Add(fsc, $"hash:{id}", fileName);
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
}
|
||||
|
||||
using var message = new HttpRequestMessage();
|
||||
message.RequestUri = new Uri($"/api/stream/{projectId}/blob", UriKind.Relative);
|
||||
message.Method = HttpMethod.Post;
|
||||
message.Content = new ProgressContent(multipartFormDataContent, progress);
|
||||
|
||||
using var response = await authorizedClient.SendAsync(message, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
response.EnsureSuccessStatusCode();
|
||||
}
|
||||
}
|
||||
@@ -59,6 +59,7 @@ public class CancellationTests
|
||||
new DummySqLiteSendManager(),
|
||||
new CancellationServerObjectManager(cancellationSource),
|
||||
null,
|
||||
null,
|
||||
cancellationSource.Token,
|
||||
new SerializeProcessOptions(true, true, false, true)
|
||||
);
|
||||
@@ -79,6 +80,7 @@ public class CancellationTests
|
||||
new DummySqLiteSendManager(),
|
||||
new CancellationServerObjectManager(cancellationSource),
|
||||
null,
|
||||
null,
|
||||
cancellationSource.Token,
|
||||
new SerializeProcessOptions(true, true, false, true)
|
||||
);
|
||||
|
||||
@@ -40,6 +40,7 @@ public class DataObjectTests
|
||||
new MemoryJsonCacheManager(json),
|
||||
new DummyServerObjectManager(),
|
||||
null,
|
||||
null,
|
||||
default,
|
||||
new SerializeProcessOptions(false, false, true, true)
|
||||
);
|
||||
|
||||
@@ -37,6 +37,7 @@ public class ExceptionTests
|
||||
new MemoryJsonCacheManager(objects),
|
||||
new ExceptionServerObjectManager(),
|
||||
null,
|
||||
null,
|
||||
default,
|
||||
new SerializeProcessOptions(false, false, false, true)
|
||||
);
|
||||
@@ -55,6 +56,7 @@ public class ExceptionTests
|
||||
new ExceptionSendCacheManager(),
|
||||
new MemoryServerObjectManager(new()),
|
||||
null,
|
||||
null,
|
||||
default,
|
||||
new SerializeProcessOptions(false, false, false, true)
|
||||
);
|
||||
@@ -92,6 +94,7 @@ public class ExceptionTests
|
||||
new ExceptionSendCacheManager(exceptionsAfter: 10),
|
||||
new MemoryServerObjectManager(new()),
|
||||
null,
|
||||
null,
|
||||
default,
|
||||
new SerializeProcessOptions(false, false, false, true)
|
||||
{
|
||||
|
||||
@@ -146,7 +146,7 @@ public class SerializationTests
|
||||
jObject.Remove("id");
|
||||
jObject.Remove("__closure");
|
||||
var jsonWithoutId = jObject.ToString(Formatting.None);
|
||||
var newId = IdGenerator.ComputeId(new Json(jsonWithoutId));
|
||||
var newId = HashUtility.ComputeObjectId(new Json(jsonWithoutId));
|
||||
id.Should().Be(newId.Value);
|
||||
}
|
||||
|
||||
@@ -227,6 +227,7 @@ public class SerializationTests
|
||||
SqLiteJsonCacheManager.FromMemory(1),
|
||||
new MemoryServerObjectManager(newIdToJson),
|
||||
null,
|
||||
null,
|
||||
default,
|
||||
new SerializeProcessOptions(false, false, false, true) { MaxCacheBatchSize = 1, MaxParallelism = concurrency }
|
||||
)
|
||||
|
||||
@@ -60,7 +60,7 @@ public class BlobApiExceptionalTests : IAsyncLifetime
|
||||
{
|
||||
await writer.WriteLineAsync(PAYLOAD);
|
||||
}
|
||||
string id = HashUtility.HashFile(filePath);
|
||||
string id = HashUtility.CalculateBlobHash(filePath);
|
||||
var ex = await Assert.ThrowsAsync<HttpRequestException>(async () =>
|
||||
await _sut.UploadBlobs("non-existent-project", [(id, filePath)], null, CancellationToken.None)
|
||||
);
|
||||
|
||||
@@ -34,7 +34,7 @@ public class BlobApiTests : IAsyncLifetime
|
||||
{
|
||||
await writer.WriteLineAsync(PAYLOAD);
|
||||
}
|
||||
string id = HashUtility.HashFile(filePath);
|
||||
string id = HashUtility.CalculateBlobHash(filePath);
|
||||
|
||||
//act
|
||||
var preDiff = await _blobApi.HasBlobs(_project.id, [id], CancellationToken.None);
|
||||
|
||||
+31
-30
@@ -10,7 +10,12 @@ namespace Speckle.Sdk.Tests.Integration.API.GraphQL.Resources;
|
||||
|
||||
public class SubscriptionResourceTests : IAsyncLifetime
|
||||
{
|
||||
private const int WAIT_PERIOD = 300;
|
||||
#if DEBUG
|
||||
private const int WAIT_PERIOD = 3000; // WSL is slow AF, so for local runs, we're being extra generous
|
||||
#else
|
||||
private const int WAIT_PERIOD = 400; // For CI runs, a much smaller wait time is acceptable
|
||||
#endif
|
||||
private const int TIMEOUT = WAIT_PERIOD + WAIT_PERIOD + 400;
|
||||
private IClient _testUser;
|
||||
private Project _testProject;
|
||||
private Model _testModel;
|
||||
@@ -32,105 +37,101 @@ public class SubscriptionResourceTests : IAsyncLifetime
|
||||
_testVersion = await Fixtures.CreateVersion(_testUser, _testProject.id, _testModel.id);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
[Fact(Timeout = TIMEOUT)]
|
||||
public async Task UserProjectsUpdated_SubscriptionIsCalled()
|
||||
{
|
||||
UserProjectsUpdatedMessage? subscriptionMessage = null;
|
||||
|
||||
TaskCompletionSource<UserProjectsUpdatedMessage> tcs = new();
|
||||
using var sub = Sut.CreateUserProjectsUpdatedSubscription();
|
||||
sub.Listeners += (_, message) => subscriptionMessage = message;
|
||||
sub.Listeners += (_, message) => tcs.SetResult(message);
|
||||
|
||||
await Task.Delay(WAIT_PERIOD); // Give time to subscription to be setup
|
||||
|
||||
var created = await _testUser.Project.Create(new(null, null, null));
|
||||
|
||||
await Task.Delay(WAIT_PERIOD); // Give time for subscription to be triggered
|
||||
var subscriptionMessage = await tcs.Task;
|
||||
|
||||
subscriptionMessage.Should().NotBeNull();
|
||||
subscriptionMessage!.id.Should().Be(created.id);
|
||||
subscriptionMessage.id.Should().Be(created.id);
|
||||
subscriptionMessage.type.Should().Be(UserProjectsUpdatedMessageType.ADDED);
|
||||
subscriptionMessage.project.Should().NotBeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
[Fact(Timeout = TIMEOUT)]
|
||||
public async Task ProjectModelsUpdated_SubscriptionIsCalled()
|
||||
{
|
||||
ProjectModelsUpdatedMessage? subscriptionMessage = null;
|
||||
|
||||
TaskCompletionSource<ProjectModelsUpdatedMessage> tcs = new();
|
||||
using var sub = Sut.CreateProjectModelsUpdatedSubscription(_testProject.id);
|
||||
sub.Listeners += (_, message) => subscriptionMessage = message;
|
||||
sub.Listeners += (_, message) => tcs.SetResult(message);
|
||||
|
||||
await Task.Delay(WAIT_PERIOD); // Give time to subscription to be setup
|
||||
|
||||
CreateModelInput input = new("my model", "myDescription", _testProject.id);
|
||||
var created = await _testUser.Model.Create(input);
|
||||
|
||||
await Task.Delay(WAIT_PERIOD); // Give time for subscription to be triggered
|
||||
var subscriptionMessage = await tcs.Task;
|
||||
|
||||
subscriptionMessage.Should().NotBeNull();
|
||||
subscriptionMessage!.id.Should().Be(created.id);
|
||||
subscriptionMessage.id.Should().Be(created.id);
|
||||
subscriptionMessage.type.Should().Be(ProjectModelsUpdatedMessageType.CREATED);
|
||||
subscriptionMessage.model.Should().NotBeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
[Fact(Timeout = TIMEOUT)]
|
||||
public async Task ProjectUpdated_SubscriptionIsCalled()
|
||||
{
|
||||
ProjectUpdatedMessage? subscriptionMessage = null;
|
||||
|
||||
TaskCompletionSource<ProjectUpdatedMessage> tcs = new();
|
||||
using var sub = Sut.CreateProjectUpdatedSubscription(_testProject.id);
|
||||
sub.Listeners += (_, message) => subscriptionMessage = message;
|
||||
sub.Listeners += (_, message) => tcs.SetResult(message);
|
||||
|
||||
await Task.Delay(WAIT_PERIOD); // Give time to subscription to be setup
|
||||
|
||||
var input = new ProjectUpdateInput(_testProject.id, "This is my new name");
|
||||
var created = await _testUser.Project.Update(input);
|
||||
|
||||
await Task.Delay(WAIT_PERIOD); // Give time for subscription to be triggered
|
||||
var subscriptionMessage = await tcs.Task;
|
||||
|
||||
subscriptionMessage.Should().NotBeNull();
|
||||
subscriptionMessage!.id.Should().Be(created.id);
|
||||
subscriptionMessage.id.Should().Be(created.id);
|
||||
subscriptionMessage.type.Should().Be(ProjectUpdatedMessageType.UPDATED);
|
||||
subscriptionMessage.project.Should().NotBeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
[Fact(Timeout = TIMEOUT)]
|
||||
public async Task ProjectVersionsUpdated_SubscriptionIsCalled()
|
||||
{
|
||||
ProjectVersionsUpdatedMessage? subscriptionMessage = null;
|
||||
|
||||
TaskCompletionSource<ProjectVersionsUpdatedMessage> tcs = new();
|
||||
using var sub = Sut.CreateProjectVersionsUpdatedSubscription(_testProject.id);
|
||||
sub.Listeners += (_, message) => subscriptionMessage = message;
|
||||
sub.Listeners += (_, message) => tcs.SetResult(message);
|
||||
|
||||
await Task.Delay(WAIT_PERIOD); // Give time to subscription to be setup
|
||||
|
||||
var created = await Fixtures.CreateVersion(_testUser, _testProject.id, _testModel.id);
|
||||
|
||||
await Task.Delay(WAIT_PERIOD); // Give time for subscription to be triggered
|
||||
var subscriptionMessage = await tcs.Task;
|
||||
|
||||
subscriptionMessage.Should().NotBeNull();
|
||||
subscriptionMessage!.id.Should().Be(created.id);
|
||||
subscriptionMessage.id.Should().Be(created.id);
|
||||
subscriptionMessage.type.Should().Be(ProjectVersionsUpdatedMessageType.CREATED);
|
||||
subscriptionMessage.version.Should().NotBeNull();
|
||||
}
|
||||
|
||||
[Fact(Skip = CommentResourceTests.SERVER_SKIP_MESSAGE)]
|
||||
[Fact(Skip = CommentResourceTests.SERVER_SKIP_MESSAGE, Timeout = TIMEOUT)]
|
||||
public async Task ProjectCommentsUpdated_SubscriptionIsCalled()
|
||||
{
|
||||
string resourceIdString = $"{_testProject.id},{_testModel.id},{_testVersion}";
|
||||
ProjectCommentsUpdatedMessage? subscriptionMessage = null;
|
||||
|
||||
TaskCompletionSource<ProjectCommentsUpdatedMessage> tcs = new();
|
||||
using var sub = Sut.CreateProjectCommentsUpdatedSubscription(new(_testProject.id, resourceIdString));
|
||||
sub.Listeners += (_, message) => subscriptionMessage = message;
|
||||
sub.Listeners += (_, message) => tcs.SetResult(message);
|
||||
|
||||
await Task.Delay(WAIT_PERIOD); // Give time to subscription to be setup
|
||||
|
||||
var created = await Fixtures.CreateComment(_testUser, _testProject.id, _testModel.id, _testVersion.id);
|
||||
|
||||
await Task.Delay(WAIT_PERIOD); // Give time for subscription to be triggered
|
||||
var subscriptionMessage = await tcs.Task;
|
||||
|
||||
subscriptionMessage.Should().NotBeNull();
|
||||
subscriptionMessage!.id.Should().Be(created.id);
|
||||
subscriptionMessage.id.Should().Be(created.id);
|
||||
subscriptionMessage.type.Should().Be(ProjectCommentsUpdatedMessageType.CREATED);
|
||||
subscriptionMessage.comment.Should().NotBeNull();
|
||||
}
|
||||
|
||||
@@ -1,85 +0,0 @@
|
||||
using FluentAssertions;
|
||||
using GraphQL.Client.Http;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Speckle.Sdk.Api.GraphQL.Models;
|
||||
using Speckle.Sdk.Credentials;
|
||||
|
||||
namespace Speckle.Sdk.Tests.Integration.Credentials;
|
||||
|
||||
public class UserServerInfoTests : IAsyncLifetime
|
||||
{
|
||||
private Account _acc;
|
||||
|
||||
public Task DisposeAsync() => Task.CompletedTask;
|
||||
|
||||
public async Task InitializeAsync()
|
||||
{
|
||||
_acc = await Fixtures.SeedUser();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task IsFrontEnd2True()
|
||||
{
|
||||
ServerInfo result = await Fixtures
|
||||
.ServiceProvider.GetRequiredService<IAccountManager>()
|
||||
.GetServerInfo(new("https://app.speckle.systems/"));
|
||||
|
||||
result.Should().NotBeNull();
|
||||
result.frontend2.Should().BeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task GetServerInfo_ExpectFail_NoServer()
|
||||
{
|
||||
Uri serverUrl = new("http://invalidserver.local");
|
||||
|
||||
await FluentActions
|
||||
.Invoking(async () =>
|
||||
await Fixtures.ServiceProvider.GetRequiredService<IAccountManager>().GetServerInfo(serverUrl)
|
||||
)
|
||||
.Should()
|
||||
.ThrowAsync<HttpRequestException>();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task GetUserInfo()
|
||||
{
|
||||
Uri serverUrl = new(_acc.serverInfo.url);
|
||||
UserInfo result = await Fixtures
|
||||
.ServiceProvider.GetRequiredService<IAccountManager>()
|
||||
.GetUserInfo(_acc.token, serverUrl);
|
||||
|
||||
result.id.Should().Be(_acc.userInfo.id);
|
||||
result.name.Should().Be(_acc.userInfo.name);
|
||||
result.email.Should().Be(_acc.userInfo.email);
|
||||
result.company.Should().Be(_acc.userInfo.company);
|
||||
result.avatar.Should().Be(_acc.userInfo.avatar);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task GetUserInfo_ExpectFail_NoServer()
|
||||
{
|
||||
Uri serverUrl = new("http://invalidserver.local");
|
||||
|
||||
await FluentActions
|
||||
.Invoking(async () =>
|
||||
await Fixtures.ServiceProvider.GetRequiredService<IAccountManager>().GetUserInfo("", serverUrl)
|
||||
)
|
||||
.Should()
|
||||
.ThrowAsync<HttpRequestException>();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task GetUserInfo_ExpectFail_NoUser()
|
||||
{
|
||||
Uri serverUrl = new(_acc.serverInfo.url);
|
||||
await FluentActions
|
||||
.Invoking(async () =>
|
||||
await Fixtures
|
||||
.ServiceProvider.GetRequiredService<IAccountManager>()
|
||||
.GetUserInfo("Bearer 08913c3c1e7ac65d779d1e1f11b942a44ad9672ca9", serverUrl)
|
||||
)
|
||||
.Should()
|
||||
.ThrowAsync<GraphQLHttpRequestException>();
|
||||
}
|
||||
}
|
||||
@@ -19,12 +19,14 @@ public class CryptSha256Hash
|
||||
[Benchmark]
|
||||
public string Sha256()
|
||||
{
|
||||
return Speckle.Sdk.Common.Sha256.GetString(testData);
|
||||
return Speckle.Sdk.Common.Sha256.Hash(testData);
|
||||
}
|
||||
|
||||
[Benchmark]
|
||||
public string Sha256_Span()
|
||||
{
|
||||
return Speckle.Sdk.Common.Sha256.GetString(testData.AsSpan());
|
||||
Span<char> resultLowerSpan = stackalloc char[Speckle.Sdk.Common.Sha256.HASH_SIZE_CHARS];
|
||||
Speckle.Sdk.Common.Sha256.Hash(testData.AsSpan(), false, resultLowerSpan);
|
||||
return new string(resultLowerSpan);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -69,8 +69,8 @@ public sealed class HashUtilityTests
|
||||
[MemberData(nameof(SmallTestCasesSha256))]
|
||||
public void Sha256(string input, string expected, string _, int length)
|
||||
{
|
||||
var resultLower = Speckle.Sdk.Common.Sha256.GetString(input, "x2", length);
|
||||
var resultUpper = Speckle.Sdk.Common.Sha256.GetString(input, "X2", length);
|
||||
var resultLower = Speckle.Sdk.Common.Sha256.Hash(input, "x2", length);
|
||||
var resultUpper = Speckle.Sdk.Common.Sha256.Hash(input, "X2", length);
|
||||
|
||||
resultLower.Should().Be(new string(expected.ToLower()[..length]));
|
||||
|
||||
@@ -86,19 +86,22 @@ public sealed class HashUtilityTests
|
||||
int length //Span version of the function must have multiple of 2
|
||||
)
|
||||
{
|
||||
var resultLowerSpan = Speckle.Sdk.Common.Sha256.GetString(input.AsSpan(), "x2", length);
|
||||
var resultUpperSpan = Speckle.Sdk.Common.Sha256.GetString(input.AsSpan(), "X2", length);
|
||||
Span<char> resultLowerSpan = stackalloc char[length];
|
||||
Speckle.Sdk.Common.Sha256.Hash(input.AsSpan(), false, resultLowerSpan);
|
||||
Span<char> resultUpperSpan = stackalloc char[length];
|
||||
Speckle.Sdk.Common.Sha256.Hash(input.AsSpan(), true, resultUpperSpan);
|
||||
|
||||
resultLowerSpan.Should().Be(new string(expected.ToLower()[..length]));
|
||||
new string(resultLowerSpan).Should().Be(new string(expected.ToLower()[..length]));
|
||||
|
||||
resultUpperSpan.Should().Be(new string(expected.ToUpper()[..length]));
|
||||
new string(resultUpperSpan).Should().Be(new string(expected.ToUpper()[..length]));
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[MemberData(nameof(LargeTestCases))]
|
||||
public void Sha256_LargeDataTests(string input, string expected)
|
||||
public void Sha256_Span_LargeDataTests(string input, string expected)
|
||||
{
|
||||
var computedHash = Speckle.Sdk.Common.Sha256.GetString(input.AsSpan());
|
||||
computedHash.Should().Be(expected);
|
||||
Span<char> output = stackalloc char[Speckle.Sdk.Common.Sha256.HASH_SIZE_CHARS];
|
||||
Speckle.Sdk.Common.Sha256.Hash(input.AsSpan(), false, output);
|
||||
new string(output).Should().Be(expected);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user