[#1] Add object Search operation

Signed-off-by: Ivan Pchelintsev <i.pchelintsev@yadro.com>
This commit is contained in:
Ivan Pchelintsev 2024-05-17 11:52:38 +03:00
parent 63f91ac627
commit bb55d093fa
5 changed files with 156 additions and 4 deletions

View file

@ -11,5 +11,7 @@ public interface IFrostFSClient
Task<ObjectHeader> GetObjectHeadAsync(ContainerId containerId, ObjectId objectId);
Task<ModelsV2.Object> GetObjectAsync(ContainerId containerId, ObjectId objectId);
Task<ObjectId> PutObjectAsync(ObjectHeader header, Stream payload);
Task<ObjectId> PutObjectAsync(ObjectHeader header, byte[] payload);
Task DeleteObjectAsync(ContainerId containerId, ObjectId objectId);
Task<ObjectId[]> SearchObjectAsync(ContainerId cid, params ObjectFilter[] filters);
}

View file

@ -1,5 +1,7 @@
using FrostFS.Object;
using FrostFS.SDK.ModelsV2;
using MatchType = FrostFS.Object.MatchType;
using ObjectType = FrostFS.Object.ObjectType;
namespace FrostFS.SDK.ClientV2.Mappers.GRPC;
@ -20,7 +22,26 @@ public static class ObjectAttributeMapper
}
}
public static class ObjectHeadMapper
public static class ObjectFilterMapper
{
public static SearchRequest.Types.Body.Types.Filter ToGrpcMessage(this ObjectFilter filter)
{
var objMatchTypeName = Enum.GetName(typeof(MatchType), filter.MatchType);
if (objMatchTypeName is null)
{
throw new ArgumentException($"Unknown MatchType. Value: '{filter.MatchType}'.");
}
return new SearchRequest.Types.Body.Types.Filter
{
MatchType = Enum.Parse<MatchType>(objMatchTypeName),
Key = filter.Key,
Value = filter.Value
};
}
}
public static class ObjectHeaderMapper
{
public static Header ToGrpcMessage(this ObjectHeader header)
{

View file

@ -72,17 +72,18 @@ public partial class Client
offset += chunk.Length;
chunk = await stream.ReadChunk();
}
obj.Payload = ByteString.CopyFrom(payload);
return obj;
}
private ObjectReader GetObjectInit(GetRequest initRequest)
{
if (initRequest is null)
{
throw new ArgumentNullException(nameof(initRequest));
}
return new ObjectReader
{
Call = _objectServiceClient.Get(initRequest)
@ -90,6 +91,16 @@ public partial class Client
}
public async Task<ObjectId> PutObjectAsync(ObjectHeader header, Stream payload)
{
return await PutObject(header, payload);
}
public async Task<ObjectId> PutObjectAsync(ObjectHeader header, byte[] payload)
{
return await PutObject(header, new MemoryStream(payload));
}
private async Task<ObjectId> PutObject(ObjectHeader header, Stream payload)
{
var sessionToken = await CreateSessionAsync(uint.MaxValue);
var hdr = header.ToGrpcMessage();
@ -167,6 +178,56 @@ public partial class Client
request.Sign(_key);
await _objectServiceClient.DeleteAsync(request);
}
public async Task<ObjectId[]> SearchObjectAsync(ContainerId cid, params ObjectFilter[] filters)
{
var request = new SearchRequest
{
Body = new SearchRequest.Types.Body
{
ContainerId = cid.ToGrpcMessage(),
Filters = { },
Version = 1
}
};
foreach (var filter in filters)
{
request.Body.Filters.Add(filter.ToGrpcMessage());
}
;
request.AddMetaHeader();
request.Sign(_key);
var ids = await SearchObject(request);
return ids.Select(oid => ObjectId.FromHash(oid.Value.ToByteArray())).ToArray();
}
private async Task<List<ObjectID>> SearchObject(SearchRequest request)
{
var objectsIds = new List<ObjectID> { };
using var stream = SearchObjectInit(request);
var ids = await stream.Read();
while (ids is not null)
{
objectsIds.AddRange(ids);
ids = await stream.Read();
}
return objectsIds;
}
private SearchReader SearchObjectInit(SearchRequest initRequest)
{
if (initRequest is null)
{
throw new ArgumentNullException(nameof(initRequest));
}
return new SearchReader
{
Call = _objectServiceClient.Search(initRequest)
};
}
}
internal class ObjectReader : IDisposable
@ -179,6 +240,7 @@ internal class ObjectReader : IDisposable
{
throw new InvalidOperationException("unexpect end of stream");
}
var response = Call.ResponseStream.Current;
if (response.Body.ObjectPartCase != GetResponse.Types.Body.ObjectPartOneofCase.Init)
throw new InvalidOperationException("unexpect message type");
@ -195,6 +257,7 @@ internal class ObjectReader : IDisposable
{
return null;
}
var response = Call.ResponseStream.Current;
if (response.Body.ObjectPartCase != GetResponse.Types.Body.ObjectPartOneofCase.Chunk)
throw new InvalidOperationException("unexpect message type");
@ -226,9 +289,30 @@ internal class ObjectStreamer : IDisposable
await Call.RequestStream.CompleteAsync();
return await Call.ResponseAsync;
}
public void Dispose()
{
Call.Dispose();
}
}
internal class SearchReader : IDisposable
{
public AsyncServerStreamingCall<SearchResponse> Call { get; init; }
public async Task<List<ObjectID>?> Read()
{
if (!await Call.ResponseStream.MoveNext())
{
return null;
}
var response = Call.ResponseStream.Current;
return response.Body?.IdList.ToList();
}
public void Dispose()
{
Call.Dispose();
}
}

View file

@ -0,0 +1,10 @@
namespace FrostFS.SDK.ModelsV2.Enums;
public enum ObjectMatchType
{
Unspecified = 0,
Equals = 1,
NotEquals = 2,
KeyAbsent = 3,
StartsWith = 4
}

View file

@ -14,6 +14,41 @@ public class ObjectAttribute
}
}
public class ObjectFilter
{
private const string HeaderPrefix = "$Object:";
public ObjectMatchType MatchType { get; set; }
public string Key { get; set; }
public string Value { get; set; }
public ObjectFilter(ObjectMatchType matchType, string key, string value)
{
MatchType = matchType;
Key = key;
Value = value;
}
public static ObjectFilter ObjectIdFilter(ObjectMatchType matchType, ObjectId objectId)
{
return new ObjectFilter(matchType, HeaderPrefix + "objectID", objectId.Value);
}
public static ObjectFilter OwnerFilter(ObjectMatchType matchType, OwnerId ownerId)
{
return new ObjectFilter(matchType, HeaderPrefix + "ownerID", ownerId.Value);
}
public static ObjectFilter RootFilter()
{
return new ObjectFilter(ObjectMatchType.Unspecified, HeaderPrefix + "ROOT", "");
}
public static ObjectFilter VersionFilter(ObjectMatchType matchType, Version version)
{
return new ObjectFilter(matchType, HeaderPrefix + "version", version.ToString());
}
}
public class ObjectHeader
{
public ObjectAttribute[] Attributes { get; set; }