Compare commits

..

2 Commits

Author SHA1 Message Date
Jedd Morgan a560f7f159 more blobs poc 2025-11-24 18:28:50 +00:00
JR-Morgan 94d2a01880 blobs-poc 2025-11-13 11:01:26 +03:00
56 changed files with 451 additions and 1292 deletions
+5 -14
View File
@@ -6,12 +6,10 @@ on:
docker-compose-file:
required: true
type: string
use-internal-image:
use-github-container-registry:
default: false
type: boolean
secrets:
CODECOV_TOKEN:
required: true
jobs:
integration-test:
env:
@@ -19,7 +17,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v6
uses: actions/checkout@v5
- name: Setup .NET
uses: actions/setup-dotnet@v5
@@ -29,7 +27,7 @@ jobs:
cache-dependency-path: "**/packages.lock.json"
- name: 🔐 Login to Github Container Registry
if: ${{ inputs.use-internal-image }}
if: ${{ inputs.use-github-container-registry }}
uses: docker/login-action@v3
with:
registry: "ghcr.io"
@@ -45,18 +43,11 @@ jobs:
- name: 🏗️ Build
run: dotnet build ${{ env.Solution }} --configuration Release --no-restore -warnaserror
- name: 🔨 Integration Tests against Public Server
if: ${{ !inputs.use-internal-image }}
run: dotnet test ${{ env.Solution }} --filter "(Category=Integration)&(Server!=Internal)" --configuration Release --no-build --no-restore --verbosity=normal /p:AltCover=true /p:AltCoverAttributeFilter=ExcludeFromCodeCoverage
- name: 🔨 Integration Tests against Internal Server
if: ${{ inputs.use-internal-image }}
- name: 🔨 Integration Tests
run: dotnet test ${{ env.Solution }} --filter "Category=Integration" --configuration Release --no-build --no-restore --verbosity=normal /p:AltCover=true /p:AltCoverAttributeFilter=ExcludeFromCodeCoverage
- name: Upload coverage reports to Codecov with GitHub Action
uses: codecov/codecov-action@v5
continue-on-error: true
with:
fail_ci_if_error: true
files: tests/**/coverage.xml
token: ${{ secrets.CODECOV_TOKEN }}
+3 -9
View File
@@ -10,7 +10,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v6
uses: actions/checkout@v5
- name: Setup .NET
uses: actions/setup-dotnet@v5
@@ -39,9 +39,7 @@ jobs:
- name: Upload coverage reports to Codecov with GitHub Action
uses: codecov/codecov-action@v5
continue-on-error: true
with:
fail_ci_if_error: true
files: tests/**/coverage.xml
token: ${{ secrets.CODECOV_TOKEN }}
@@ -49,13 +47,9 @@ jobs:
uses: "./.github/workflows/integration-test.yml"
with:
docker-compose-file: "docker-compose-internal.yml"
use-internal-image: true
secrets:
CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
use-github-container-registry: true
integration-test-public:
uses: "./.github/workflows/integration-test.yml"
with:
docker-compose-file: "docker-compose.yml"
secrets:
CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+1 -3
View File
@@ -14,7 +14,7 @@ jobs:
steps:
- name: Checkout
uses: actions/checkout@v6
uses: actions/checkout@v5
- name: Setup .NET
uses: actions/setup-dotnet@v5
@@ -48,9 +48,7 @@ jobs:
- name: Upload coverage reports to Codecov with GitHub Action
uses: codecov/codecov-action@v5
continue-on-error: true
with:
fail_ci_if_error: true
files: tests/**/coverage.xml
token: ${{ secrets.CODECOV_TOKEN }}
-1
View File
@@ -1 +0,0 @@
dotnet 8.0.400
-1
View File
@@ -17,7 +17,6 @@
<File Path=".github\git-commit-instructions.md" />
</Folder>
<Folder Name="/config/workflows/">
<File Path=".github/workflows/integration-test.yml" />
<File Path=".github/workflows/pr.yml" />
<File Path=".github/workflows/release.yml" />
</Folder>
@@ -63,7 +63,7 @@ internal sealed class AutomationContext(IOperations operations) : IAutomationCon
);
}
Base rootObject = await operations
Base? rootObject = await operations
.Receive2(
SpeckleClient.ServerUrl,
AutomationRunData.ProjectId,
@@ -74,10 +74,6 @@ internal sealed class AutomationContext(IOperations operations) : IAutomationCon
)
.ConfigureAwait(false);
await SpeckleClient
.Version.Received(new(version.id, AutomationRunData.ProjectId, "automate_function"), cancellationToken)
.ConfigureAwait(false);
Console.WriteLine($"It took {Elapsed.TotalSeconds} seconds to receive the speckle version {versionId}");
return rootObject;
}
-6
View File
@@ -1,6 +0,0 @@
using Speckle.Sdk.Models;
namespace Speckle.Objects.Geometry;
[SpeckleType("Objects.Geometry.SolidX")]
public class SolidX : RawEncodedObject;
-2
View File
@@ -20,6 +20,4 @@ public class RawEncoding : Base // note: at this stage, since we're using this f
public static class RawEncodingFormats
{
public const string RHINO_3DM = "3dm";
public const string ACAD_DWG = "dwg";
public const string ACAD_SAT = "sat";
}
@@ -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);
}
}
-2
View File
@@ -35,7 +35,6 @@ public sealed class Client : ISpeckleGraphQLClient, IClient
public WorkspaceResource Workspace { get; }
public ServerResource Server { get; }
public FileImportResource FileImport { get; }
public ModelIngestionResource Ingestion { get; }
public Uri ServerUrl => new(Account.serverInfo.url);
@@ -72,7 +71,6 @@ public sealed class Client : ISpeckleGraphQLClient, IClient
Workspace = new(this);
Server = new(this);
FileImport = new(this, blobApiFactory.Create(account));
Ingestion = new(this);
}
[AutoInterfaceIgnore]
@@ -1,9 +1,6 @@
// ReSharper disable InconsistentNaming
namespace Speckle.Sdk.Api.GraphQL.Enums;
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// This enum isn't explicitly defined in the schema, instead its usages are int typed (But represent an enum)
/// </remarks>
//This enum isn't explicitly defined in the schema, instead its usages are int typed (But represent an enum)
public enum FileUploadConversionStatus
{
Queued = 0,
@@ -1,14 +0,0 @@
// ReSharper disable InconsistentNaming
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// string based enum
/// </remarks>
public enum ModelIngestionStatus
{
cancelled,
failed,
processing,
queued,
success,
}
@@ -1,8 +1,5 @@
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// string based enum
/// </remarks>
public enum ProjectCommentsUpdatedMessageType
{
ARCHIVED,
@@ -1,8 +1,5 @@
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// string based enum
/// </remarks>
public enum ProjectFileImportUpdatedMessageType
{
CREATED,
@@ -1,13 +0,0 @@
// ReSharper disable InconsistentNaming
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// string based enum
/// </remarks>
public enum ProjectModelIngestionUpdatedMessageType
{
cancellationRequested,
created,
deleted,
updated,
}
@@ -1,8 +1,5 @@
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// string based enum
/// </remarks>
public enum ProjectModelsUpdatedMessageType
{
CREATED,
@@ -1,8 +1,5 @@
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// string based enum
/// </remarks>
public enum ProjectPendingModelsUpdatedMessageType
{
CREATED,
@@ -1,8 +1,5 @@
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// string based enum
/// </remarks>
public enum ProjectUpdatedMessageType
{
DELETED,
@@ -1,8 +1,5 @@
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// string based enum
/// </remarks>
public enum ProjectVersionsUpdatedMessageType
{
CREATED,
@@ -1,9 +1,5 @@
// ReSharper disable InconsistentNaming
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// string based enum
/// </remarks>
public enum ProjectVisibility
{
Private,
@@ -1,9 +1,5 @@
// ReSharper disable InconsistentNaming
namespace Speckle.Sdk.Api.GraphQL.Enums;
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// string based enum
/// </remarks>
public enum ResourceType
{
commit,
@@ -1,8 +1,5 @@
namespace Speckle.Sdk.Api.GraphQL.Enums;
/// <remarks>
/// string based enum
/// </remarks>
public enum UserProjectsUpdatedMessageType
{
ADDED,
@@ -1,65 +0,0 @@
using Speckle.Newtonsoft.Json;
using Speckle.Sdk.Api.GraphQL.Enums;
namespace Speckle.Sdk.Api.GraphQL.Inputs;
public record SourceDataInput(
string sourceApplicationSlug,
string sourceApplicationVersion,
string? fileName,
long? fileSizeBytes
);
public record ModelIngestionCreateInput(
string modelId,
string projectId,
string progressMessage,
SourceDataInput sourceData
);
public record ModelIngestionUpdateInput(string ingestionId, string projectId, string progressMessage, double? progress);
public record ModelIngestionSuccessInput(string ingestionId, string projectId, string rootObjectId);
public record ModelIngestionFailedInput(
string ingestionId,
string projectId,
string errorReason,
string? errorStacktrace
)
{
public static ModelIngestionFailedInput FromException(string ingestionId, string projectId, Exception ex)
{
return new ModelIngestionFailedInput(ingestionId, projectId, ex.Message, ex.ToString());
}
}
public record ModelIngestionCancelledInput(string ingestionId, string projectId, string cancellationMessage);
public record ModelIngestionStartProcessingInput(
string ingestionId,
string projectId,
string progressMessage,
SourceDataInput sourceData
);
public record ModelIngestionRequeueInput(string ingestionId, string projectId, string progressMessage);
public record ProjectModelIngestionSubscriptionInput(
string projectId,
ModelIngestionReference ingestionReference,
[property: JsonIgnore] ProjectModelIngestionUpdatedMessageType messageType
)
{
// The Newtonsoft serializer is setup to handle SCREAMING_CASE enums.
// But the API requires the enum to look exactly like they are
[JsonProperty(nameof(messageType))]
public string serializedType => messageType.ToString();
}
/// <remarks>
/// <c>@oneOf</c> i.e. server expects <b>either</b> <paramref name="ingestionId"/> or <paramref name="modelId"/>, but not both.
/// </remarks>
/// <param name="ingestionId"></param>
/// <param name="modelId"></param>
public record ModelIngestionReference(string? ingestionId, string? modelId);
@@ -1,12 +0,0 @@
namespace Speckle.Sdk.Api.GraphQL.Models;
public sealed class ModelIngestion
{
public required string id { get; init; }
public required DateTime createdAt { get; init; }
public required DateTime updatedAt { get; init; }
public required string modelId { get; init; }
public required bool cancellationRequested { get; init; }
public required ModelIngestionStatusData statusData { get; init; }
// public required LimitedUser user { get; init; }
}
@@ -1,9 +0,0 @@
using Speckle.Sdk.Api.GraphQL.Enums;
namespace Speckle.Sdk.Api.GraphQL.Models;
public sealed class ModelIngestionStatusData
{
public required ModelIngestionStatus status { get; init; }
public required string? progressMessage { get; init; }
}
@@ -6,10 +6,10 @@ namespace Speckle.Sdk.Api.GraphQL.Models;
public sealed class UserProjectsUpdatedMessage : EventArgs
{
[JsonRequired]
public required string id { get; init; }
public string id { get; init; }
[JsonRequired]
public required UserProjectsUpdatedMessageType type { get; init; }
public UserProjectsUpdatedMessageType type { get; init; }
public Project? project { get; init; }
}
@@ -17,10 +17,10 @@ public sealed class UserProjectsUpdatedMessage : EventArgs
public sealed class ProjectCommentsUpdatedMessage : EventArgs
{
[JsonRequired]
public required string id { get; init; }
public string id { get; init; }
[JsonRequired]
public required ProjectCommentsUpdatedMessageType type { get; init; }
public ProjectCommentsUpdatedMessageType type { get; init; }
public Comment? comment { get; init; }
}
@@ -28,10 +28,10 @@ public sealed class ProjectCommentsUpdatedMessage : EventArgs
public sealed class ProjectFileImportUpdatedMessage : EventArgs
{
[JsonRequired]
public required string id { get; init; }
public string id { get; init; }
[JsonRequired]
public required ProjectFileImportUpdatedMessageType type { get; init; }
public ProjectFileImportUpdatedMessageType type { get; init; }
public FileUpload? upload { get; init; }
}
@@ -39,10 +39,10 @@ public sealed class ProjectFileImportUpdatedMessage : EventArgs
public sealed class ProjectModelsUpdatedMessage : EventArgs
{
[JsonRequired]
public required string id { get; init; }
public string id { get; init; }
[JsonRequired]
public required ProjectModelsUpdatedMessageType type { get; init; }
public ProjectModelsUpdatedMessageType type { get; init; }
public Model? model { get; init; }
}
@@ -50,10 +50,10 @@ public sealed class ProjectModelsUpdatedMessage : EventArgs
public sealed class ProjectPendingModelsUpdatedMessage : EventArgs
{
[JsonRequired]
public required string id { get; init; }
public string id { get; init; }
[JsonRequired]
public required ProjectPendingModelsUpdatedMessageType type { get; init; }
public ProjectPendingModelsUpdatedMessageType type { get; init; }
public FileUpload? model { get; init; }
}
@@ -61,10 +61,10 @@ public sealed class ProjectPendingModelsUpdatedMessage : EventArgs
public sealed class ProjectUpdatedMessage : EventArgs
{
[JsonRequired]
public required string id { get; init; }
public string id { get; init; }
[JsonRequired]
public required ProjectUpdatedMessageType type { get; init; }
public ProjectUpdatedMessageType type { get; init; }
public Project? project { get; init; }
}
@@ -72,22 +72,13 @@ public sealed class ProjectUpdatedMessage : EventArgs
public sealed class ProjectVersionsUpdatedMessage : EventArgs
{
[JsonRequired]
public required string id { get; init; }
public string id { get; init; }
[JsonRequired]
public required ProjectVersionsUpdatedMessageType type { get; init; }
public ProjectVersionsUpdatedMessageType type { get; init; }
[JsonRequired]
public required string modelId { get; init; }
public string modelId { get; init; }
public Version? version { get; init; }
}
public sealed class ProjectModelIngestionUpdatedMessage : EventArgs
{
[JsonRequired]
public required ModelIngestion modelIngestion { get; init; }
[JsonRequired]
public required ProjectModelIngestionUpdatedMessageType type { get; init; }
}
@@ -1,486 +0,0 @@
using GraphQL;
using Speckle.Sdk.Api.GraphQL.Inputs;
using Speckle.Sdk.Api.GraphQL.Models;
using Speckle.Sdk.Api.GraphQL.Models.Responses;
namespace Speckle.Sdk.Api.GraphQL.Resources;
/// <remarks>
/// Model Ingestion API is available for server versions <c>3.0.3-alpha.583</c> and above
/// </remarks>
public sealed class ModelIngestionResource
{
private readonly ISpeckleGraphQLClient _client;
internal ModelIngestionResource(ISpeckleGraphQLClient client)
{
_client = client;
}
/// <summary>
/// Create a new model ingestion
/// </summary>
/// <remarks>
/// The model ingestion created will have a <c>processing</c> state (not <c>queued</c>). This mutation is designed to be used
/// by client/connectors that are immediately processing
/// Model Ingestion API is available for server versions <c>3.0.3-alpha.583</c> and above
/// </remarks>
/// <param name="input"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
/// <inheritdoc cref="ISpeckleGraphQLClient.ExecuteGraphQLRequest{T}"/>
public async Task<ModelIngestion> Create(
ModelIngestionCreateInput input,
CancellationToken cancellationToken = default
)
{
//language=graphql
const string QUERY = """
mutation IngestionCreate($input: ModelIngestionCreateInput!) {
data: projectMutations {
data: modelIngestionMutations {
data: create(input: $input) {
id
createdAt
updatedAt
modelId
cancellationRequested
statusData {
... on HasModelIngestionStatus {
status
}
... on HasProgressMessage {
progressMessage
}
}
}
}
}
}
""";
GraphQLRequest request = new() { Query = QUERY, Variables = new { input } };
var res = await _client
.ExecuteGraphQLRequest<RequiredResponse<RequiredResponse<RequiredResponse<ModelIngestion>>>>(
request,
cancellationToken
)
.ConfigureAwait(false);
return res.data.data.data;
}
/// <remarks>
/// Model Ingestion API is available for server versions <c>3.0.3-alpha.583</c> and above
/// </remarks>
/// <param name="modelIngestionId"></param>
/// <param name="projectId"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
/// <inheritdoc cref="ISpeckleGraphQLClient.ExecuteGraphQLRequest{T}"/>
public async Task<ModelIngestion> Get(
string modelIngestionId,
string projectId,
CancellationToken cancellationToken = default
)
{
//language=graphql
const string QUERY = """
query Query($projectId: String!, $modelIngestionId: ID!) {
data:project(id: $projectId) {
data:ingestion(id: $modelIngestionId) {
id
createdAt
updatedAt
modelId
cancellationRequested
statusData {
... on HasModelIngestionStatus {
status
}
... on HasProgressMessage {
progressMessage
}
}
}
}
}
""";
GraphQLRequest request = new() { Query = QUERY, Variables = new { projectId, modelIngestionId } };
var res = await _client
.ExecuteGraphQLRequest<RequiredResponse<RequiredResponse<ModelIngestion>>>(request, cancellationToken)
.ConfigureAwait(false);
return res.data.data;
}
/// <summary>
/// For File Import / Cloud integrations only
/// </summary>
/// <remarks>
/// Model Ingestion API is available for server versions <c>3.0.3-alpha.583</c> and above
/// </remarks>
/// <param name="input"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
/// <inheritdoc cref="ISpeckleGraphQLClient.ExecuteGraphQLRequest{T}"/>
public async Task<ModelIngestion> StartProcessing(
ModelIngestionStartProcessingInput input,
CancellationToken cancellationToken = default
)
{
//language=graphql
const string QUERY = """
mutation IngestionStartProcessing($input: ModelIngestionStartProcessingInput!) {
data: projectMutations {
data: modelIngestionMutations {
data: startProcessing(input: $input) {
id
createdAt
updatedAt
modelId
cancellationRequested
statusData {
... on HasModelIngestionStatus {
status
}
... on HasProgressMessage {
progressMessage
}
}
}
}
}
}
""";
GraphQLRequest request = new() { Query = QUERY, Variables = new { input } };
var res = await _client
.ExecuteGraphQLRequest<RequiredResponse<RequiredResponse<RequiredResponse<ModelIngestion>>>>(
request,
cancellationToken
)
.ConfigureAwait(false);
return res.data.data.data;
}
/// <summary>
/// For File Import / Cloud integrations only
/// </summary>
/// <remarks>
/// Model Ingestion API is available for server versions <c>3.0.3-alpha.583</c> and above
/// </remarks>
/// <param name="input"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
/// <inheritdoc cref="ISpeckleGraphQLClient.ExecuteGraphQLRequest{T}"/>
public async Task<ModelIngestion> Requeue(
ModelIngestionRequeueInput input,
CancellationToken cancellationToken = default
)
{
//language=graphql
const string QUERY = """
mutation IngestionStartProcessing($input: ModelIngestionRequeueInput!) {
data: projectMutations {
data: modelIngestionMutations {
data: requeue(input: $input) {
id
createdAt
updatedAt
modelId
cancellationRequested
statusData {
... on HasModelIngestionStatus {
status
}
... on HasProgressMessage {
progressMessage
}
}
}
}
}
}
""";
GraphQLRequest request = new() { Query = QUERY, Variables = new { input } };
var res = await _client
.ExecuteGraphQLRequest<RequiredResponse<RequiredResponse<RequiredResponse<ModelIngestion>>>>(
request,
cancellationToken
)
.ConfigureAwait(false);
return res.data.data.data;
}
/// <remarks>
/// Model Ingestion API is available for server versions <c>3.0.3-alpha.583</c> and above
/// </remarks>
/// <param name="input"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
/// <inheritdoc cref="ISpeckleGraphQLClient.ExecuteGraphQLRequest{T}"/>
public async Task<ModelIngestion> UpdateProgress(
ModelIngestionUpdateInput input,
CancellationToken cancellationToken = default
)
{
//language=graphql
const string QUERY = """
mutation IngestionUpdateProgress(
$input: ModelIngestionUpdateInput!
) {
data: projectMutations {
data: modelIngestionMutations {
data: updateProgress(input: $input) {
id
createdAt
updatedAt
modelId
cancellationRequested
statusData {
... on HasModelIngestionStatus {
status
}
... on HasProgressMessage {
progressMessage
}
}
}
}
}
}
""";
GraphQLRequest request = new() { Query = QUERY, Variables = new { input } };
var res = await _client
.ExecuteGraphQLRequest<RequiredResponse<RequiredResponse<RequiredResponse<ModelIngestion>>>>(
request,
cancellationToken
)
.ConfigureAwait(false);
return res.data.data.data;
}
/// <summary>
/// Request that the server completes the ingestion by creating a version
/// If successful, the job will be in a terminal "successful" state.
/// </summary>
/// <remarks>
/// Model Ingestion API is available for server versions <c>3.0.3-alpha.583</c> and above
/// </remarks>
/// <seealso cref="FailWithError"/>
/// <seealso cref="FailWithCancel"/>
/// <param name="input"></param>
/// <param name="cancellationToken"></param>
/// <returns>The version id</returns>
/// <inheritdoc cref="ISpeckleGraphQLClient.ExecuteGraphQLRequest{T}"/>
public async Task<string> Complete(ModelIngestionSuccessInput input, CancellationToken cancellationToken = default)
{
//language=graphql
const string QUERY = """
mutation IngestionComplete($input: ModelIngestionSuccessInput!) {
data: projectMutations {
data: modelIngestionMutations {
data: completeWithVersion(input: $input) {
data:statusData {
... on ModelIngestionSuccessStatus {
data:versionId
}
}
}
}
}
}
""";
GraphQLRequest request = new() { Query = QUERY, Variables = new { input } };
var res = await _client
.ExecuteGraphQLRequest<
RequiredResponse<RequiredResponse<RequiredResponse<RequiredResponse<RequiredResponse<string>>>>>
>(request, cancellationToken)
.ConfigureAwait(false);
return res.data.data.data.data.data;
}
/// <summary>
/// Fail the job with an error.
/// </summary>
/// <remarks>
/// For requested user cancellation, use <see cref="FailWithCancel"/> instead<br/>
/// Model Ingestion API is available for server versions <c>3.0.3-alpha.583</c> and above
/// </remarks>
/// <seealso cref="FailWithCancel"/>
/// <seealso cref="Complete"/>
/// <param name="input"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
/// <inheritdoc cref="ISpeckleGraphQLClient.ExecuteGraphQLRequest{T}"/>
public async Task<ModelIngestion> FailWithError(
ModelIngestionFailedInput input,
CancellationToken cancellationToken = default
)
{
//language=graphql
const string QUERY = """
mutation IngestionFailWithError($input: ModelIngestionFailedInput!) {
data: projectMutations {
data: modelIngestionMutations {
data: failWithError(input: $input) {
id
createdAt
updatedAt
modelId
cancellationRequested
statusData {
... on HasModelIngestionStatus {
status
}
... on HasProgressMessage {
progressMessage
}
}
}
}
}
}
""";
GraphQLRequest request = new() { Query = QUERY, Variables = new { input } };
var res = await _client
.ExecuteGraphQLRequest<RequiredResponse<RequiredResponse<RequiredResponse<ModelIngestion>>>>(
request,
cancellationToken
)
.ConfigureAwait(false);
return res.data.data.data;
}
/// <summary>
/// Fail the ingestion with a <c>canceled</c> status.
/// This should only be done if the user has explicitly requested cancellation
/// Other forms of cancellation use <see cref="FailWithError"/>.
/// The ingestion should then enter a terminal "canceled" state.<br/>
/// Model Ingestion API is available for server versions <c>3.0.3-alpha.583</c> and above
/// </summary>
/// <seealso cref="FailWithError"/>
/// <seealso cref="Complete"/>
/// <param name="input"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
/// <inheritdoc cref="ISpeckleGraphQLClient.ExecuteGraphQLRequest{T}"/>
public async Task<ModelIngestion> FailWithCancel(
ModelIngestionCancelledInput input,
CancellationToken cancellationToken = default
)
{
//language=graphql
const string QUERY = """
mutation IngestionFailWithCancel($input: ModelIngestionCancelledInput!) {
data: projectMutations {
data: modelIngestionMutations {
data: failWithCancel(input: $input) {
id
createdAt
updatedAt
modelId
cancellationRequested
statusData {
... on HasModelIngestionStatus {
status
}
... on HasProgressMessage {
progressMessage
}
}
}
}
}
}
""";
GraphQLRequest request = new() { Query = QUERY, Variables = new { input } };
var res = await _client
.ExecuteGraphQLRequest<RequiredResponse<RequiredResponse<RequiredResponse<ModelIngestion>>>>(
request,
cancellationToken
)
.ConfigureAwait(false);
return res.data.data.data;
}
/// <summary>
/// Request that the <see cref="ModelIngestion"/> is canceled.
/// </summary>
/// <remarks>
/// Note simply calling this mutation does not imediatly cancel, it doesn't even guarantee it will be canceled at all.
/// It's up to the client to observe this cancellation request
/// via <see cref="SubscriptionResource.CreateProjectModelIngestionCancellationRequestedSubscription"/>
/// and report it as canceled via <see cref="ModelIngestionResource.FailWithCancel"/>
/// See "cooperative cancellation pattern"<br/>
/// Model Ingestion API is available for server versions <c>3.0.3-alpha.583</c> and above
/// </remarks>
/// <seealso cref="FailWithError"/>
/// <seealso cref="Complete"/>
/// <param name="input"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
/// <inheritdoc cref="ISpeckleGraphQLClient.ExecuteGraphQLRequest{T}"/>
public async Task<ModelIngestion> RequestCancellation(
ModelIngestionCancelledInput input,
CancellationToken cancellationToken = default
)
{
//language=graphql
const string QUERY = """
mutation IngestionRequestCancellation($input: ModelIngestionRequestCancellationInput!) {
data: projectMutations {
data: modelIngestionMutations {
data: requestCancellation (input: $input) {
id
createdAt
updatedAt
modelId
cancellationRequested
statusData {
... on HasModelIngestionStatus {
status
}
... on HasProgressMessage {
progressMessage
}
}
}
}
}
}
""";
GraphQLRequest request = new() { Query = QUERY, Variables = new { input } };
var res = await _client
.ExecuteGraphQLRequest<RequiredResponse<RequiredResponse<RequiredResponse<ModelIngestion>>>>(
request,
cancellationToken
)
.ConfigureAwait(false);
return res.data.data.data;
}
}
@@ -1,5 +1,4 @@
using GraphQL;
using Speckle.Sdk.Api.GraphQL.Enums;
using Speckle.Sdk.Api.GraphQL.Inputs;
using Speckle.Sdk.Api.GraphQL.Models;
using Speckle.Sdk.Api.GraphQL.Models.Responses;
@@ -213,60 +212,6 @@ public sealed class SubscriptionResource : IDisposable
return subscription;
}
/// <summary>Subscribe to a cancellation request being made for a Model Ingestion</summary>
/// <remarks><inheritdoc cref="CreateUserProjectsUpdatedSubscription"/></remarks>
/// <inheritdoc cref="ISpeckleGraphQLClient.SubscribeTo{T}"/>
public Subscription<ProjectModelIngestionUpdatedMessage> CreateProjectModelIngestionUpdatedSubscription(
ProjectModelIngestionSubscriptionInput input
)
{
//language=graphql
const string QUERY = """
subscription IngestionUpdated($input: ProjectModelIngestionSubscriptionInput!) {
data: projectModelIngestionUpdated(input: $input) {
modelIngestion {
id
createdAt
updatedAt
modelId
cancellationRequested
statusData {
... on HasModelIngestionStatus {
status
}
... on HasProgressMessage {
progressMessage
}
}
}
type
}
}
""";
GraphQLRequest request = new() { Query = QUERY, Variables = new { input } };
Subscription<ProjectModelIngestionUpdatedMessage> subscription = new(_client, request);
_subscriptions.Add(subscription);
return subscription;
}
/// <summary>Subscribe to a cancellation request being made for a Model Ingestion</summary>
/// <remarks><inheritdoc cref="CreateUserProjectsUpdatedSubscription"/></remarks>
/// <inheritdoc cref="ISpeckleGraphQLClient.SubscribeTo{T}"/>
public Subscription<ProjectModelIngestionUpdatedMessage> CreateProjectModelIngestionCancellationRequestedSubscription(
string ingestionId,
string projectId
)
{
return CreateProjectModelIngestionUpdatedSubscription(
new ProjectModelIngestionSubscriptionInput(
projectId,
new(ingestionId, null),
ProjectModelIngestionUpdatedMessageType.cancellationRequested
)
);
}
public void Dispose()
{
foreach (var subscription in _subscriptions)
+65 -22
View File
@@ -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 &#x2264; Length &#x2264; 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 &#x2264; Length &#x2264; 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);
}
}
+12 -11
View File
@@ -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;
+27 -14
View File
@@ -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);
@@ -1,8 +0,0 @@
{
"Type": "AggregateException",
"InnerException": {
"Data": {},
"Message": "NOT_FOUND_ERROR: Model ingestion not found",
"Type": "SpeckleGraphQLException"
}
}
@@ -1,8 +0,0 @@
{
"Type": "AggregateException",
"InnerException": {
"Data": {},
"Message": "STREAM_NOT_FOUND: Project not found",
"Type": "SpeckleGraphQLStreamNotFoundException"
}
}
@@ -1,8 +0,0 @@
{
"Type": "AggregateException",
"InnerException": {
"Data": {},
"Message": "NOT_FOUND_ERROR: Model ingestion not found",
"Type": "SpeckleGraphQLException"
}
}
@@ -1,74 +0,0 @@
using System.Reflection;
using Speckle.Sdk.Api;
using Speckle.Sdk.Api.GraphQL.Inputs;
using Speckle.Sdk.Api.GraphQL.Models;
using Speckle.Sdk.Api.GraphQL.Resources;
using Speckle.Sdk.Host;
using Speckle.Sdk.Models;
namespace Speckle.Sdk.Tests.Integration.API.GraphQL.Resources;
[Trait("Server", "Internal")]
public sealed class ModelIngestionResourceExceptionalTests : IAsyncLifetime
{
private IClient _testUser;
private ModelIngestionResource Sut => _testUser.Ingestion;
private Project _project;
private Model _model;
public Task DisposeAsync() => Task.CompletedTask;
public async Task InitializeAsync()
{
TypeLoader.Reset();
TypeLoader.Initialize(typeof(Base).Assembly, Assembly.GetExecutingAssembly());
_testUser = await Fixtures.SeedUserWithClient();
_project = await _testUser.Project.Create(new("Test project", "", null));
_model = await _testUser.Model.Create(new("Test Model 1", "", _project.id));
}
[Fact]
public async Task CreateIngestionNonExistentProject()
{
var createInput = new ModelIngestionCreateInput(
_model.id,
"Doesn't exist...",
"Starting processing",
new(".NET test runner", "0.0.0", null, null)
);
var ex = await Assert.ThrowsAsync<AggregateException>(async () =>
{
_ = await Sut.Create(createInput);
});
await Verify(ex);
}
[Fact]
public async Task UpdateNonExistentNonExistent()
{
var updateInput = new ModelIngestionUpdateInput("Doesn't exist", _project.id, "Can't be", 0.5);
var ex = await Assert.ThrowsAsync<AggregateException>(async () =>
{
_ = await Sut.UpdateProgress(updateInput);
});
await Verify(ex);
}
[Fact]
public async Task CancelNonExistentIngestion()
{
var input = new ModelIngestionCancelledInput(
"Non-existent-ingestion",
_project.id,
cancellationMessage: "This was cancelled for testing purposes"
);
var ex = await Assert.ThrowsAsync<AggregateException>(async () =>
{
_ = await Sut.FailWithCancel(input);
});
await Verify(ex);
}
}
@@ -1,193 +0,0 @@
using System.Reflection;
using Microsoft.Extensions.DependencyInjection;
using Speckle.Sdk.Api;
using Speckle.Sdk.Api.GraphQL.Enums;
using Speckle.Sdk.Api.GraphQL.Inputs;
using Speckle.Sdk.Api.GraphQL.Models;
using Speckle.Sdk.Api.GraphQL.Resources;
using Speckle.Sdk.Host;
using Speckle.Sdk.Models;
using Speckle.Sdk.Transports;
using Version = Speckle.Sdk.Api.GraphQL.Models.Version;
namespace Speckle.Sdk.Tests.Integration.API.GraphQL.Resources;
[Trait("Server", "Internal")]
public sealed class ModelIngestionResourceTests : IAsyncLifetime
{
private IClient _testUser;
private ModelIngestionResource Sut => _testUser.Ingestion;
private Project _project;
private Model _model;
private IOperations _operations;
public Task DisposeAsync() => Task.CompletedTask;
public async Task InitializeAsync()
{
TypeLoader.Reset();
TypeLoader.Initialize(typeof(Base).Assembly, Assembly.GetExecutingAssembly());
var serviceProvider = TestServiceSetup.GetServiceProvider();
_operations = serviceProvider.GetRequiredService<IOperations>();
_testUser = await Fixtures.SeedUserWithClient();
_project = await _testUser.Project.Create(new("Test project", "", null));
_model = await _testUser.Model.Create(new("Test Model 1", "", _project.id));
}
[Fact]
public async Task CreateAndError()
{
var createInput = new ModelIngestionCreateInput(
_model.id,
_project.id,
"Starting processing",
new(".NET test runner", "0.0.0", null, null)
);
ModelIngestion ingest = await Sut.Create(createInput);
var errorInput = new ModelIngestionFailedInput(ingest.id, _project.id, "A bad thing happened", "Over hear!");
var res = await Sut.FailWithError(errorInput);
Assert.Equal(ingest.id, res.id);
}
[Fact]
public async Task CreateAndUpdate()
{
var createInput = new ModelIngestionCreateInput(
_model.id,
_project.id,
"Starting processing",
new(".NET test runner", "0.0.0", null, null)
);
ModelIngestion ingest = await Sut.Create(createInput);
await Update(null, "None");
await Update(0.1, "0.1");
await Update(0.5, "Whoa-oh! We're half way there!");
await Update(1, "Finished");
await Update(0.2, "Back to processing again");
async Task Update(double? progress, string message)
{
var updateInput = new ModelIngestionUpdateInput(ingest.id, _project.id, message, progress);
var res = await Sut.UpdateProgress(updateInput);
Assert.Equal(message, res.statusData.progressMessage);
Assert.False(res.cancellationRequested);
Assert.Equal(ModelIngestionStatus.processing, res.statusData.status);
}
}
[Fact]
public async Task CreateAndCancel()
{
var createInput = new ModelIngestionCreateInput(
_model.id,
_project.id,
"Starting processing",
new(".NET test runner", "0.0.0", null, null)
);
ModelIngestion ingest = await Sut.Create(createInput);
var input = new ModelIngestionCancelledInput(
ingest.id,
_project.id,
cancellationMessage: "This was cancelled for testing purposes"
);
var res = await Sut.FailWithCancel(input);
Assert.Equal(ingest.id, res.id);
}
[Fact]
public async Task CreateAndComplete()
{
ModelIngestionCreateInput createInput = new(
_model.id,
_project.id,
"Starting processing",
new(".NET test runner", "0.0.0", null, null)
);
ModelIngestion ingest = await Sut.Create(createInput);
Base myObject = Fixtures.GenerateNestedObject();
var sendResult = await _operations.Send2(
_testUser.ServerUrl,
_project.id,
_testUser.Account.token,
myObject,
new Progress<ProgressArgs>(x =>
{
var updateInput = new ModelIngestionUpdateInput(
ingest.id,
_project.id,
$"{x.Count} / {x.Total}",
x.Total == null ? null : x.Count / x.Total
);
_ = Sut.UpdateProgress(updateInput).Result;
}),
CancellationToken.None,
new(true, true)
);
ModelIngestionSuccessInput finish = new(ingest.id, _project.id, sendResult.RootId);
string versionId = await Sut.Complete(finish);
Version version = await _testUser.Version.Get(versionId, _project.id);
Assert.Equal(version.id, versionId);
Assert.Equal(sendResult.RootId, version.referencedObject);
}
[Fact]
public async Task CreateAndGet()
{
var createInput = new ModelIngestionCreateInput(
_model.id,
_project.id,
"Starting processing",
new(".NET test runner", "0.0.0", null, null)
);
ModelIngestion ingest = await Sut.Create(createInput);
ModelIngestion res = await Sut.Get(ingest.id, _project.id);
Assert.Equal(ingest.id, res.id);
Assert.Equal(ingest.statusData.status, res.statusData.status);
}
[Fact]
public async Task TestRequeue()
{
//Not sure if is desirable that ingestions created by the modelIngestionMutations.create mutation can be re-queued
//But the server allows it, so we test it
var createInput = new ModelIngestionCreateInput(
_model.id,
_project.id,
"Starting processing",
new(".NET test runner", "0.0.0", null, null)
);
var ingestion = await Sut.Create(createInput);
var res = await Sut.Requeue(new(ingestion.id, _project.id, "we'll try and requeue this ingestion"));
Assert.Equal(ingestion.id, res.id);
Assert.Equal(ModelIngestionStatus.queued, res.statusData.status);
}
[Fact]
public async Task TestStartProcessing()
{
//Not sure if is desirable that StartProcessing can be used by ingestions created by the modelIngestionMutations.create mutation
//But the server allows it, so we test it
var createInput = new ModelIngestionCreateInput(
_model.id,
_project.id,
"Starting processing",
new(".NET test runner", "0.0.0", null, null)
);
var ingestion = await Sut.Create(createInput);
var res = await Sut.StartProcessing(
new(ingestion.id, _project.id, "", new SourceDataInput("what", "happens", "now", 0))
);
Assert.Equal(ingestion.id, res.id);
Assert.Equal(ModelIngestionStatus.processing, res.statusData.status);
}
}
@@ -1,4 +1,4 @@
using FluentAssertions;
using FluentAssertions;
using Speckle.Sdk.Api;
using Speckle.Sdk.Api.GraphQL.Enums;
using Speckle.Sdk.Api.GraphQL.Inputs;
@@ -11,7 +11,7 @@ namespace Speckle.Sdk.Tests.Integration.API.GraphQL.Resources;
public class SubscriptionResourceTests : IAsyncLifetime
{
#if DEBUG
private const int WAIT_PERIOD = 4000; // WSL is slow AF, so for local runs, we're being extra generous
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
@@ -80,15 +80,15 @@ public class SubscriptionResourceTests : IAsyncLifetime
public async Task ProjectUpdated_SubscriptionIsCalled()
{
TaskCompletionSource<ProjectUpdatedMessage> tcs = new();
using Subscription<ProjectUpdatedMessage> sub = Sut.CreateProjectUpdatedSubscription(_testProject.id);
using var sub = Sut.CreateProjectUpdatedSubscription(_testProject.id);
sub.Listeners += (_, message) => tcs.SetResult(message);
await Task.Delay(WAIT_PERIOD); // Give time to subscription to be setup
ProjectUpdateInput input = new(_testProject.id, "This is my new name");
Project created = await _testUser.Project.Update(input);
var input = new ProjectUpdateInput(_testProject.id, "This is my new name");
var created = await _testUser.Project.Update(input);
ProjectUpdatedMessage subscriptionMessage = await tcs.Task;
var subscriptionMessage = await tcs.Task;
subscriptionMessage.Should().NotBeNull();
subscriptionMessage.id.Should().Be(created.id);
@@ -135,74 +135,4 @@ public class SubscriptionResourceTests : IAsyncLifetime
subscriptionMessage.type.Should().Be(ProjectCommentsUpdatedMessageType.CREATED);
subscriptionMessage.comment.Should().NotBeNull();
}
[Fact(Timeout = TIMEOUT), Trait("Server", "Internal")]
public async Task ProjectModelIngestionCancellationRequested_SubscriptionIsCalled()
{
ModelIngestion ingestion = await _testUser.Ingestion.Create(
new(_testModel.id, _testProject.id, "", new(".NET test", "0.0.0", null, null))
);
TaskCompletionSource<ProjectModelIngestionUpdatedMessage> tcs = new();
using var sub = Sut.CreateProjectModelIngestionCancellationRequestedSubscription(ingestion.id, _testProject.id);
sub.Listeners += (_, message) => tcs.SetResult(message);
await Task.Delay(WAIT_PERIOD); // Give time to subscription to be setup
await _testUser.Ingestion.RequestCancellation(new(ingestion.id, _testProject.id, "please cancel"));
var subscriptionMessage = await tcs.Task;
subscriptionMessage.Should().NotBeNull();
subscriptionMessage.type.Should().Be(ProjectModelIngestionUpdatedMessageType.cancellationRequested);
subscriptionMessage.modelIngestion.id.Should().Be(ingestion.id);
}
[Fact(Timeout = TIMEOUT), Trait("Server", "Internal")]
public async Task ProjectModelIngestionUpdate_UpdateSubscriptionIs()
{
ModelIngestion ingestion = await _testUser.Ingestion.Create(
new(_testModel.id, _testProject.id, "", new(".NET test", "0.0.0", null, null))
);
TaskCompletionSource<ProjectModelIngestionUpdatedMessage> tcs = new();
using var sub = Sut.CreateProjectModelIngestionUpdatedSubscription(
new(
_testProject.id,
new ModelIngestionReference(ingestion.id, null),
ProjectModelIngestionUpdatedMessageType.updated
)
);
sub.Listeners += (_, message) => tcs.SetResult(message);
await Task.Delay(WAIT_PERIOD); // Give time to subscription to be setup
await _testUser.Ingestion.UpdateProgress(new(ingestion.id, _testProject.id, "Here's an update", 0.314));
var subscriptionMessage = await tcs.Task;
subscriptionMessage.Should().NotBeNull();
subscriptionMessage.type.Should().Be(ProjectModelIngestionUpdatedMessageType.updated);
subscriptionMessage.modelIngestion.id.Should().Be(ingestion.id);
}
[Fact(Timeout = TIMEOUT), Trait("Server", "Internal")]
public async Task ProjectModelIngestionUpdate_CancelSubscriptionIsNotCalled()
{
ModelIngestion ingestion = await _testUser.Ingestion.Create(
new(_testModel.id, _testProject.id, "", new(".NET test", "0.0.0", null, null))
);
TaskCompletionSource<ProjectModelIngestionUpdatedMessage> tcs = new();
using var sub = Sut.CreateProjectModelIngestionCancellationRequestedSubscription(ingestion.id, _testProject.id);
sub.Listeners += (_, message) => tcs.SetResult(message);
await Task.Delay(WAIT_PERIOD); // Give time to subscription to be setup
await _testUser.Ingestion.UpdateProgress(new(ingestion.id, _testProject.id, "this shouldn't cancel", null));
await Task.Delay(WAIT_PERIOD); // Give time to subscription to maybe fire
tcs.Task.IsCompleted.Should().BeFalse();
}
}
@@ -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);
}
}