diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 00000000..5008ddfc Binary files /dev/null and b/.DS_Store differ diff --git a/file-server/README.md b/file-server/README.md index c709d04f..16590b40 100644 --- a/file-server/README.md +++ b/file-server/README.md @@ -52,17 +52,44 @@ paths: ## Server side -The handler receives a `io.ReadCloser` as the file to consume. +For `x-go-server-streaming: true`, generated binding only constructs a +`*runtime.MultipartFormStream` and passes ownership to the handler. It does not +consume parts or try to populate generated file and form fields. -Under the hood, the runtime builds this with a `*runtime.File`, which provides access to some header information, such as: +The handler traverses the multipart body sequentially: ```go - if namedFile, ok := params.File.(*runtime.File); ok { - log.Printf("received file name: %s", namedFile.Header.Filename) - log.Printf("received file size: %d", namedFile.Header.Size) + for { + file, err := params.MultipartForm.NextFile() + if errors.Is(err, io.EOF) { + break } + // Validate file.FieldName and consume file here. + } ``` +Each `runtime.StreamedFile` reads directly from the HTTP request body. The +filename and MIME headers are available before the payload is consumed, but the +complete file size is not known in advance. + +## Server-side streaming + +The server binding is generated from `x-go-server-streaming: true`. The +generated binder creates a `*runtime.MultipartFormStream` without reading +multipart parts ahead of the handler. Required fields, accepted file field +names, multiplicity and other application-specific rules are validated by the +handler while traversing the stream. + +`MultipartFormStream.Fields()` and `MultipartFormStream.Files()` return +snapshots of ordinary fields and file metadata discovered so far. They do not +read ahead: trailing fields become visible only after the active file is +consumed or closed and the stream advances. + +The handler owns the multipart stream and must either: + +- call `Drain()` to process all remaining parts and close the request body; or +- call `Close()` to abort multipart processing. + ## Client side The local file is handled as a `runtime.NamedReadCloser` (that is, a `io.ReadCloser` plus the `Name() string` method). diff --git a/file-server/restapi/configure_file_upload.go b/file-server/restapi/configure_file_upload.go index 2db53567..6e4f41aa 100644 --- a/file-server/restapi/configure_file_upload.go +++ b/file-server/restapi/configure_file_upload.go @@ -46,8 +46,8 @@ func configureAPI(api *operations.FileUploadAPI) http.Handler { api.JSONProducer = runtime.JSONProducer() - // You may change here the memory limit for this multipart form parser. Below is the default (32 MB). - // uploads.UploadFileMaxParseMemory = 32 << 20 + // You may change here the maximum body size for this streaming multipart form. Below is the default (32 MB). + // uploads.UploadFileMaxBodySize = 32 << 20 uploadFolder, err := os.MkdirTemp(".", "upload") if err != nil { @@ -57,34 +57,55 @@ func configureAPI(api *operations.FileUploadAPI) http.Handler { // snippet:upload-handler api.UploadsUploadFileHandler = uploads.UploadFileHandlerFunc(func(params uploads.UploadFileParams) middleware.Responder { - if params.File == nil { - return middleware.Error(http.StatusNotFound, stderrors.New("no file provided")) + if params.MultipartForm == nil { + return middleware.Error(http.StatusInternalServerError, stderrors.New("multipart stream is not initialized")) } defer func() { - _ = params.File.Close() + _ = params.MultipartForm.Close() }() - if namedFile, ok := params.File.(*runtime.File); ok { - log.Printf("received file name: %s", namedFile.Header.Filename) - log.Printf("received file size: %d", namedFile.Header.Size) + uploadedFiles := 0 + for { + file, err := params.MultipartForm.NextFile() + if stderrors.Is(err, io.EOF) { + break + } + if err != nil { + return uploadError(err) + } + + if file.FieldName != "file" { + if err := file.Close(); err != nil { + return uploadError(err) + } + + continue + } + + log.Printf("received file name: %s", file.Filename) + log.Printf("received content type: %s", file.Header.Get(runtime.HeaderContentType)) + + filename := path.Join(uploadFolder, fmt.Sprintf("uploaded_file_%d.dat", uploadCounter)) + uploadCounter++ + n, err := saveStreamedFile(filename, file) + if err != nil { + return uploadError(err) + } + uploadedFiles++ + + log.Printf("copied bytes %d", n) + log.Printf("file uploaded copied as %s", filename) } - // uploads file and save it locally - filename := path.Join(uploadFolder, fmt.Sprintf("uploaded_file_%d.dat", uploadCounter)) - uploadCounter++ - f, err := os.OpenFile(filename, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0o600) - if err != nil { - return middleware.Error(http.StatusInternalServerError, stderrors.New("could not create file on server")) + if uploadedFiles == 0 { + return middleware.Error(http.StatusBadRequest, stderrors.New("no file provided")) } - - n, err := io.Copy(f, params.File) - if err != nil { - return middleware.Error(http.StatusInternalServerError, stderrors.New("could not upload file on server")) + if err := params.MultipartForm.Drain(); err != nil { + return uploadError(err) } - log.Printf("copied bytes %d", n) - - log.Printf("file uploaded copied as %s", filename) + log.Printf("discovered multipart fields: %v", params.MultipartForm.Fields()) + log.Printf("discovered multipart files: %d", len(params.MultipartForm.Files())) return uploads.NewUploadFileOK() }) @@ -122,5 +143,26 @@ func setupMiddlewares(handler http.Handler) http.Handler { // The middleware configuration happens before anything, this middleware also applies to serving the swagger.json document. // So this is a good place to plug in a panic handling middleware, logging and metrics. func setupGlobalMiddleware(handler http.Handler) http.Handler { - return handler + return http.MaxBytesHandler(handler, uploads.UploadFileMaxBodySize) +} + +func saveStreamedFile(filename string, file io.Reader) (int64, error) { + f, err := os.OpenFile(filename, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0o600) + if err != nil { + return 0, fmt.Errorf("create upload file: %w", err) + } + + n, copyErr := io.Copy(f, file) + closeErr := f.Close() + + return n, stderrors.Join(copyErr, closeErr) +} + +func uploadError(err error) middleware.Responder { + var maxBytesErr *http.MaxBytesError + if stderrors.As(err, &maxBytesErr) { + return middleware.Error(http.StatusRequestEntityTooLarge, err) + } + + return middleware.Error(http.StatusInternalServerError, stderrors.New("could not upload file on server")) } diff --git a/file-server/restapi/embedded_spec.go b/file-server/restapi/embedded_spec.go index f8341ec5..1fc0c69b 100644 --- a/file-server/restapi/embedded_spec.go +++ b/file-server/restapi/embedded_spec.go @@ -45,6 +45,7 @@ func init() { "parameters": [ { "type": "file", + "x-go-server-streaming": true, "name": "file", "in": "formData", "required": true @@ -90,6 +91,7 @@ func init() { "parameters": [ { "type": "file", + "x-go-server-streaming": true, "name": "file", "in": "formData", "required": true diff --git a/file-server/restapi/operations/uploads/upload_file_parameters.go b/file-server/restapi/operations/uploads/upload_file_parameters.go index 6f2c8618..36d71025 100644 --- a/file-server/restapi/operations/uploads/upload_file_parameters.go +++ b/file-server/restapi/operations/uploads/upload_file_parameters.go @@ -3,8 +3,6 @@ package uploads import ( - "io" - "mime/multipart" "net/http" "github.com/go-openapi/errors" @@ -12,13 +10,6 @@ import ( "github.com/go-openapi/runtime/middleware" ) -// UploadFileMaxParseMemory sets the maximum size in bytes for -// the multipart form parser for this operation. -// -// The default value is 32 MB. -// The multipart parser stores up to this + 10MB. -var UploadFileMaxParseMemory int64 = 32 << 20 - // UploadFileMaxBodySize caps the size of the form body. // // The default value is 32 MB. Larger bodies will error with http status 413. @@ -39,9 +30,11 @@ func NewUploadFileParams() UploadFileParams { type UploadFileParams struct { // HTTP Request Object HTTPRequest *http.Request `json:"-"` - // Required: true - // In: formData - File io.ReadCloser + + // MultipartForm is the lazily consumed multipart form stream. + // + // The handler owns traversal, validation and closing or draining this stream. + MultipartForm *runtime.MultipartFormStream `json:"-"` } // BindRequest both binds and validates a request, it assumes that complex things implement a Validatable(strfmt.Registry) error interface @@ -52,30 +45,18 @@ func (o *UploadFileParams) BindRequest(r *http.Request, route *middleware.Matche var res []error o.HTTPRequest = r - isBlocking, err := runtime.BindForm(r, - runtime.BindFormMaxParseMemory(UploadFileMaxParseMemory), - runtime.BindFormMaxBody(UploadFileMaxBodySize), - runtime.BindFormFile("file", true, o.bindFile), - ) - if err != nil { - if isBlocking { - return err - } - - res = append(res, err) - } if len(res) > 0 { return errors.CompositeValidationError(res...) } - return nil -} - -// bindFile validates file parameter File1 and assigns it as a *runtime.File on success. -// -// The only supported validations on files are MinLength and MaxLength -func (o *UploadFileParams) bindFile(file multipart.File, header *multipart.FileHeader) error { - o.File = &runtime.File{Data: file, Header: header} + multipartForm, err := runtime.NewMultipartFormStream( + r, + runtime.MultipartFormStreamMaxBody(UploadFileMaxBodySize), + ) + if err != nil { + return err + } + o.MultipartForm = multipartForm return nil } diff --git a/file-server/restapi/streaming_upload_integration_test.go b/file-server/restapi/streaming_upload_integration_test.go new file mode 100644 index 00000000..6a6b2337 --- /dev/null +++ b/file-server/restapi/streaming_upload_integration_test.go @@ -0,0 +1,174 @@ +// SPDX-FileCopyrightText: Copyright 2015-2026 go-swagger maintainers +// SPDX-License-Identifier: Apache-2.0 + +package restapi + +import ( + stderrors "errors" + "io" + "mime/multipart" + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/go-openapi/loads" + "github.com/go-openapi/runtime" + "github.com/go-openapi/runtime/middleware" + "github.com/go-openapi/testify/v2/assert" + "github.com/go-openapi/testify/v2/require" + + "github.com/go-swagger/examples/file-server/restapi/operations" + "github.com/go-swagger/examples/file-server/restapi/operations/uploads" +) + +type uploadHTTPResult struct { + statusCode int + err error +} + +type streamingHandlerProgress struct { + firstChunk string + fields url.Values + files []runtime.MultipartFileInfo +} + +type streamingHandlerResult struct { + fields url.Values + files []runtime.MultipartFileInfo + err error +} + +func TestStreamingUploadReachesHandlerBeforeRequestBodyCompletes(t *testing.T) { + const ( + firstChunk = "first" + secondChunk = "second" + ) + + swaggerSpec, err := loads.Embedded(SwaggerJSON, FlatSwaggerJSON) + require.NoError(t, err) + + handlerRead := make(chan streamingHandlerProgress, 1) + handlerDone := make(chan streamingHandlerResult, 1) + + api := operations.NewFileUploadAPI(swaggerSpec) + api.MultipartformConsumer = runtime.DiscardConsumer + api.UploadsUploadFileHandler = uploads.UploadFileHandlerFunc(func(params uploads.UploadFileParams) middleware.Responder { + defer func() { + _ = params.MultipartForm.Close() + }() + + file, handlerErr := params.MultipartForm.NextFile() + if handlerErr == nil { + first := make([]byte, len(firstChunk)) + _, handlerErr = io.ReadFull(file, first) + if handlerErr == nil { + handlerRead <- streamingHandlerProgress{ + firstChunk: string(first), + fields: params.MultipartForm.Fields(), + files: params.MultipartForm.Files(), + } + _, handlerErr = io.Copy(io.Discard, file) + } + } + if handlerErr == nil { + _, handlerErr = params.MultipartForm.NextFile() + if stderrors.Is(handlerErr, io.EOF) { + handlerErr = nil + } + } + if handlerErr == nil { + handlerErr = params.MultipartForm.Drain() + } + + handlerDone <- streamingHandlerResult{ + fields: params.MultipartForm.Fields(), + files: params.MultipartForm.Files(), + err: handlerErr, + } + if handlerErr != nil { + return middleware.Error(http.StatusInternalServerError, handlerErr) + } + + return uploads.NewUploadFileOK() + }) + + server := httptest.NewServer(api.Serve(nil)) + defer server.Close() + + bodyReader, bodyWriter := io.Pipe() + defer func() { + _ = bodyReader.Close() + }() + defer func() { + _ = bodyWriter.Close() + }() + + multipartWriter := multipart.NewWriter(bodyWriter) + request, err := http.NewRequestWithContext(t.Context(), http.MethodPost, server.URL+"/upload", bodyReader) + require.NoError(t, err) + request.Header.Set(runtime.HeaderContentType, multipartWriter.FormDataContentType()) + + responseDone := make(chan uploadHTTPResult, 1) + go func() { + response, requestErr := server.Client().Do(request) + if requestErr != nil { + responseDone <- uploadHTTPResult{err: requestErr} + + return + } + defer func() { + _ = response.Body.Close() + }() + + responseDone <- uploadHTTPResult{statusCode: response.StatusCode} + }() + + require.NoError(t, multipartWriter.WriteField("before", "one")) + part, err := multipartWriter.CreateFormFile("file", "payload.bin") + require.NoError(t, err) + _, err = io.WriteString(part, firstChunk) + require.NoError(t, err) + + select { + case progress := <-handlerRead: + assert.EqualT(t, firstChunk, progress.firstChunk) + assert.EqualT(t, "one", progress.fields.Get("before")) + assert.Empty(t, progress.fields.Get("after")) + require.Len(t, progress.files, 1) + assert.EqualT(t, "file", progress.files[0].FieldName) + assert.EqualT(t, "payload.bin", progress.files[0].Filename) + case result := <-responseDone: + require.NoError(t, result.err) + t.Fatal("request completed before the multipart body was resumed") + case <-time.After(time.Second): + t.Fatal("handler did not receive the first file chunk while the request body was still open") + } + + _, err = io.WriteString(part, secondChunk) + require.NoError(t, err) + require.NoError(t, multipartWriter.WriteField("after", "two")) + require.NoError(t, multipartWriter.Close()) + require.NoError(t, bodyWriter.Close()) + + var result uploadHTTPResult + select { + case result = <-responseDone: + case <-time.After(time.Second): + t.Fatal("request did not complete after the multipart body was closed") + } + require.NoError(t, result.err) + assert.EqualT(t, http.StatusOK, result.statusCode) + + select { + case handlerResult := <-handlerDone: + require.NoError(t, handlerResult.err) + assert.EqualT(t, "one", handlerResult.fields.Get("before")) + assert.EqualT(t, "two", handlerResult.fields.Get("after")) + require.Len(t, handlerResult.files, 1) + assert.EqualT(t, "payload.bin", handlerResult.files[0].Filename) + case <-time.After(time.Second): + t.Fatal("upload handler did not complete") + } +} diff --git a/file-server/swagger.yml b/file-server/swagger.yml index 9626f1ab..79f15ed4 100644 --- a/file-server/swagger.yml +++ b/file-server/swagger.yml @@ -26,6 +26,7 @@ paths: in: formData type: file required: true + x-go-server-streaming: true # endsnippet:upload-path responses: "200": diff --git a/go.mod b/go.mod index 9400d416..b8fdab2a 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( github.com/coreos/go-oidc/v3 v3.20.0 github.com/go-openapi/errors v0.22.8 github.com/go-openapi/loads v0.25.0 - github.com/go-openapi/runtime v0.32.6 + github.com/go-openapi/runtime v0.33.0 github.com/go-openapi/spec v0.22.9 github.com/go-openapi/strfmt v0.27.0 github.com/go-openapi/swag/cmdutils v0.27.3 diff --git a/go.sum b/go.sum index 45246c79..d1616796 100644 --- a/go.sum +++ b/go.sum @@ -32,8 +32,8 @@ github.com/go-openapi/jsonreference v1.0.0 h1:jlmTr6torcd1YgDQvSfNmRtKzYDO4FGBkr github.com/go-openapi/jsonreference v1.0.0/go.mod h1:jtwdyGbJk0Xhe5Y+rwtglQP6Sb1WZST4rT32LWB+sv0= github.com/go-openapi/loads v0.25.0 h1:74Bc2snfaVlsHzwdQj/3gsA9XJz3daXTJVs+4ZaK7jI= github.com/go-openapi/loads v0.25.0/go.mod h1:JFBw4SIB9+PTIFHDfcXuSSy5h6aWzjtUCrPYyx3qWU8= -github.com/go-openapi/runtime v0.32.6 h1:hrcTTF8P7ZZr2Majzq11I65QtL/s85o7Q+zJf+AvFN4= -github.com/go-openapi/runtime v0.32.6/go.mod h1:+rsupH3+TFKqmFysqkmgBOTxpVJV8eV+j9myvvea2Xw= +github.com/go-openapi/runtime v0.33.0 h1:Dd3Oj2ig+WH8ckK95l0Wn2V8a4bH/UqWPRZVT0vc8yU= +github.com/go-openapi/runtime v0.33.0/go.mod h1:+rsupH3+TFKqmFysqkmgBOTxpVJV8eV+j9myvvea2Xw= github.com/go-openapi/runtime/server-middleware v0.32.6 h1:IGTYzybyFrUeSqQEwwO1y/9KnOk4QsabFNtAQtIHxDE= github.com/go-openapi/runtime/server-middleware v0.32.6/go.mod h1:OQHTBqMGquJShXhPYQ62yAqDMtC1rYpsEwldNWjYKhA= github.com/go-openapi/spec v0.22.9 h1:/vKIFDcGKp0ktZWGbym/tJEWbk6/XOEmAVU0kqKMH+w=