feat(storage): provide optional chunk size field for GetObject

internal/plugin/loopback/options.go

@@ -38,6 +38,10 @@ const (
 	PutObject
 )
 
+const (
+	defaultStreamChunckSize = 8
+)
+
 // PluginMockError is used to mock an error when interacting with an external object store.
 type PluginMockError struct {
 	BucketName   string

internal/plugin/loopback/storage.go

@@ -284,9 +284,21 @@ func (l *LoopbackStorage) getObject(req *plgpb.GetObjectRequest, stream plgpb.St
 		}
 	}
 	go func() {
+		chunkSize := req.GetChunkSize()
+		if chunkSize == 0 {
+			chunkSize = defaultStreamChunckSize
+		}
+		data := []byte{}
 		for _, chunk := range object.DataChunks {
+			data = append(data, chunk...)
+		}
+		for i := 0; i < len(data); i += int(chunkSize) {
+			end := i + int(chunkSize)
+			if end > len(data) {
+				end = len(data)
+			}
 			if err := stream.Send(&plgpb.GetObjectResponse{
-				FileChunk: chunk,
+				FileChunk: append([]byte{}, data[i:end]...),
 			}); err != nil {
 				stream.SendMsg(status.Errorf(codes.Internal, "%s: failed to send object data: %v", op, err))
 				return

internal/plugin/loopback/storage_test.go

@@ -629,6 +629,18 @@ func TestLoopbackGetObject(t *testing.T) {
 			},
 			expectedErr: codes.PermissionDenied,
 		},
+		{
+			name: "with chunk size",
+			request: &plgpb.GetObjectRequest{
+				Key: "mock_object",
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+				},
+				ChunkSize: 3,
+			},
+			expectedObj: mockStorageMapData["aws_s3_mock"]["mock_object"],
+		},
 		{
 			name: "object retrieved",
 			request: &plgpb.GetObjectRequest{