Upload streaming (#37)

* Provide a way to stream Request.body Motivation: Users may want to optimize memory usage when executing request with big bodies, same as streaming download Modifications: Added ChunkProvider typealias in HTTPHandler.swift Added new Body enum case - stream in HTTPHandler.swift Extracted body processing in HTTPHandler.swift to a separate method Added .stream enum processing in HTTPHandler.swift Added upload streaming test to SwiftNIOHTTPTests.swift Result: HTTPClient library now provides methods to stream Request body

Author: artemredkin

README.md

@@ -116,26 +116,38 @@ class CountingDelegate: HTTPResponseDelegate{
 
 var count =0
 
-funcdidTransmitRequestBody(){
-// this is executed when request is sent, called once
+funcdidSendRequestHead(task: HTTPClient.Task<Response>, _head: HTTPRequestHead){
+// this is executed right after request head was sent, called once
  }
 
-funcdidReceiveHead(_head: HTTPResponseHead){
+funcdidSendRequestPart(task: HTTPClient.Task<Response>, _part: IOData){
+// this is executed when request body part is sent, could be called zero or more times
+ }
+
+funcdidSendRequest(task: HTTPClient.Task<Response>){
+// this is executed when request is fully sent, called once
+ }
+
+funcdidReceiveHead(task: HTTPClient.Task<Response>, _head: HTTPResponseHead) -> EventLoopFuture<Void>{
 // this is executed when we receive HTTP Reponse head part of the request (it contains response code and headers), called once
+// in case backpressure is needed, all reads will be paused until returned future is resolved
+return task.eventLoop.makeSucceededFuture(())
  }
 
-funcdidReceivePart(_buffer: ByteBuffer){
+funcdidReceivePart(task: HTTPClient.Task<Response>, _buffer: ByteBuffer)-> EventLoopFuture<Void>{
 // this is executed when we receive parts of the response body, could be called zero or more times
  count += buffer.readableBytes
+// in case backpressure is needed, all reads will be paused until returned future is resolved
+return task.eventLoop.makeSucceededFuture(())
  }
 
-funcdidFinishRequest() throws->Int{
+funcdidFinishRequest(task: HTTPClient.Task<Response>) throws->Int{
 // this is called when the request is fully read, called once
 // this is where you return a result or throw any errors you require to propagate to the client
 return count
  }
 
-funcdidReceiveError(_error: Error){
+funcdidReceiveError(task: HTTPClient.Task<Response>, _error: Error){
 // this is called when we receive any network-related error, called once
  }
 }

Sources/NIOHTTPClient/HTTPHandler.swift

@@ -18,25 +18,48 @@ import NIOConcurrencyHelpers
 import NIOHTTP1
 import NIOSSL
 
-publicextensionHTTPClient{
-enumBody:Equatable{
-case byteBuffer(ByteBuffer)
-case data(Data)
-case string(String)
-
-varlength:Int{
-switchself{
-case.byteBuffer(let buffer):
-return buffer.readableBytes
-case.data(let data):
-return data.count
-case.string(let string):
-return string.utf8.count
+extensionHTTPClient{
+
+publicstructBody{
+publicstructStreamWriter{
+letclosure:(IOData)->EventLoopFuture<Void>
+
+publicfunc write(_ data:IOData)->EventLoopFuture<Void>{
+returnself.closure(data)
+}
+}
+
+publicvarlength:Int?
+publicvarstream:(StreamWriter)->EventLoopFuture<Void>
+
+publicstaticfunc byteBuffer(_ buffer:ByteBuffer)->Body{
+returnBody(length: buffer.readableBytes){ writer in
+ writer.write(.byteBuffer(buffer))
+}
+}
+
+publicstaticfunc stream(length:Int?=nil, _ stream:@escaping(StreamWriter)->EventLoopFuture<Void>)->Body{
+returnBody(length: length, stream: stream)
+}
+
+publicstaticfunc data(_ data:Data)->Body{
+returnBody(length: data.count){ writer in
+varbuffer=ByteBufferAllocator().buffer(capacity: data.count)
+ buffer.writeBytes(data)
+return writer.write(.byteBuffer(buffer))
+}
+}
+
+publicstaticfunc string(_ string:String)->Body{
+returnBody(length: string.utf8.count){ writer in
+varbuffer=ByteBufferAllocator().buffer(capacity: string.utf8.count)
+ buffer.writeString(string)
+return writer.write(.byteBuffer(buffer))
 }
 }
 }
 
-structRequest:Equatable{
+publicstructRequest{
 publicvarversion:HTTPVersion
 publicvarmethod:HTTPMethod
 publicvarurl:URL
@@ -53,7 +76,7 @@ public extension HTTPClient{
 tryself.init(url: url, version: version, method: method, headers: headers, body: body)
 }
 
-publicinit(url:URL, version:HTTPVersion, method:HTTPMethod=.GET, headers:HTTPHeaders=HTTPHeaders(), body:Body?=nil)throws{
+publicinit(url:URL, version:HTTPVersion=HTTPVersion(major:1, minor:1), method:HTTPMethod=.GET, headers:HTTPHeaders=HTTPHeaders(), body:Body?=nil)throws{
 guardlet scheme = url.scheme else{
 throwHTTPClientError.emptyScheme
 }
@@ -88,7 +111,7 @@ public extension HTTPClient{
 }
 }
 
-structResponse:Equatable{
+publicstructResponse{
 publicvarhost:String
 publicvarstatus:HTTPResponseStatus
 publicvarheaders:HTTPHeaders
@@ -114,9 +137,7 @@ internal class ResponseAccumulator: HTTPClientResponseDelegate{
 self.request = request
 }
 
-func didTransmitRequestBody(task:HTTPClient.Task<Response>){}
-
-func didReceiveHead(task:HTTPClient.Task<Response>, _ head:HTTPResponseHead){
+func didReceiveHead(task:HTTPClient.Task<Response>, _ head:HTTPResponseHead)->EventLoopFuture<Void>{
 switchself.state {
 case.idle:
 self.state =.head(head)
@@ -129,9 +150,10 @@ internal class ResponseAccumulator: HTTPClientResponseDelegate{
 case.error:
 break
 }
+return task.eventLoop.makeSucceededFuture(())
 }
 
-func didReceivePart(task:HTTPClient.Task<Response>, _ part:ByteBuffer){
+func didReceivePart(task:HTTPClient.Task<Response>, _ part:ByteBuffer)->EventLoopFuture<Void>{
 switchself.state {
 case.idle:
 preconditionFailure("no head received before body")
@@ -146,6 +168,7 @@ internal class ResponseAccumulator: HTTPClientResponseDelegate{
 case.error:
 break
 }
+return task.eventLoop.makeSucceededFuture(())
 }
 
 func didReceiveError(task:HTTPClient.Task<Response>, _ error:Error){
@@ -174,25 +197,33 @@ internal class ResponseAccumulator: HTTPClientResponseDelegate{
 publicprotocolHTTPClientResponseDelegate:AnyObject{
 associatedtypeResponse
 
-func didTransmitRequestBody(task:HTTPClient.Task<Response>)
+func didSendRequestHead(task:HTTPClient.Task<Response>, _ head:HTTPRequestHead)
+
+func didSendRequestPart(task:HTTPClient.Task<Response>, _ part:IOData)
+
+func didSendRequest(task:HTTPClient.Task<Response>)
 
-func didReceiveHead(task:HTTPClient.Task<Response>, _ head:HTTPResponseHead)
+func didReceiveHead(task:HTTPClient.Task<Response>, _ head:HTTPResponseHead)->EventLoopFuture<Void>
 
-func didReceivePart(task:HTTPClient.Task<Response>, _ buffer:ByteBuffer)
+func didReceivePart(task:HTTPClient.Task<Response>, _ buffer:ByteBuffer)->EventLoopFuture<Void>
 
 func didReceiveError(task:HTTPClient.Task<Response>, _ error:Error)
 
 func didFinishRequest(task:HTTPClient.Task<Response>)throws->Response
 }
 
 extensionHTTPClientResponseDelegate{
-funcdidTransmitRequestBody(task:HTTPClient.Task<Response>){}
+publicfuncdidSendRequestHead(task:HTTPClient.Task<Response>, _ head:HTTPRequestHead){}
 
-funcdidReceiveHead(task:HTTPClient.Task<Response>, _:HTTPResponseHead){}
+publicfuncdidSendRequestPart(task:HTTPClient.Task<Response>, _ part:IOData){}
 
-funcdidReceivePart(task:HTTPClient.Task<Response>, _:ByteBuffer){}
+publicfuncdidSendRequest(task:HTTPClient.Task<Response>){}
 
-func didReceiveError(task:HTTPClient.Task<Response>, _:Error){}
+publicfunc didReceiveHead(task:HTTPClient.Task<Response>, _:HTTPResponseHead)->EventLoopFuture<Void>{return task.eventLoop.makeSucceededFuture(())}
+
+publicfunc didReceivePart(task:HTTPClient.Task<Response>, _:ByteBuffer)->EventLoopFuture<Void>{return task.eventLoop.makeSucceededFuture(())}
+
+publicfunc didReceiveError(task:HTTPClient.Task<Response>, _:Error){}
 }
 
 internalextensionURL{
@@ -207,13 +238,15 @@ internal extension URL{
 
 publicextensionHTTPClient{
 finalclassTask<Response>{
+publicleteventLoop:EventLoop
 letfuture:EventLoopFuture<Response>
 
 privatevarchannel:Channel?
 privatevarcancelled:Bool
 privateletlock:Lock
 
-init(future:EventLoopFuture<Response>){
+init(eventLoop:EventLoop, future:EventLoopFuture<Response>){
+self.eventLoop = eventLoop
 self.future = future
 self.cancelled =false
 self.lock =Lock()
@@ -267,6 +300,8 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler
 letredirectHandler:RedirectHandler<T.Response>?
 
 varstate:State=.idle
+varpendingRead=false
+varmayRead=true
 
 init(task:HTTPClient.Task<T.Response>, delegate:T, promise:EventLoopPromise<T.Response>, redirectHandler:RedirectHandler<T.Response>?){
 self.task = task
@@ -298,35 +333,52 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler
 
  head.headers = headers
 
- context.write(wrapOutboundOut(.head(head)), promise:nil)
+ context.write(wrapOutboundOut(.head(head))).whenSuccess{
+self.delegate.didSendRequestHead(task:self.task, head)
+}
 
-iflet body = request.body {
-letpart:HTTPClientRequestPart
-switch body {
-case.byteBuffer(let buffer):
- part =HTTPClientRequestPart.body(.byteBuffer(buffer))
-case.data(let data):
-varbuffer= context.channel.allocator.buffer(capacity: data.count)
- buffer.writeBytes(data)
- part =HTTPClientRequestPart.body(.byteBuffer(buffer))
-case.string(let string):
-varbuffer= context.channel.allocator.buffer(capacity: string.utf8.count)
- buffer.writeString(string)
- part =HTTPClientRequestPart.body(.byteBuffer(buffer))
-}
+self.writeBody(request: request, context: context).whenComplete{ result in
+switch result {
+case.success:
+ context.write(self.wrapOutboundOut(.end(nil)), promise: promise)
+ context.flush()
 
-context.write(wrapOutboundOut(part), promise:nil)
-}
+self.state =.sent
+self.delegate.didSendRequest(task:self.task)
 
- context.write(wrapOutboundOut(.end(nil)), promise: promise)
- context.flush()
+letchannel= context.channel
+self.promise.futureResult.whenComplete{ _ in
+ channel.close(promise:nil)
+}
+case.failure(let error):
+self.state =.end
+self.delegate.didReceiveError(task:self.task, error)
+self.promise.fail(error)
+ context.close(promise:nil)
+}
+}
+}
 
-self.state =.sent
-self.delegate.didTransmitRequestBody(task:self.task)
+privatefunc writeBody(request:HTTPClient.Request, context:ChannelHandlerContext)->EventLoopFuture<Void>{
+iflet body = request.body {
+return body.stream(HTTPClient.Body.StreamWriter{ part in
+letfuture= context.writeAndFlush(self.wrapOutboundOut(.body(part)))
+ future.whenSuccess{ _ in
+self.delegate.didSendRequestPart(task:self.task, part)
+}
+return future
+})
+}else{
+return context.eventLoop.makeSucceededFuture(())
+}
+}
 
-letchannel= context.channel
-self.promise.futureResult.whenComplete{ _ in
- channel.close(promise:nil)
+publicfunc read(context:ChannelHandlerContext){
+ifself.mayRead {
+self.pendingRead =false
+ context.read()
+}else{
+self.pendingRead =true
 }
 }
 
@@ -338,15 +390,21 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler
 self.state =.redirected(head, redirectURL)
 }else{
 self.state =.head
-self.delegate.didReceiveHead(task:self.task, head)
+self.mayRead =false
+self.delegate.didReceiveHead(task:self.task, head).whenComplete{ result in
+self.handleBackpressureResult(context: context, result: result)
+}
 }
 case.body(let body):
 switchself.state {
 case.redirected:
 break
 default:
 self.state =.body
-self.delegate.didReceivePart(task:self.task, body)
+self.mayRead =false
+self.delegate.didReceivePart(task:self.task, body).whenComplete{ result in
+self.handleBackpressureResult(context: context, result: result)
+}
 }
 case.end:
 switchself.state {
@@ -365,6 +423,20 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler
 }
 }
 
+privatefunc handleBackpressureResult(context:ChannelHandlerContext, result:Result<Void,Error>){
+switch result {
+case.success:
+self.mayRead =true
+ifself.pendingRead {
+ context.read()
+}
+case.failure(let error):
+self.state =.end
+self.delegate.didReceiveError(task:self.task, error)
+self.promise.fail(error)
+}
+}
+
 func userInboundEventTriggered(context:ChannelHandlerContext, event:Any){
 if(event as?IdleStateHandler.IdleStateEvent)==.read {
 self.state =.end

Sources/NIOHTTPClient/RequestValidation.swift

@@ -35,11 +35,17 @@ extension HTTPHeaders{
 }
 
 if encodings.isEmpty {
- contentLength = body.length
+guardlet length = body.length else{
+throwHTTPClientError.contentLengthMissing
+}
+ contentLength = length
 }else{
  transferEncoding = encodings.joined(separator:", ")
 if !encodings.contains("chunked"){
- contentLength = body.length
+guardlet length = body.length else{
+throwHTTPClientError.contentLengthMissing
+}
+ contentLength = length
 }
 }
 }else{

Sources/NIOHTTPClient/SwiftNIOHTTP.swift

@@ -110,7 +110,8 @@ public class HTTPClient{
 
 publicfunc execute<T:HTTPClientResponseDelegate>(request:Request, delegate:T, timeout:Timeout?=nil)->Task<T.Response>{
 lettimeout= timeout ?? configuration.timeout
-letpromise:EventLoopPromise<T.Response>=self.eventLoopGroup.next().makePromise()
+leteventLoop=self.eventLoopGroup.next()
+letpromise:EventLoopPromise<T.Response>= eventLoop.makePromise()
 
 letredirectHandler:RedirectHandler<T.Response>?
 ifself.configuration.followRedirects {
@@ -121,7 +122,7 @@ public class HTTPClient{
  redirectHandler =nil
 }
 
-lettask=Task(future: promise.futureResult)
+lettask=Task(eventLoop: eventLoop,future: promise.futureResult)
 
 varbootstrap=ClientBootstrap(group:self.eventLoopGroup)
 .channelOption(ChannelOptions.socket(SocketOptionLevel(IPPROTO_TCP), TCP_NODELAY), value:1)
@@ -256,6 +257,7 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible{
 case identityCodingIncorrectlyPresent
 case chunkedSpecifiedMultipleTimes
 case invalidProxyResponse
+case contentLengthMissing
 }
 
 privatevarcode:Code
@@ -279,4 +281,5 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible{
 publicstaticletidentityCodingIncorrectlyPresent=HTTPClientError(code:.identityCodingIncorrectlyPresent)
 publicstaticletchunkedSpecifiedMultipleTimes=HTTPClientError(code:.chunkedSpecifiedMultipleTimes)
 publicstaticletinvalidProxyResponse=HTTPClientError(code:.invalidProxyResponse)
+publicstaticletcontentLengthMissing=HTTPClientError(code:.contentLengthMissing)
 }