Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

support stream resets through user events #488

Closed
wants to merge 1 commit into from
Closed

support stream resets through user events #488

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 19 additions & 4 deletions Sources/NIOHTTP2/HTTP2StreamChannel.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -605,6 +605,10 @@ final class HTTP2StreamChannel: Channel, ChannelCore, @unchecked Sendable {
}

public func close0(error: Error, mode: CloseMode, promise: EventLoopPromise<Void>?) {
self.closeWithCode(code: .cancel, promise: promise)
}

func closeWithCode(code: HTTP2ErrorCode, promise: EventLoopPromise<Void>?) {
// If the stream is already closed, we can fail this early and abort processing. If it's not, we need to emit a
// RST_STREAM frame.
guard self.state != .closed else {
Expand All @@ -627,12 +631,17 @@ final class HTTP2StreamChannel: Channel, ChannelCore, @unchecked Sendable {
self.closedCleanly()
case .remoteActive, .active, .closing, .closingNeverActivated:
// In all of these states the stream is still on the network and we need to wait.
self.closedWhileOpen()
self.closedWhileOpen(code: code)
}
}

public func triggerUserOutboundEvent0(_ event: Any, promise: EventLoopPromise<Void>?) {
// do nothing
switch event {
case .reset(let code) as HTTP2StreamChannelEvent:
self.closeWithCode(code: code, promise: promise)
default:
break
}
}

public func channelRead0(_ data: NIOAny) {
Expand All @@ -647,7 +656,7 @@ final class HTTP2StreamChannel: Channel, ChannelCore, @unchecked Sendable {
///
/// Will emit a RST_STREAM frame in order to close the stream. Note that this function does not
/// directly close the stream: it waits until the stream closed notification is fired.
private func closedWhileOpen() {
private func closedWhileOpen(code: HTTP2ErrorCode = .cancel) {
precondition(self.state != .closed)
guard self.state != .closing else {
// If we're already closing, nothing to do here.
Expand All @@ -656,7 +665,7 @@ final class HTTP2StreamChannel: Channel, ChannelCore, @unchecked Sendable {

self.modifyingState { $0.beginClosing() }
// We should have a stream ID here, force-unwrap is safe.
let resetFrame = HTTP2Frame(streamID: self.streamID!, payload: .rstStream(.cancel))
let resetFrame = HTTP2Frame(streamID: self.streamID!, payload: .rstStream(code))
self.receiveOutboundFrame(resetFrame, promise: nil)
self.multiplexer.flushStream(self.streamID!)
}
Expand Down Expand Up @@ -998,3 +1007,9 @@ extension HTTP2StreamChannel {
SynchronousOptions(channel: self)
}
}

/// Events that can be sent by the application to be handled by the `HTTP2StreamChannel`
public enum HTTP2StreamChannelEvent {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So I think we'll probably end up regressing making this a public enum, as we can't easily evolve that as an API. I think we should store the "real" value in an enum, but have this wrapped in a struct that has a few static func constructors..

We'll also need tests for this.

/// Send a `RST_STREAM` with the specified code
case reset(HTTP2ErrorCode)
}
Loading