frostfs-sdk-csharp/src/FrostFS.SDK.ClientV2/Services/ContainerServiceProvider.cs
Pavel Gross 1a02ac2ae7
All checks were successful
DCO / DCO (pull_request) Successful in 26s
[#22] Client: Container session
Signed-off-by: Pavel Gross <p.gross@yando.com>
2024-08-19 14:18:11 +03:00

206 lines
6.1 KiB
C#

using System;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Threading.Tasks;
using FrostFS.SDK.ClientV2.Mappers.GRPC.Netmap;
using FrostFS.Container;
using FrostFS.SDK.ClientV2.Mappers.GRPC;
using FrostFS.SDK.Cryptography;
using FrostFS.SDK.ModelsV2;
using FrostFS.SDK.ClientV2.Parameters;
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<Session.SessionToken> GetOrCreateSession(ISessionToken args, Context ctx)
{
return await sessions.GetOrCreateSession(args, ctx);
}
internal async Task<ModelsV2.Container> GetContainerAsync(PrmContainerGet args)
{
GetRequest request = GetContainerRequest(args.ContainerId.ToMessage(), 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<ContainerId> ListContainersAsync(PrmContainerGetAll args)
{
var ctx = args.Context!;
var request = new ListRequest
{
Body = new ListRequest.Types.Body
{
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 ContainerId(Base58.Encode(cid.Value.ToByteArray()));
}
}
internal async Task<ContainerId> CreateContainerAsync(PrmContainerCreate args)
{
var ctx = args.Context!;
var grpcContainer = args.Container.ToMessage();
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.PublicKeyCache);
var v = sessionToken.Body.OwnerId == grpcContainer.OwnerId;
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 ContainerId(Base58.Encode(response.Body.ContainerId.Value.ToByteArray()));
}
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.PublicKeyCache);
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 != ModelsV2.Enums.StatusCode.ContainerNotFound)
throw;
if (expect == WaitExpects.Removed)
return;
await Task.Delay(waitParams.PollInterval);
}
}
}
}