KurlesHS/frostfs-sdk-csharp
1
0
Fork 0
forked from TrueCloudLab/frostfs-sdk-csharp
Code Pull requests Activity

[#31] Client: fix for session

Browse source 
Signed-off-by: Pavel Gross <p.gross@yadro.com>
This commit is contained in:
Pavel Gross 2025-02-17 16:01:41 +03:00
parent 195854a45b
commit 2e56c13946
10 changed files with 247 additions and 146 deletions
Download patch file Download diff file Expand all files Collapse all files

84
src/FrostFS.SDK.Client/Services/ObjectServiceProvider.cs
View file

@ -10,7 +10,6 @@ using FrostFS.Refs;
using FrostFS.SDK.Client;
using FrostFS.SDK.Client.Interfaces;
using FrostFS.SDK.Client.Mappers.GRPC;
using FrostFS.SDK.Cryptography;
using FrostFS.Session;
using Google.Protobuf;
@ -278,7 +277,7 @@ internal sealed class ObjectServiceProvider(ObjectService.ObjectServiceClient cl
var sessionToken = args.SessionToken ?? await GetDefaultSession(args, ctx).ConfigureAwait(false);
var protoToken = sessionToken.CreateObjectTokenContext(
new Address { ContainerId = grpcObject.Header.ContainerId, ObjectId = grpcObject.ObjectId },
new Address { ContainerId = grpcObject.Header.ContainerId },
ObjectSessionContext.Types.Verb.Put,
ClientContext.Key);
@ -297,7 +296,7 @@ internal sealed class ObjectServiceProvider(ObjectService.ObjectServiceClient cl
{
var chunkSize = args.MaxChunkLength;
Stream payload = args.Payload ?? throw new ArgumentNullException(nameof(args), "Stream parameter is null");
var call = client.Patch(null, ctx.GetDeadline(), ctx.CancellationToken);
byte[]? chunkBuffer = null;
@ -306,31 +305,15 @@ internal sealed class ObjectServiceProvider(ObjectService.ObjectServiceClient cl
// common
chunkBuffer = ArrayPool<byte>.Shared.Rent(chunkSize);
bool isFirstChunk = true;
ulong currentPos = args.Range.Offset;
var address = new Address
{
ObjectId = args.Address.ObjectId,
ContainerId = args.Address.ContainerId
};
var sessionToken = args.SessionToken ?? await GetDefaultSession(args, ctx).ConfigureAwait(false);
var protoToken = sessionToken.CreateObjectTokenContext(
address,
ObjectSessionContext.Types.Verb.Patch,
ClientContext.Key);
var request = new PatchRequest()
{
Body = new()
{
Address = address,
ReplaceAttributes = args.ReplaceAttributes,
}
};
bool isFirstChunk = true;
ulong currentPos = args.Range.Offset;
while (true)
{
var bytesCount = await payload.ReadAsync(chunkBuffer, 0, chunkSize, ctx.CancellationToken).ConfigureAwait(false);
@ -340,41 +323,62 @@ internal sealed class ObjectServiceProvider(ObjectService.ObjectServiceClient cl
break;
}
if (isFirstChunk && args.NewAttributes != null && args.NewAttributes.Length > 0)
var request = new PatchRequest()
{
foreach (var attr in args.NewAttributes)
Body = new()
{
request.Body.NewAttributes.Add(attr.ToMessage());
Address = address,
Patch = new PatchRequest.Types.Body.Types.Patch
{
Chunk = ByteString.CopyFrom(chunkBuffer, 0, bytesCount),
SourceRange = new Object.Range { Offset = currentPos, Length = (ulong)bytesCount }
}
}
}
request.Body.Patch = new PatchRequest.Types.Body.Types.Patch
{
Chunk = ByteString.CopyFrom(chunkBuffer, 0, bytesCount),
SourceRange = new Object.Range { Offset = currentPos, Length = (ulong)bytesCount }
};
currentPos += (ulong)bytesCount;
if (isFirstChunk)
{
var sessionToken = args.SessionToken ?? await GetDefaultSession(args, ctx).ConfigureAwait(false);
request.AddMetaHeader(args.XHeaders, protoToken);
var protoToken = sessionToken.CreateObjectTokenContext(
address,
ObjectSessionContext.Types.Verb.Patch,
ClientContext.Key);
request.AddMetaHeader(args.XHeaders, protoToken);
if (args.NewAttributes != null && args.NewAttributes.Length > 0)
{
foreach (var attr in args.NewAttributes)
{
request.Body.NewAttributes.Add(attr.ToMessage());
request.Body.ReplaceAttributes = args.ReplaceAttributes;
}
}
isFirstChunk = false;
}
else
{
request.AddMetaHeader(args.XHeaders);
}
request.Sign(ClientContext.Key.ECDsaKey);
await call.RequestStream.WriteAsync(request).ConfigureAwait(false);
isFirstChunk = false;
currentPos += (ulong)bytesCount;
}
}
finally
{
await call.RequestStream.CompleteAsync().ConfigureAwait(false);
if (chunkBuffer != null)
{
ArrayPool<byte>.Shared.Return(chunkBuffer);
}
}
await call.RequestStream.CompleteAsync().ConfigureAwait(false);
var response = await call.ResponseAsync.ConfigureAwait(false);
Verifier.CheckResponse(response);
@ -513,7 +517,7 @@ internal sealed class ObjectServiceProvider(ObjectService.ObjectServiceClient cl
{
// send chunks limited to default or user's settings
var bufferSize = objectLimitSize > 0 ?
(int)Math.Min(objectLimitSize - sentBytes, chunkSize)
Math.Min(objectLimitSize - sentBytes, chunkSize)
: chunkSize;
var bytesCount = await payload.ReadAsync(chunkBuffer, 0, bufferSize, ctx.CancellationToken).ConfigureAwait(false);
@ -531,6 +535,8 @@ internal sealed class ObjectServiceProvider(ObjectService.ObjectServiceClient cl
}
};
chunkRequest.AddMetaHeader(args.XHeaders);
chunkRequest.Sign(ClientContext.Key.ECDsaKey);
await stream.Write(chunkRequest).ConfigureAwait(false);
@ -571,8 +577,6 @@ internal sealed class ObjectServiceProvider(ObjectService.ObjectServiceClient cl
ObjectTools.SetSplitValues(grpcHeader, header.Split, ClientContext.Owner, ClientContext.Version, ClientContext.Key);
}
var oid = new ObjectID { Value = grpcHeader.Sha256() };
var initRequest = new PutRequest
{
Body = new PutRequest.Types.Body
@ -587,7 +591,7 @@ internal sealed class ObjectServiceProvider(ObjectService.ObjectServiceClient cl
var sessionToken = args.SessionToken ?? await GetDefaultSession(args, ctx).ConfigureAwait(false);
var protoToken = sessionToken.CreateObjectTokenContext(
new Address { ContainerId = grpcHeader.ContainerId, ObjectId = oid },
new Address { ContainerId = grpcHeader.ContainerId },
ObjectSessionContext.Types.Verb.Put,
ClientContext.Key);