Support HTTP/1.1 and add a demo server

Package.swift

@@ -3,7 +3,7 @@
 //
 // This source file is part of the SwiftNIO open source project
 //
-// Copyright (c) 2017-2022 Apple Inc. and the SwiftNIO project authors
+// Copyright (c) 2017-2024 Apple Inc. and the SwiftNIO project authors
 // Licensed under Apache License v2.0
 //
 // See LICENSE.txt for license information
@@ -180,6 +180,16 @@ var targets: [PackageDescription.Target] = [
             .product(name: "Atomics", package: "swift-atomics"),
         ]
     ),
+    .executableTarget(
+        name: "NIOResumableUploadDemo",
+        dependencies: [
+            "NIOResumableUpload",
+            "NIOHTTPTypesHTTP1",
+            .product(name: "HTTPTypes", package: "swift-http-types"),
+            .product(name: "NIOCore", package: "swift-nio"),
+            .product(name: "NIOPosix", package: "swift-nio"),
+        ]
+    ),
     .testTarget(
         name: "NIOResumableUploadTests",
         dependencies: [

Sources/NIOResumableUpload/HTTPResumableUpload.swift

@@ -2,7 +2,7 @@
 //
 // This source file is part of the SwiftNIO open source project
 //
-// Copyright (c) 2023 Apple Inc. and the SwiftNIO project authors
+// Copyright (c) 2023-2024 Apple Inc. and the SwiftNIO project authors
 // Licensed under Apache License v2.0
 //
 // See LICENSE.txt for license information

Sources/NIOResumableUpload/HTTPResumableUploadChannel.swift

@@ -2,7 +2,7 @@
 //
 // This source file is part of the SwiftNIO open source project
 //
-// Copyright (c) 2023 Apple Inc. and the SwiftNIO project authors
+// Copyright (c) 2023-2024 Apple Inc. and the SwiftNIO project authors
 // Licensed under Apache License v2.0
 //
 // See LICENSE.txt for license information

Sources/NIOResumableUpload/HTTPResumableUploadContext.swift

@@ -2,7 +2,7 @@
 //
 // This source file is part of the SwiftNIO open source project
 //
-// Copyright (c) 2023 Apple Inc. and the SwiftNIO project authors
+// Copyright (c) 2023-2024 Apple Inc. and the SwiftNIO project authors
 // Licensed under Apache License v2.0
 //
 // See LICENSE.txt for license information

Sources/NIOResumableUpload/HTTPResumableUploadHandler.swift

@@ -2,7 +2,7 @@
 //
 // This source file is part of the SwiftNIO open source project
 //
-// Copyright (c) 2023 Apple Inc. and the SwiftNIO project authors
+// Copyright (c) 2023-2024 Apple Inc. and the SwiftNIO project authors
 // Licensed under Apache License v2.0
 //
 // See LICENSE.txt for license information
@@ -24,7 +24,9 @@ public final class HTTPResumableUploadHandler: ChannelDuplexHandler {
     public typealias OutboundIn = Never
     public typealias OutboundOut = HTTPResponsePart
 
-    var upload: HTTPResumableUpload
+    var upload: HTTPResumableUpload? = nil
+    let createUpload: () -> HTTPResumableUpload
+    var shouldReset: Bool = false
 
     private var context: ChannelHandlerContext!
     private var eventLoop: EventLoop!
@@ -38,10 +40,12 @@ public final class HTTPResumableUploadHandler: ChannelDuplexHandler {
         context: HTTPResumableUploadContext,
         channelConfigurator: @escaping (Channel) -> Void
     ) {
-        self.upload = HTTPResumableUpload(
+        self.createUpload = {
-            context: context,
+            HTTPResumableUpload(
-            channelConfigurator: channelConfigurator
+                context: context,
-        )
+                channelConfigurator: channelConfigurator
+            )
+        }
     }
 
     /// Create an `HTTPResumableUploadHandler` within a given `HTTPResumableUploadContext`.
@@ -53,19 +57,31 @@ public final class HTTPResumableUploadHandler: ChannelDuplexHandler {
         context: HTTPResumableUploadContext,
         handlers: [ChannelHandler] = []
     ) {
-        self.upload = HTTPResumableUpload(context: context) { channel in
+        self.createUpload = {
-            if !handlers.isEmpty {
+            HTTPResumableUpload(context: context) { channel in
-                _ = channel.pipeline.addHandlers(handlers)
+                if !handlers.isEmpty {
+                    _ = channel.pipeline.addHandlers(handlers)
+                }
             }
         }
     }
 
+    private func resetUpload() {
+        if let existingUpload = self.upload {
+            existingUpload.end(handler: self, error: nil)
+        }
+        let upload = self.createUpload()
+        upload.scheduleOnEventLoop(self.eventLoop)
+        upload.attachUploadHandler(self, channel: self.context.channel)
+        self.upload = upload
+        self.shouldReset = false
+    }
+
     public func handlerAdded(context: ChannelHandlerContext) {
         self.context = context
         self.eventLoop = context.eventLoop
 
-        self.upload.scheduleOnEventLoop(context.eventLoop)
+        self.resetUpload()
-        self.upload.attachUploadHandler(self, channel: context.channel)
     }
 
     public func channelActive(context: ChannelHandlerContext) {
@@ -73,29 +89,38 @@ public final class HTTPResumableUploadHandler: ChannelDuplexHandler {
     }
 
     public func channelInactive(context: ChannelHandlerContext) {
-        self.upload.end(handler: self, error: nil)
+        self.upload?.end(handler: self, error: nil)
     }
 
     public func channelRead(context: ChannelHandlerContext, data: NIOAny) {
-        self.upload.receive(handler: self, channel: self.context.channel, part: unwrapInboundIn(data))
+        if self.shouldReset {
+            self.resetUpload()
+        }
+        let part = self.unwrapInboundIn(data)
+        if case .end = part {
+            self.shouldReset = true
+        }
+        self.upload?.receive(handler: self, channel: self.context.channel, part: part)
     }
 
     public func channelReadComplete(context: ChannelHandlerContext) {
-        self.upload.receiveComplete(handler: self)
+        self.upload?.receiveComplete(handler: self)
     }
 
     public func channelWritabilityChanged(context: ChannelHandlerContext) {
-        self.upload.writabilityChanged(handler: self)
+        self.upload?.writabilityChanged(handler: self)
     }
 
     public func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) {}
 
     public func errorCaught(context: ChannelHandlerContext, error: Error) {
-        self.upload.end(handler: self, error: error)
+        self.upload?.end(handler: self, error: error)
     }
 
     public func read(context: ChannelHandlerContext) {
-        // Do nothing.
+        if self.shouldReset {
+            context.read()
+        }
     }
 }
 

Sources/NIOResumableUpload/HTTPResumableUploadProtocol.swift

@@ -2,7 +2,7 @@
 //
 // This source file is part of the SwiftNIO open source project
 //
-// Copyright (c) 2023 Apple Inc. and the SwiftNIO project authors
+// Copyright (c) 2023-2024 Apple Inc. and the SwiftNIO project authors
 // Licensed under Apache License v2.0
 //
 // See LICENSE.txt for license information

Sources/NIOResumableUploadDemo/main.swift

@@ -0,0 +1,114 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) 2024 Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import Foundation
+import HTTPTypes
+import NIOCore
+import NIOHTTP1
+import NIOHTTPTypes
+import NIOHTTPTypesHTTP1
+import NIOPosix
+import NIOResumableUpload
+import System
+
+@available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
+final class UploadServerHandler: ChannelDuplexHandler {
+    typealias InboundIn = HTTPRequestPart
+    typealias OutboundIn = Never
+    typealias OutboundOut = HTTPResponsePart
+
+    let directory: FilePath
+    var fileHandle: FileHandle? = nil
+
+    init(directory: FilePath) {
+        self.directory = directory
+    }
+
+    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+        switch self.unwrapInboundIn(data) {
+        case .head(let request):
+            switch request.method {
+            case .post, .put:
+                if let requestPath = request.path {
+                    let path = self.directory.appending(requestPath)
+                    if let url = URL(path) {
+                        FileManager.default.createFile(atPath: path.string, contents: nil)
+                        self.fileHandle = try? FileHandle(forWritingTo: url)
+                        print("Writing to \(url)")
+                    }
+                }
+                if self.fileHandle == nil {
+                    let response = HTTPResponse(status: .internalServerError)
+                    self.write(context: context, data: self.wrapOutboundOut(.head(response)), promise: nil)
+                    self.write(context: context, data: self.wrapOutboundOut(.end(nil)), promise: nil)
+                    self.flush(context: context)
+                }
+            default:
+                let response = HTTPResponse(status: .notImplemented)
+                self.write(context: context, data: self.wrapOutboundOut(.head(response)), promise: nil)
+                self.write(context: context, data: self.wrapOutboundOut(.end(nil)), promise: nil)
+                self.flush(context: context)
+            }
+        case .body(let body):
+            do {
+                try body.withUnsafeReadableBytes { buffer in
+                    try fileHandle?.write(contentsOf: buffer)
+                }
+            } catch {
+                print("failed to write \(error)")
+                exit(1)
+            }
+        case .end:
+            if fileHandle != nil {
+                let response = HTTPResponse(status: .created)
+                self.write(context: context, data: self.wrapOutboundOut(.head(response)), promise: nil)
+                self.write(context: context, data: self.wrapOutboundOut(.end(nil)), promise: nil)
+                self.flush(context: context)
+            }
+        }
+    }
+}
+
+guard let outputFile = CommandLine.arguments.dropFirst().first else {
+    print("Usage: \(CommandLine.arguments[0]) <Upload Directory>")
+    exit(1)
+}
+
+if #available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *) {
+    let uploadContext = HTTPResumableUploadContext(origin: "http://localhost:8080")
+
+    let group = MultiThreadedEventLoopGroup(numberOfThreads: System.coreCount)
+    let server = try ServerBootstrap(group: group).childChannelInitializer { channel in
+        let handler = HTTP1ToHTTPServerCodec(secure: false)
+        return channel.pipeline.addHandlers([
+            handler,
+            HTTPResumableUploadHandler(
+                context: uploadContext,
+                handlers: [
+                    UploadServerHandler(directory: FilePath(CommandLine.arguments[1]))
+                ]
+            ),
+        ]).flatMap { _ in
+            channel.pipeline.configureHTTPServerPipeline(position: .before(handler))
+        }
+    }
+    .bind(host: "0.0.0.0", port: 8080)
+    .wait()
+
+    print("Listening on 8080")
+    try server.closeFuture.wait()
+} else {
+    print("Unsupported OS")
+    exit(1)
+}

Tests/NIOResumableUploadTests/NIOResumableUploadTests.swift

@@ -2,7 +2,7 @@
 //
 // This source file is part of the SwiftNIO open source project
 //
-// Copyright (c) 2023 Apple Inc. and the SwiftNIO project authors
+// Copyright (c) 2023-2024 Apple Inc. and the SwiftNIO project authors
 // Licensed under Apache License v2.0
 //
 // See LICENSE.txt for license information
@@ -25,11 +25,11 @@ private final class InboundRecorder<FrameIn, FrameOut>: ChannelDuplexHandler {
     typealias OutboundIn = Never
     typealias OutboundOut = FrameOut
 
-    private var context: ChannelHandlerContext? = nil
+    private var context: ChannelHandlerContext! = nil
 
     var receivedFrames: [FrameIn] = []
 
-    func channelActive(context: ChannelHandlerContext) {
+    func handlerAdded(context: ChannelHandlerContext) {
         self.context = context
     }
 
@@ -38,8 +38,8 @@ private final class InboundRecorder<FrameIn, FrameOut>: ChannelDuplexHandler {
     }
 
     func write(_ frame: FrameOut) {
-        self.write(context: self.context!, data: self.wrapOutboundOut(frame), promise: nil)
+        self.write(context: self.context, data: self.wrapOutboundOut(frame), promise: nil)
-        self.flush(context: self.context!)
+        self.flush(context: self.context)
     }
 }