Feat/large file upload

CHANGELOG.md

@@ -11,6 +11,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+## [1.2.0] - 2024-02-02
+
+### Added
+
+- Added support for large file uploads.
+
 ## [1.0.2] - 2023-12-01
 
 ### Changed

fileuploader/file_uploader_util.go

@@ -0,0 +1,95 @@
+package fileuploader
+
+import (
+	"strings"
+	"time"
+)
+
+type rangePair struct {
+	Start int64
+	End   int64
+}
+
+func stringIsNullOrEmpty(s string) bool {
+	s = strings.TrimSpace(s)
+	if s == "" || len(s) == 0 {
+		return true
+	}
+	return false
+}
+
+type UploadSession interface {
+	GetExpirationDateTime() *time.Time
+	SetExpirationDateTime(expirationDateTime *time.Time)
+	GetNextExpectedRanges() []string
+	SetNextExpectedRanges(nextExpectedRanges []string)
+	GetOdataType() *string
+	GetUploadUrl() *string
+}
+
+type ProgressCallBack func(current int64, total int64)
+
+type UploadResult[T interface{}] interface {
+	SetItemResponse(response T)
+	GetItemResponse() T
+	SetUploadSession(uploadSession UploadSession)
+	GetUploadSession() UploadSession
+	SetURI(uri *string)
+	GetURI() *string
+	SetUploadSucceeded(isSuccessful bool)
+	GetUploadSucceeded() bool
+	SetResponseErrors(errors []error)
+	GetResponseErrors() []error
+}
+
+func NewUploadResult[T interface{}]() UploadResult[T] {
+	return &uploadResult[T]{}
+}
+
+type uploadResult[T interface{}] struct {
+	itemResponse    T
+	uploadSession   UploadSession
+	uri             *string
+	uploadSucceeded bool
+	responseErrors  []error
+}
+
+func (u *uploadResult[T]) SetItemResponse(response T) {
+	u.itemResponse = response
+}
+
+func (u *uploadResult[T]) GetItemResponse() T {
+	return u.itemResponse
+}
+
+func (u *uploadResult[T]) SetUploadSession(uploadSession UploadSession) {
+	u.uploadSession = uploadSession
+}
+
+func (u *uploadResult[T]) GetUploadSession() UploadSession {
+	return u.uploadSession
+}
+
+func (u *uploadResult[T]) SetURI(uri *string) {
+	u.uri = uri
+}
+
+func (u *uploadResult[T]) GetURI() *string {
+	return u.uri
+}
+
+func (u *uploadResult[T]) SetUploadSucceeded(isSuccessful bool) {
+	u.uploadSucceeded = isSuccessful
+}
+
+func (u *uploadResult[T]) GetUploadSucceeded() bool {
+	return u.uploadSucceeded
+}
+
+func (u *uploadResult[T]) SetResponseErrors(errors []error) {
+	u.responseErrors = errors
+}
+
+func (u *uploadResult[T]) GetResponseErrors() []error {
+	return u.responseErrors
+}

fileuploader/large_file_session.go

@@ -0,0 +1,89 @@
+package fileuploader
+
+import (
+	"github.com/microsoft/kiota-abstractions-go/serialization"
+	"time"
+)
+
+type UploadSessionResponse interface {
+	serialization.Parsable
+	GetExpirationDateTime() *time.Time
+	SetExpirationDateTime(expirationDateTime *time.Time)
+	GetNextExpectedRanges() []string
+	SetNextExpectedRanges(nextExpectedRanges []string)
+}
+
+type largeFileUploadSession struct {
+	expirationDateTime *time.Time
+	nextExpectedRanges []string
+}
+
+func (l *largeFileUploadSession) Serialize(writer serialization.SerializationWriter) error {
+	if l.expirationDateTime != nil {
+		if err := writer.WriteTimeValue("expirationDateTime", l.expirationDateTime); err != nil {
+			return err
+		}
+	}
+	if l.nextExpectedRanges != nil {
+		if err := writer.WriteCollectionOfStringValues("nextExpectedRanges", l.nextExpectedRanges); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (l *largeFileUploadSession) GetFieldDeserializers() map[string]func(serialization.ParseNode) error {
+	return map[string]func(serialization.ParseNode) error{
+		"expirationDateTime": func(n serialization.ParseNode) error {
+			val, err := n.GetTimeValue()
+			if err != nil {
+				return err
+			}
+			if val != nil {
+				l.SetExpirationDateTime(val)
+			}
+			return nil
+		},
+		"nextExpectedRanges": func(n serialization.ParseNode) error {
+			val, err := n.GetCollectionOfPrimitiveValues("string")
+			if err != nil {
+				return err
+			}
+			if val != nil {
+				res := make([]string, len(val))
+				for i, v := range val {
+					if v != nil {
+						res[i] = *(v.(*string))
+					}
+				}
+				l.SetNextExpectedRanges(res)
+			}
+			return nil
+		},
+	}
+}
+
+func (l *largeFileUploadSession) GetExpirationDateTime() *time.Time {
+	return l.expirationDateTime
+}
+
+func (l *largeFileUploadSession) SetExpirationDateTime(expirationDateTime *time.Time) {
+	l.expirationDateTime = expirationDateTime
+}
+
+func (l *largeFileUploadSession) GetNextExpectedRanges() []string {
+	return l.nextExpectedRanges
+}
+
+func (l *largeFileUploadSession) SetNextExpectedRanges(nextExpectedRanges []string) {
+	l.nextExpectedRanges = nextExpectedRanges
+}
+
+func newLargeFileUploadSession() UploadSessionResponse {
+	return &largeFileUploadSession{}
+}
+
+// CreateUploadSessionDiscriminator creates a new instance of the appropriate class based on discriminator value
+func CreateUploadSessionDiscriminator(serialization.ParseNode) (serialization.Parsable, error) {
+	return newLargeFileUploadSession(), nil
+}

fileuploader/large_file_upload_task.go

@@ -0,0 +1,196 @@
+package fileuploader
+
+import (
+	"context"
+	"errors"
+	abstractions "github.com/microsoft/kiota-abstractions-go"
+	"github.com/microsoft/kiota-abstractions-go/serialization"
+	"io"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+type LargeFileUploadTask[T serialization.Parsable] interface {
+	Upload(progress ProgressCallBack) UploadResult[T]
+	Resume(progress ProgressCallBack) (UploadResult[T], error)
+	RefreshUploadStatus() error
+	Cancel() error
+}
+
+// ByteStream is an interface that represents a stream of bytes
+type ByteStream interface {
+	io.ReaderAt
+	Stat() (os.FileInfo, error)
+}
+
+type largeFileUploadTask[T serialization.Parsable] struct {
+	uploadSession   UploadSession
+	adapter         abstractions.RequestAdapter
+	byteStream      ByteStream // *os.File by default implements ByteStream
+	maxSlice        int64
+	parsableFactory serialization.ParsableFactory
+	errorMappings   abstractions.ErrorMappings
+}
+
+func NewLargeFileUploadTask[T serialization.Parsable](adapter abstractions.RequestAdapter, uploadSession UploadSession, byteStream ByteStream, maxSlice int64, parsableFactory serialization.ParsableFactory, errorMappings abstractions.ErrorMappings) LargeFileUploadTask[T] {
+	return &largeFileUploadTask[T]{
+		adapter:         adapter,
+		uploadSession:   uploadSession,
+		byteStream:      byteStream,
+		maxSlice:        maxSlice,
+		parsableFactory: parsableFactory,
+		errorMappings:   errorMappings,
+	}
+}
+
+// Upload uploads the byteStream in slices and returns the result of the upload
+func (l *largeFileUploadTask[T]) Upload(progress ProgressCallBack) UploadResult[T] {
+	result := NewUploadResult[T]()
+	slices := l.createUploadSlices()
+	maxRetriesPerRequest := 3
+
+	// slices of errors
+	var responseErrors []error
+	var itemResponse T
+	var location *string
+
+	var wg sync.WaitGroup
+	wg.Add(len(slices))
+
+	for _, slice := range slices {
+		uploadSlice := slice
+		go func() {
+			defer wg.Done()
+			response, uploadLocation, err := l.uploadWithRetry(uploadSlice, maxRetriesPerRequest)
+			if err != nil {
+				responseErrors = append(responseErrors, err)
+			} else {
+				progress(uploadSlice.RangeEnd, uploadSlice.TotalSessionLength)
+			}
+			if response != nil {
+				itemResponse = response.(T)
+			}
+			location = uploadLocation
+		}()
+	}
+
+	wg.Wait()
+
+	if len(responseErrors) > 0 {
+		result.SetUploadSucceeded(false)
+		result.SetResponseErrors(responseErrors)
+	} else {
+		result.SetUploadSucceeded(true)
+		result.SetUploadSession(l.uploadSession)
+		result.SetItemResponse(itemResponse)
+		result.SetURI(location)
+	}
+
+	return result
+}
+
+// Resume uploads the byteStream in slices and returns the result of the upload
+func (l *largeFileUploadTask[T]) Resume(progress ProgressCallBack) (UploadResult[T], error) {
+	err := l.RefreshUploadStatus()
+	if err != nil {
+		return nil, err
+	}
+
+	if len(l.uploadSession.GetNextExpectedRanges()) == 0 {
+		return nil, errors.New("UploadSession does not have next expected ranges")
+	}
+
+	if l.uploadSession.GetExpirationDateTime().Before(time.Now()) {
+		return nil, errors.New("UploadSession has expired")
+	}
+
+	return l.Upload(progress), nil
+}
+
+func (l *largeFileUploadTask[T]) RefreshUploadStatus() error {
+	requestInfo := abstractions.NewRequestInformation()
+	requestInfo.UrlTemplate = *l.uploadSession.GetUploadUrl()
+	requestInfo.Method = abstractions.GET
+	requestInfo.Headers.TryAdd("Accept", "application/json")
+
+	result, err := l.adapter.Send(context.Background(), requestInfo, CreateUploadSessionDiscriminator, l.errorMappings)
+	if err != nil {
+		return err
+	}
+
+	sessionResponse := result.(UploadSessionResponse)
+
+	l.uploadSession.SetExpirationDateTime(sessionResponse.GetExpirationDateTime())
+	l.uploadSession.SetNextExpectedRanges(sessionResponse.GetNextExpectedRanges())
+
+	return nil
+}
+
+// Cancel cancels the upload
+func (l *largeFileUploadTask[T]) Cancel() error {
+	requestInfo := abstractions.NewRequestInformationWithMethodAndUrlTemplateAndPathParameters(abstractions.DELETE, *l.uploadSession.GetUploadUrl(), make(map[string]string))
+	err := l.adapter.SendNoContent(context.Background(), requestInfo, l.errorMappings)
+	return err
+}
+
+func (l *largeFileUploadTask[T]) uploadWithRetry(slice uploadSlice[T], maxRetry int) (interface{}, *string, error) {
+	retry := 1
+	var parseable interface{}
+	var location *string
+	var err error
+	for retry < maxRetry {
+		// store the result of the upload
+		parseable, location, err = slice.Upload(l.parsableFactory) // check if successful
+		if err != nil {
+			if retry >= maxRetry {
+				return nil, nil, err
+			}
+			// backoff before retrying
+			time.Sleep(time.Duration(retry) * time.Second)
+		}
+		retry++
+	}
+	return parseable, location, err
+}
+
+func (l *largeFileUploadTask[T]) getRangesRemaining() []rangePair {
+	rangePairs := make([]rangePair, len(l.uploadSession.GetNextExpectedRanges()))
+
+	for i, ranges := range l.uploadSession.GetNextExpectedRanges() {
+		rangeValues := strings.Split(ranges, "-")
+
+		var startRange int64
+		if s, err := strconv.ParseInt(rangeValues[0], 10, 64); err == nil {
+			startRange = s
+		}
+
+		var endRange int64
+		if !stringIsNullOrEmpty(rangeValues[1]) {
+			if s, err := strconv.ParseInt(rangeValues[1], 10, 64); err == nil {
+				if endRange > l.fileSize() {
+					endRange = l.fileSize() - 1
+				} else {
+					endRange = s
+				}
+			}
+		} else {
+			endRange = l.fileSize() - 1
+		}
+
+		rangePairs[i] = rangePair{
+			Start: startRange,
+			End:   endRange,
+		}
+	}
+
+	return rangePairs
+}
+
+// returns the size of a byteStream
+func (l *largeFileUploadTask[T]) fileSize() int64 {
+	fileInfo, _ := l.byteStream.Stat()
+	return fileInfo.Size()
+}