mirrors/versitygw
1
0
Fork 0
mirror of https://github.com/versity/versitygw.git synced 2026-01-10 13:27:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Fixes #286, Created a struct which handles s3 select event streaming and event message construction

Browse Source
This commit is contained in:
jonaustin09
2023-10-27 12:38:47 -04:00
committed by Ben McClelland
parent cd4821baa6
commit 48818927bb
5 changed files with 85 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
backend/backend.go
View File

@@ -15,6 +15,7 @@
package backend
import (
"bufio"
"context"
"fmt"
"io"
@@ -22,6 +23,7 @@ import (
"github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/versity/versitygw/s3err"
"github.com/versity/versitygw/s3response"
"github.com/versity/versitygw/s3select"
)
//go:generate moq -out ../s3api/controllers/backend_moq_test.go -pkg controllers . Backend
@@ -61,7 +63,7 @@ type Backend interface {
// special case object operations
RestoreObject(context.Context, *s3.RestoreObjectInput) error
SelectObjectContent(context.Context, *s3.SelectObjectContentInput) (s3response.SelectObjectContentResult, error)
SelectObjectContent(ctx context.Context, input *s3.SelectObjectContentInput) func(w *bufio.Writer)
// object tags operations
GetObjectTagging(_ context.Context, bucket, object string) (map[string]string, error)
@@ -162,8 +164,19 @@ func (BackendUnsupported) PutObjectAcl(context.Context, *s3.PutObjectAclInput) e
func (BackendUnsupported) RestoreObject(context.Context, *s3.RestoreObjectInput) error {
return s3err.GetAPIError(s3err.ErrNotImplemented)
}
func (BackendUnsupported) SelectObjectContent(context.Context, *s3.SelectObjectContentInput) (s3response.SelectObjectContentResult, error) {
return s3response.SelectObjectContentResult{}, s3err.GetAPIError(s3err.ErrNotImplemented)
func (BackendUnsupported) SelectObjectContent(ctx context.Context, input *s3.SelectObjectContentInput) func(w *bufio.Writer) {
return func(w *bufio.Writer) {
var getProgress s3select.GetProgress
progress := input.RequestProgress
if progress != nil && *progress.Enabled {
getProgress = func() (bytesScanned int64, bytesProcessed int64) {
return -1, -1
}
}
mh := s3select.NewMessageHandler(ctx, w, getProgress)
apiErr := s3err.GetAPIError(s3err.ErrNotImplemented)
mh.FinishWithError(apiErr.Code, apiErr.Description)
}
}
func (BackendUnsupported) GetObjectTagging(_ context.Context, bucket, object string) (map[string]string, error) {

33
s3api/controllers/backend_moq_test.go
View File

@@ -4,6 +4,7 @@
package controllers
import (
"bufio"
"context"
"github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/versity/versitygw/backend"
@@ -106,7 +107,7 @@ var _ backend.Backend = &BackendMock{}
// RestoreObjectFunc: func(contextMoqParam context.Context, restoreObjectInput *s3.RestoreObjectInput) error {
// panic("mock out the RestoreObject method")
// },
// SelectObjectContentFunc: func(contextMoqParam context.Context, selectObjectContentInput *s3.SelectObjectContentInput) (s3response.SelectObjectContentResult, error) {
// SelectObjectContentFunc: func(ctx context.Context, input *s3.SelectObjectContentInput) func(w *bufio.Writer) {
// panic("mock out the SelectObjectContent method")
// },
// ShutdownFunc: func() {
@@ -213,7 +214,7 @@ type BackendMock struct {
RestoreObjectFunc func(contextMoqParam context.Context, restoreObjectInput *s3.RestoreObjectInput) error
// SelectObjectContentFunc mocks the SelectObjectContent method.
SelectObjectContentFunc func(contextMoqParam context.Context, selectObjectContentInput *s3.SelectObjectContentInput) (s3response.SelectObjectContentResult, error)
SelectObjectContentFunc func(ctx context.Context, input *s3.SelectObjectContentInput) func(w *bufio.Writer)
// ShutdownFunc mocks the Shutdown method.
ShutdownFunc func()
@@ -441,10 +442,10 @@ type BackendMock struct {
}
// SelectObjectContent holds details about calls to the SelectObjectContent method.
SelectObjectContent []struct {
// ContextMoqParam is the contextMoqParam argument value.
ContextMoqParam context.Context
// SelectObjectContentInput is the selectObjectContentInput argument value.
SelectObjectContentInput *s3.SelectObjectContentInput
// Ctx is the ctx argument value.
Ctx context.Context
// Input is the input argument value.
Input *s3.SelectObjectContentInput
}
// Shutdown holds details about calls to the Shutdown method.
Shutdown []struct {
@@ -1539,21 +1540,21 @@ func (mock *BackendMock) RestoreObjectCalls() []struct {
}
// SelectObjectContent calls SelectObjectContentFunc.
func (mock *BackendMock) SelectObjectContent(contextMoqParam context.Context, selectObjectContentInput *s3.SelectObjectContentInput) (s3response.SelectObjectContentResult, error) {
func (mock *BackendMock) SelectObjectContent(ctx context.Context, input *s3.SelectObjectContentInput) func(w *bufio.Writer) {
if mock.SelectObjectContentFunc == nil {
panic("BackendMock.SelectObjectContentFunc: method is nil but Backend.SelectObjectContent was just called")
}
callInfo := struct {
ContextMoqParam context.Context
SelectObjectContentInput *s3.SelectObjectContentInput
Ctx context.Context
Input *s3.SelectObjectContentInput
}{
ContextMoqParam: contextMoqParam,
SelectObjectContentInput: selectObjectContentInput,
Ctx: ctx,
Input: input,
}
mock.lockSelectObjectContent.Lock()
mock.calls.SelectObjectContent = append(mock.calls.SelectObjectContent, callInfo)
mock.lockSelectObjectContent.Unlock()
return mock.SelectObjectContentFunc(contextMoqParam, selectObjectContentInput)
return mock.SelectObjectContentFunc(ctx, input)
}
// SelectObjectContentCalls gets all the calls that were made to SelectObjectContent.
@@ -1561,12 +1562,12 @@ func (mock *BackendMock) SelectObjectContent(contextMoqParam context.Context, se
//
// len(mockedBackend.SelectObjectContentCalls())
func (mock *BackendMock) SelectObjectContentCalls() []struct {
ContextMoqParam context.Context
SelectObjectContentInput *s3.SelectObjectContentInput
Ctx context.Context
Input *s3.SelectObjectContentInput
} {
var calls []struct {
ContextMoqParam context.Context
SelectObjectContentInput *s3.SelectObjectContentInput
Ctx context.Context
Input *s3.SelectObjectContentInput
}
mock.lockSelectObjectContent.RLock()
calls = mock.calls.SelectObjectContent

7
s3api/controllers/base.go
View File

@@ -906,7 +906,7 @@ func (c S3ApiController) CreateActions(ctx *fiber.Ctx) error {
return SendXMLResponse(ctx, nil, err, &MetaOpts{Logger: c.logger, Action: "SelectObjectContent", BucketOwner: parsedAcl.Owner})
}
res, err := c.be.SelectObjectContent(ctx.Context(), &s3.SelectObjectContentInput{
sw := c.be.SelectObjectContent(ctx.Context(), &s3.SelectObjectContentInput{
Bucket: &bucket,
Key: &key,
Expression: payload.Expression,
@@ -916,7 +916,10 @@ func (c S3ApiController) CreateActions(ctx *fiber.Ctx) error {
RequestProgress: payload.RequestProgress,
ScanRange: payload.ScanRange,
})
return SendXMLResponse(ctx, res, err, &MetaOpts{Logger: c.logger, Action: "SelectObjectContent", BucketOwner: parsedAcl.Owner})
ctx.Context().SetBodyStreamWriter(sw)
return nil
}
if uploadId != "" {

5
s3api/controllers/base_test.go
View File

@@ -15,6 +15,7 @@
package controllers
import (
"bufio"
"context"
"encoding/json"
"fmt"
@@ -1308,8 +1309,8 @@ func TestS3ApiController_CreateActions(t *testing.T) {
CreateMultipartUploadFunc: func(context.Context, *s3.CreateMultipartUploadInput) (*s3.CreateMultipartUploadOutput, error) {
return &s3.CreateMultipartUploadOutput{}, nil
},
SelectObjectContentFunc: func(contextMoqParam context.Context, selectObjectContentInput *s3.SelectObjectContentInput) (s3response.SelectObjectContentResult, error) {
return s3response.SelectObjectContentResult{}, nil
SelectObjectContentFunc: func(context.Context, *s3.SelectObjectContentInput) func(w *bufio.Writer) {
return func(w *bufio.Writer) {}
},
},
}

44
s3select/message-handler.go Normal file
View File

@@ -0,0 +1,44 @@
// Copyright 2023 Versity Software
// This file is licensed under the Apache License, Version 2.0
// (the "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
package s3select
import (
"bufio"
"context"
)
type GetProgress func() (bytesScanned int64, bytesProcessed int64)
type MessageHandler struct{}
// Creates a new MessageHandler instance and starts the event streaming
func NewMessageHandler(ctx context.Context, w *bufio.Writer, getProgressFunc GetProgress) *MessageHandler {
return &MessageHandler{}
}
// SendRecord sends a single Records message
func (mh *MessageHandler) SendRecord(payload []byte) error {
return nil
}
// Finish terminates message stream with Stat and End message
func (mh *MessageHandler) Finish() error {
return nil
}
// FinishWithError terminates event stream with error
func (mh *MessageHandler) FinishWithError(errorCode, errorMessage string) error {
return nil
}