frostfs-sdk-csharp/src/FrostFS.SDK.ClientV2/Services/ContainerServiceProvider.cs
Pavel Gross d7dbbf8da8 [#24] Client: Add Ape manager
Signed-off-by: Pavel Gross <p.gross@yadro.com>
2024-09-23 10:20:40 +03:00

204 lines
5.9 KiB
C#

using System;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Threading.Tasks;
using FrostFS.SDK.ClientV2;
using FrostFS.Container;
using FrostFS.SDK.ClientV2.Mappers.GRPC;
using FrostFS.SDK.Cryptography;
using FrostFS.Refs;
using FrostFS.Session;
namespace FrostFS.SDK.ClientV2;
internal class ContainerServiceProvider(ContainerService.ContainerServiceClient service, ClientEnvironment context) : ContextAccessor(context), ISessionProvider
{
readonly SessionProvider sessions = new(context);
public async ValueTask<SessionToken> GetOrCreateSession(ISessionToken args, Context ctx)
{
return await sessions.GetOrCreateSession(args, ctx);
}
internal async Task<FrostFsContainerInfo> GetContainerAsync(PrmContainerGet args)
{
GetRequest request = GetContainerRequest(args.Container.ContainerID, args.XHeaders, args.Context!);
var response = await service.GetAsync(request, null, args.Context!.Deadline, args.Context.CancellationToken);
Verifier.CheckResponse(response);
return response.Body.Container.ToModel();
}
internal async IAsyncEnumerable<FrostFsContainerId> ListContainersAsync(PrmContainerGetAll args)
{
var ctx = args.Context!;
var request = new ListRequest
{
Body = new ()
{
OwnerId = ctx.OwnerId.ToMessage()
}
};
request.AddMetaHeader(args.XHeaders);
request.Sign(ctx.Key);
var response = await service.ListAsync(request, null, ctx.Deadline, ctx.CancellationToken);
Verifier.CheckResponse(response);
foreach (var cid in response.Body.ContainerIds)
{
yield return new FrostFsContainerId(Base58.Encode(cid.Value.ToByteArray()));
}
}
internal async Task<FrostFsContainerId> CreateContainerAsync(PrmContainerCreate args)
{
var ctx = args.Context!;
var grpcContainer = args.Container.GetContainer();
grpcContainer.OwnerId ??= ctx.OwnerId.ToMessage();
grpcContainer.Version ??= ctx.Version.ToMessage();
var request = new PutRequest
{
Body = new PutRequest.Types.Body
{
Container = grpcContainer,
Signature = ctx.Key.SignRFC6979(grpcContainer)
}
};
var sessionToken = await GetOrCreateSession(args, ctx);
sessionToken.CreateContainerTokenContext(
null,
ContainerSessionContext.Types.Verb.Put,
ctx.Key,
ctx.GetPublicKeyCache());
request.AddMetaHeader(args.XHeaders, sessionToken);
request.Sign(ctx.Key);
var response = await service.PutAsync(request, null, ctx.Deadline, ctx.CancellationToken);
Verifier.CheckResponse(response);
await WaitForContainer(WaitExpects.Exists, response.Body.ContainerId, args.WaitParams, ctx);
return new FrostFsContainerId(response.Body.ContainerId);
}
internal async Task DeleteContainerAsync(PrmContainerDelete args)
{
var ctx = args.Context!;
var request = new DeleteRequest
{
Body = new DeleteRequest.Types.Body
{
ContainerId = args.ContainerId.ToMessage(),
Signature = ctx.Key.SignRFC6979(args.ContainerId.ToMessage().Value)
}
};
var sessionToken = await GetOrCreateSession(args, ctx);
sessionToken.CreateContainerTokenContext(
request.Body.ContainerId,
ContainerSessionContext.Types.Verb.Delete,
ctx.Key,
ctx.GetPublicKeyCache());
request.AddMetaHeader(args.XHeaders, sessionToken);
request.Sign(ctx.Key);
var response = await service.DeleteAsync(request, null, ctx.Deadline, ctx.CancellationToken);
Verifier.CheckResponse(response);
await WaitForContainer(WaitExpects.Removed, request.Body.ContainerId, args.WaitParams, ctx);
Verifier.CheckResponse(response);
}
private static GetRequest GetContainerRequest(ContainerID id, NameValueCollection? xHeaders, Context ctx)
{
var request = new GetRequest
{
Body = new GetRequest.Types.Body
{
ContainerId = id
}
};
request.AddMetaHeader(xHeaders);
request.Sign(ctx.Key);
return request;
}
private enum WaitExpects
{
Exists,
Removed
}
private async Task WaitForContainer(WaitExpects expect, ContainerID id, PrmWait? waitParams, Context ctx)
{
var request = GetContainerRequest(id, null, ctx);
async Task action()
{
var response = await service.GetAsync(request, null, ctx.Deadline, ctx.CancellationToken);
Verifier.CheckResponse(response);
}
await WaitFor(action, expect, waitParams);
}
private static async Task WaitFor(
Func<Task> action,
WaitExpects expect,
PrmWait? waitParams)
{
waitParams ??= PrmWait.DefaultParams;
var deadLine = waitParams.GetDeadline();
while (true)
{
try
{
await action();
if (expect == WaitExpects.Exists)
return;
if (DateTime.UtcNow >= deadLine)
throw new TimeoutException();
await Task.Delay(waitParams.PollInterval);
}
catch (ResponseException ex)
{
if (DateTime.UtcNow >= deadLine)
throw new TimeoutException();
if (ex.Status.Code != FrostFsStatusCode.ContainerNotFound)
throw;
if (expect == WaitExpects.Removed)
return;
await Task.Delay(waitParams.PollInterval);
}
}
}
}