[#165] Support latest only stream listing
Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>
This commit is contained in:
parent
c7ee628ab0
commit
739a6ec9df
3 changed files with 74 additions and 5 deletions
|
@ -210,11 +210,12 @@ func TestGetObjectEnabledMD5(t *testing.T) {
|
||||||
require.Equal(t, data.Quote(objInfo.MD5Sum), headers.Get(api.ETag))
|
require.Equal(t, data.Quote(objInfo.MD5Sum), headers.Get(api.ETag))
|
||||||
}
|
}
|
||||||
|
|
||||||
func putObjectContent(hc *handlerContext, bktName, objName, content string) {
|
func putObjectContent(hc *handlerContext, bktName, objName, content string) http.Header {
|
||||||
body := bytes.NewReader([]byte(content))
|
body := bytes.NewReader([]byte(content))
|
||||||
w, r := prepareTestPayloadRequest(hc, bktName, objName, body)
|
w, r := prepareTestPayloadRequest(hc, bktName, objName, body)
|
||||||
hc.Handler().PutObjectHandler(w, r)
|
hc.Handler().PutObjectHandler(w, r)
|
||||||
assertStatus(hc.t, w, http.StatusOK)
|
assertStatus(hc.t, w, http.StatusOK)
|
||||||
|
return w.Result().Header
|
||||||
}
|
}
|
||||||
|
|
||||||
func getObjectRange(t *testing.T, tc *handlerContext, bktName, objName string, start, end int) []byte {
|
func getObjectRange(t *testing.T, tc *handlerContext, bktName, objName string, start, end int) []byte {
|
||||||
|
|
|
@ -8,6 +8,7 @@ import (
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/data"
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/data"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer"
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer/encryption"
|
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer/encryption"
|
||||||
|
@ -60,6 +61,42 @@ func TestListObjectNullVersions(t *testing.T) {
|
||||||
require.Equal(t, data.UnversionedObjectVersionID, result.Version[1].VersionID)
|
require.Equal(t, data.UnversionedObjectVersionID, result.Version[1].VersionID)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestListObjectsLatestVersions(t *testing.T) {
|
||||||
|
hc := prepareHandlerContext(t)
|
||||||
|
|
||||||
|
bktName := "bucket-versioning-enabled"
|
||||||
|
createTestBucket(hc, bktName)
|
||||||
|
putBucketVersioning(t, hc, bktName, true)
|
||||||
|
|
||||||
|
objName1, objName2 := "object1", "object2"
|
||||||
|
objContent1, objContent2 := "content1", "content2"
|
||||||
|
|
||||||
|
putObjectContent(hc, bktName, objName1, objContent1)
|
||||||
|
hdr1 := putObjectContent(hc, bktName, objName1, objContent2)
|
||||||
|
putObjectContent(hc, bktName, objName2, objContent1)
|
||||||
|
hdr2 := putObjectContent(hc, bktName, objName2, objContent2)
|
||||||
|
|
||||||
|
t.Run("listv1", func(t *testing.T) {
|
||||||
|
result := listObjectsV1(hc, bktName, "", "", "", -1)
|
||||||
|
|
||||||
|
require.Len(t, result.Contents, 2)
|
||||||
|
require.Equal(t, objName1, result.Contents[0].Key)
|
||||||
|
require.Equal(t, hdr1.Get(api.ETag), result.Contents[0].ETag)
|
||||||
|
require.Equal(t, objName2, result.Contents[1].Key)
|
||||||
|
require.Equal(t, hdr2.Get(api.ETag), result.Contents[1].ETag)
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("listv2", func(t *testing.T) {
|
||||||
|
result := listObjectsV2(hc, bktName, "", "", "", "", -1)
|
||||||
|
|
||||||
|
require.Len(t, result.Contents, 2)
|
||||||
|
require.Equal(t, objName1, result.Contents[0].Key)
|
||||||
|
require.Equal(t, hdr1.Get(api.ETag), result.Contents[0].ETag)
|
||||||
|
require.Equal(t, objName2, result.Contents[1].Key)
|
||||||
|
require.Equal(t, hdr2.Get(api.ETag), result.Contents[1].ETag)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
func TestListObjectsPaging(t *testing.T) {
|
func TestListObjectsPaging(t *testing.T) {
|
||||||
hc := prepareHandlerContext(t)
|
hc := prepareHandlerContext(t)
|
||||||
|
|
||||||
|
|
|
@ -668,11 +668,11 @@ type LatestVersionsByPrefixStreamImpl struct {
|
||||||
tailPrefix string
|
tailPrefix string
|
||||||
namesMap map[uint64]string
|
namesMap map[uint64]string
|
||||||
ended bool
|
ended bool
|
||||||
|
latestOnly bool
|
||||||
|
currentLatest *data.NodeVersion
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *LatestVersionsByPrefixStreamImpl) Next(ctx context.Context) (*data.NodeVersion, error) {
|
func (s *LatestVersionsByPrefixStreamImpl) Next(ctx context.Context) (*data.NodeVersion, error) {
|
||||||
const latestOnly = true
|
|
||||||
|
|
||||||
if s.ended {
|
if s.ended {
|
||||||
return nil, io.EOF
|
return nil, io.EOF
|
||||||
}
|
}
|
||||||
|
@ -680,6 +680,12 @@ func (s *LatestVersionsByPrefixStreamImpl) Next(ctx context.Context) (*data.Node
|
||||||
if s.innerStream == nil {
|
if s.innerStream == nil {
|
||||||
node, err := s.mainStream.Next()
|
node, err := s.mainStream.Next()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
if errors.Is(err, io.EOF) {
|
||||||
|
s.ended = true
|
||||||
|
if s.latestOnly && s.currentLatest != nil {
|
||||||
|
return s.currentLatest, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
return nil, fmt.Errorf("main stream next: %w", err)
|
return nil, fmt.Errorf("main stream next: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -706,6 +712,9 @@ func (s *LatestVersionsByPrefixStreamImpl) Next(ctx context.Context) (*data.Node
|
||||||
if errors.Is(err, io.EOF) {
|
if errors.Is(err, io.EOF) {
|
||||||
s.innerStream = nil
|
s.innerStream = nil
|
||||||
s.namesMap = map[uint64]string{}
|
s.namesMap = map[uint64]string{}
|
||||||
|
if s.latestOnly && s.currentLatest != nil && s.currentLatest.ID != s.intermediateRootID {
|
||||||
|
return s.currentLatest, nil
|
||||||
|
}
|
||||||
return s.Next(ctx)
|
return s.Next(ctx)
|
||||||
}
|
}
|
||||||
return nil, fmt.Errorf("inner stream: %w", err)
|
return nil, fmt.Errorf("inner stream: %w", err)
|
||||||
|
@ -731,11 +740,32 @@ func (s *LatestVersionsByPrefixStreamImpl) Next(ctx context.Context) (*data.Node
|
||||||
s.namesMap[treeNode.ID] = filepath
|
s.namesMap[treeNode.ID] = filepath
|
||||||
}
|
}
|
||||||
|
|
||||||
if treeNode.ObjID.Equals(oid.ID{}) { // The node can be intermediate but we still want to update namesMap
|
if treeNode.ObjID.Equals(oid.ID{}) { // The node can be intermediate, but we still want to update namesMap
|
||||||
return s.Next(ctx)
|
return s.Next(ctx)
|
||||||
}
|
}
|
||||||
|
|
||||||
return newNodeVersionFromTreeNode(filepath, treeNode), nil
|
nodeVersion := newNodeVersionFromTreeNode(filepath, treeNode)
|
||||||
|
|
||||||
|
if s.latestOnly {
|
||||||
|
if s.currentLatest == nil {
|
||||||
|
s.currentLatest = nodeVersion
|
||||||
|
return s.Next(ctx)
|
||||||
|
}
|
||||||
|
|
||||||
|
if s.currentLatest.FilePath != nodeVersion.FilePath {
|
||||||
|
res := s.currentLatest
|
||||||
|
s.currentLatest = nodeVersion
|
||||||
|
return res, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if s.currentLatest.Timestamp < nodeVersion.Timestamp {
|
||||||
|
s.currentLatest = nodeVersion
|
||||||
|
}
|
||||||
|
|
||||||
|
return s.Next(ctx)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nodeVersion, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Tree) GetLatestVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string) (data.VersionsStream, error) {
|
func (c *Tree) GetLatestVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string) (data.VersionsStream, error) {
|
||||||
|
@ -756,6 +786,7 @@ func (c *Tree) GetLatestVersionsByPrefixStream(ctx context.Context, bktInfo *dat
|
||||||
mainStream: mainStream,
|
mainStream: mainStream,
|
||||||
headPrefix: strings.TrimSuffix(prefix, tailPrefix),
|
headPrefix: strings.TrimSuffix(prefix, tailPrefix),
|
||||||
tailPrefix: tailPrefix,
|
tailPrefix: tailPrefix,
|
||||||
|
latestOnly: true,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue