in Sources/NIOHTTP2/StreamStateMachine.swift [600:669]
mutating func receiveData(contentLength: Int, flowControlledBytes: Int, isEndStreamSet endStream: Bool) -> StateMachineResultWithStreamEffect {
do {
// We can receive DATA frames in the following states:
//
// - halfOpenRemoteLocalIdle, in which case we are a server receiving request data before we have
// sent our final response headers.
// - fullyOpen, where we could be either a client or a server using a fully bi-directional stream.
// - halfClosedLocalPeerActive, whe have completed our data, but the remote peer has more to send.
switch self.state {
case .halfOpenRemoteLocalIdle(localWindow: let localWindow, remoteContentLength: var remoteContentLength, remoteWindow: var remoteWindow):
try remoteWindow.consume(flowControlledBytes: flowControlledBytes)
try remoteContentLength.receivedDataChunk(length: contentLength)
let effect: StreamStateChange
if endStream {
try remoteContentLength.endOfStream()
self.state = .halfClosedRemoteLocalIdle(localWindow: localWindow)
effect = .windowSizeChange(.init(streamID: self.streamID, localStreamWindowSize: Int(localWindow), remoteStreamWindowSize: nil))
} else {
self.state = .halfOpenRemoteLocalIdle(localWindow: localWindow, remoteContentLength: remoteContentLength, remoteWindow: remoteWindow)
effect = .windowSizeChange(.init(streamID: self.streamID, localStreamWindowSize: Int(localWindow), remoteStreamWindowSize: Int(remoteWindow)))
}
return .init(result: .succeed, effect: effect)
case .fullyOpen(let localRole, localContentLength: let localContentLength, remoteContentLength: var remoteContentLength, localWindow: let localWindow, remoteWindow: var remoteWindow):
try remoteWindow.consume(flowControlledBytes: flowControlledBytes)
try remoteContentLength.receivedDataChunk(length: contentLength)
let effect: StreamStateChange
if endStream {
try remoteContentLength.endOfStream()
self.state = .halfClosedRemoteLocalActive(localRole: localRole, initiatedBy: .client, localContentLength: localContentLength, localWindow: localWindow)
effect = .windowSizeChange(.init(streamID: self.streamID, localStreamWindowSize: Int(localWindow), remoteStreamWindowSize: nil))
} else {
self.state = .fullyOpen(localRole: localRole, localContentLength: localContentLength, remoteContentLength: remoteContentLength, localWindow: localWindow, remoteWindow: remoteWindow)
effect = .windowSizeChange(.init(streamID: self.streamID, localStreamWindowSize: Int(localWindow), remoteStreamWindowSize: Int(remoteWindow)))
}
return .init(result: .succeed, effect: effect)
case .halfClosedLocalPeerActive(let localRole, let initiatedBy, var remoteContentLength, var remoteWindow):
try remoteWindow.consume(flowControlledBytes: flowControlledBytes)
try remoteContentLength.receivedDataChunk(length: contentLength)
let effect: StreamStateChange
if endStream {
try remoteContentLength.endOfStream()
self.state = .closed(reason: nil)
effect = .streamClosed(.init(streamID: self.streamID, reason: nil))
} else {
self.state = .halfClosedLocalPeerActive(localRole: localRole, initiatedBy: initiatedBy, remoteContentLength: remoteContentLength, remoteWindow: remoteWindow)
effect = .windowSizeChange(.init(streamID: self.streamID, localStreamWindowSize: nil, remoteStreamWindowSize: Int(remoteWindow)))
}
return .init(result: .succeed, effect: effect)
// Receiving a DATA frame outside any of these states is a stream error of type STREAM_CLOSED (RFC7540 § 6.1)
case .idle, .halfOpenLocalPeerIdle, .reservedLocal, .reservedRemote, .halfClosedLocalPeerIdle,
.halfClosedRemoteLocalActive, .halfClosedRemoteLocalIdle, .closed:
return .init(result: .streamError(streamID: self.streamID, underlyingError: NIOHTTP2Errors.badStreamStateTransition(from: NIOHTTP2StreamState.get(self.state)), type: .streamClosed), effect: nil)
}
} catch let error where error is NIOHTTP2Errors.FlowControlViolation {
return .init(result: .streamError(streamID: self.streamID, underlyingError: error, type: .flowControlError), effect: nil)
} catch let error where error is NIOHTTP2Errors.ContentLengthViolated {
return .init(result: .streamError(streamID: self.streamID, underlyingError: error, type: .protocolError), effect: nil)
} catch {
preconditionFailure("Unexpected error: \(error)")
}
}