refactor: migrate from Channel/Slot to PipelineState API

Update HTTP server, client, and WebSocket code to use the new
closure-based PipelineState from Reseau.

Key changes:
- Server: Connection.channel → Connection.pipeline, uses
  pipeline_shutdown!/pipeline_trigger_read/pipeline_schedule_task_now!
- Client stream: h1conn.slot → h1conn.pipeline, uses pipeline_write!
- WebSockets: WsChannelHandler → WsHandler (no longer extends
  AbstractChannelHandler), installs read handler via
  downstream_read_setter instead of slot manipulation
- utils: _H2_CHANNEL_SUPPORTED = true (H2Connection no longer extends
  AbstractChannelHandler but is supported via pipeline API)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: quinnj

src/client/stream.jl

@@ -172,23 +172,23 @@ end
 function _h1_flush_outgoing!(s::Stream)
     !isdefined(s, :aws_stream) && return
     h1conn = s.aws_stream.owning_connection
-    slot = h1conn.slot
-    slot === nothing && return
-    channel = slot.channel
-    channel === nothing && return
-    if !Reseau.Sockets.channel_thread_is_callers_thread(channel)
+    pipeline = h1conn.pipeline
+    pipeline === nothing && return
+    socket = pipeline.socket
+    socket === nothing && return
+    if !Reseau.Sockets.pipeline_thread_is_callers_thread(pipeline)
         fut = Future{Nothing}()
-        task = Reseau.Sockets.ChannelTask((task, ctx, status) -> begin
-            status == Reseau.TaskStatus.RUN_READY || return notify(fut, nothing)
+        task = Reseau.Sockets.ChannelTask(Reseau.EventCallable(status -> begin
+            Reseau.TaskStatus.T(status) == Reseau.TaskStatus.RUN_READY || return notify(fut, nothing)
             try
                 _h1_flush_outgoing!(s)
                 notify(fut, nothing)
             catch e
                 notify(fut, CapturedException(e, catch_backtrace()))
             end
             return nothing
-        end, nothing, "http_h1_flush_outgoing")
-        Reseau.Sockets.channel_schedule_task_now!(channel, task)
+        end), "http_h1_flush_outgoing")
+        Reseau.Sockets.pipeline_schedule_task_now!(pipeline, task)
         wait(fut)
         return
     end
@@ -203,10 +203,10 @@ function _h1_flush_outgoing!(s::Stream)
         end
         buf.len = Csize_t(length(encoded))
         try
-            Reseau.Sockets.channel_slot_send_message(slot, msg, Reseau.Sockets.ChannelDirection.WRITE)
+            Reseau.Sockets.pipeline_write!(socket, msg)
         catch e
             e isa Reseau.ReseauError || rethrow()
-            throw(AWSError("channel slot send failed"))
+            throw(AWSError("pipeline write failed"))
         end
     end
     return
@@ -857,8 +857,8 @@ function with_stream(conn, req::Request, chunkedbody, on_stream_response_body, d
             (@atomic stream.fut.set) != 0 && return
             if !stream.http2 && isdefined(stream, :connection)
                 conn = stream.connection
-                if conn !== nothing && conn.slot !== nothing && conn.slot.channel !== nothing
-                    Reseau.Sockets.channel_shutdown!(conn.slot.channel, AwsHTTP.ERROR_HTTP_RESPONSE_FIRST_BYTE_TIMEOUT; shutdown_immediately=true)
+                if conn !== nothing && conn.pipeline !== nothing
+                    Reseau.Sockets.pipeline_shutdown!(conn.pipeline, AwsHTTP.ERROR_HTTP_RESPONSE_FIRST_BYTE_TIMEOUT; shutdown_immediately=true)
                 elseif conn !== nothing
                     AwsHTTP.http_connection_close(conn)
                 end

src/server.jl

@@ -30,14 +30,14 @@ const _BACKLOG_DEFAULT = 511
 mutable struct Connection{S}
     const server::S # Server{F, C}
     const h1conn::Any # AwsHTTP.H1Connection or AwsHTTP.H2Connection
-    const channel::Any # Reseau.Channel
+    const pipeline::Any # Reseau.PipelineState
     const streams_lock::ReentrantLock
     const streams::Set{Stream}
     const remote_addr::String
     const remote_port_num::Int
 
-    Connection(server::S, h1conn, channel, remote_addr::String, remote_port_num::Int) where {S} =
-        new{S}(server, h1conn, channel, ReentrantLock(), Set{Stream}(), remote_addr, remote_port_num)
+    Connection(server::S, h1conn, pipeline, remote_addr::String, remote_port_num::Int) where {S} =
+        new{S}(server, h1conn, pipeline, ReentrantLock(), Set{Stream}(), remote_addr, remote_port_num)
 end
 
 Base.hash(c::Connection, h::UInt) = hash(objectid(c), h)
@@ -267,7 +267,7 @@ function _create_request_handler!(conn::Connection, aws_conn; http2::Bool=false)
                 end
             end
             if shutdown_channel
-                Reseau.Sockets.channel_shutdown!(conn.channel; shutdown_immediately=true)
+                Reseau.Sockets.pipeline_shutdown!(conn.pipeline; shutdown_immediately=true)
                 @lock server.connections_lock begin
                     delete!(server.connections, conn)
                 end
@@ -395,28 +395,26 @@ function serve!(f, host="127.0.0.1", port=8080;
     )
     alpn_list = _tls_alpn_list(tls_conn_opts)
     initial_window = Csize_t(min(UInt64(initial_window_size), UInt64(typemax(Csize_t))))
-    on_incoming_channel_setup = (bootstrap, error_code, channel, user_data) -> begin
+    on_incoming_channel_setup = (bootstrap, error_code, pipeline, user_data) -> begin
         Base.CoreLogging.with_logstate(server.logstate) do
             if error_code != 0
                 @error "incoming channel setup error" error_code
                 return
             end
             st = @atomic(server.state)
             if st == :closing || st == :closed
-                Reseau.Sockets.channel_shutdown!(channel; shutdown_immediately=true)
+                Reseau.Sockets.pipeline_shutdown!(pipeline; shutdown_immediately=true)
                 return
             end
-            slot = Reseau.Sockets.channel_slot_new!(channel)
-            Reseau.Sockets.channel_slot_insert_end!(channel, slot)
             version = AwsHTTP.HttpVersion.HTTP_1_1
             if tls_conn_opts !== nothing
-                tls_slot = slot.adj_left
-                if tls_slot === nothing || tls_slot.handler === nothing || !(tls_slot.handler isa Reseau.Sockets.TlsChannelHandler)
+                tls_handler = pipeline.tls_handler
+                if tls_handler === nothing
                     @error "incoming channel setup error" error_code=Reseau.ERROR_INVALID_STATE
-                    Reseau.Sockets.channel_shutdown!(channel, Reseau.ERROR_INVALID_STATE)
+                    Reseau.Sockets.pipeline_shutdown!(pipeline, Reseau.ERROR_INVALID_STATE)
                     return
                 end
-                protocol = Reseau.Sockets.tls_handler_protocol(tls_slot.handler)
+                protocol = Reseau.Sockets.tls_handler_protocol(tls_handler)
                 if protocol.len > 0
                     protocol_str = Reseau.byte_buffer_as_string(protocol)
                     if protocol_str == "h2"
@@ -426,26 +424,28 @@ function serve!(f, host="127.0.0.1", port=8080;
                     end
                 end
             end
-            http_conn = AwsHTTP.http_connection_new_channel_handler(;
+            http_conn = AwsHTTP.http_connection_new_handler(;
                 is_server=true,
                 version=version,
                 initial_window_size=initial_window,
             )
             http_conn === nothing && return
-            Reseau.Sockets.channel_slot_set_handler!(slot, http_conn)
-            http_conn.slot = slot
-            # Extract remote endpoint from the socket handler (first slot in pipeline)
+            if version == AwsHTTP.HttpVersion.HTTP_2
+                AwsHTTP.h2_connection_install!(http_conn, pipeline, pipeline.socket)
+            else
+                AwsHTTP.h1_connection_install!(http_conn, pipeline, pipeline.socket)
+            end
             remote_addr = "0.0.0.0"
             remote_port_num = 0
             try
-                socket_handler = channel.first.handler
-                ep = socket_handler.socket.remote_endpoint
+                socket = pipeline.socket
+                ep = socket.remote_endpoint
                 remote_addr = Reseau.Sockets.get_address(ep)
                 remote_port_num = Int(ep.port)
             catch
             end
             http_conn.remote_endpoint = "$remote_addr:$remote_port_num"
-            conn = Connection(server, http_conn, channel, remote_addr, remote_port_num)
+            conn = Connection(server, http_conn, pipeline, remote_addr, remote_port_num)
             @lock server.connections_lock begin
                 push!(server.connections, conn)
             end
@@ -470,26 +470,26 @@ function serve!(f, host="127.0.0.1", port=8080;
             else
                 _create_request_handler!(conn, http_conn; http2=false)
             end
-            if Reseau.Sockets.channel_thread_is_callers_thread(channel)
-                Reseau.Sockets.channel_trigger_read(channel)
+            if Reseau.Sockets.pipeline_thread_is_callers_thread(pipeline)
+                Reseau.Sockets.pipeline_trigger_read(pipeline.socket)
             else
                 task = Reseau.Sockets.ChannelTask(Reseau.EventCallable(status -> begin
                     Reseau.TaskStatus.T(status) == Reseau.TaskStatus.RUN_READY || return nothing
-                    Reseau.Sockets.channel_trigger_read(channel)
+                    Reseau.Sockets.pipeline_trigger_read(pipeline.socket)
                     return nothing
                 end), "http_server_trigger_read")
-                Reseau.Sockets.channel_schedule_task_now!(channel, task)
+                Reseau.Sockets.pipeline_schedule_task_now!(pipeline, task)
             end
         end
         return
     end
-    on_incoming_channel_shutdown = (bootstrap, error_code, channel, user_data) -> begin
+    on_incoming_channel_shutdown = (bootstrap, error_code, pipeline, user_data) -> begin
         Base.CoreLogging.with_logstate(server.logstate) do
             if _should_log_channel_shutdown_error(error_code)
                 @error "incoming channel shutdown error" error_code
             end
             @lock server.connections_lock begin
-                filter!(c -> c.channel !== channel, server.connections)
+                filter!(c -> c.pipeline !== pipeline, server.connections)
             end
         end
         return
@@ -633,7 +633,7 @@ function _forceclose!(server::Server; skip_shutdown::Bool=false)
         append!(conns, server.connections)
     end
     for conn in conns
-        Reseau.Sockets.channel_shutdown!(conn.channel; shutdown_immediately=true)
+        Reseau.Sockets.pipeline_shutdown!(conn.pipeline; shutdown_immediately=true)
     end
     @atomic server.state = :closed
     notify(server.closed)
@@ -658,7 +658,7 @@ function Base.close(server::Server)
         _stop_new_requests!(conn)
         @lock conn.streams_lock begin
             if isempty(conn.streams)
-                Reseau.Sockets.channel_shutdown!(conn.channel; shutdown_immediately=true)
+                Reseau.Sockets.pipeline_shutdown!(conn.pipeline; shutdown_immediately=true)
                 @lock server.connections_lock begin
                     delete!(server.connections, conn)
                 end

src/utils.jl

@@ -6,7 +6,7 @@ const HTTP2_MAX_WINDOW_SIZE = 0x7fffffff
 const AWS_HTTP2_SETTINGS_MAX_CONCURRENT_STREAMS = AwsHTTP.Http2SettingsId.MAX_CONCURRENT_STREAMS
 const AWS_HTTP2_SETTINGS_INITIAL_WINDOW_SIZE = AwsHTTP.Http2SettingsId.INITIAL_WINDOW_SIZE
 const AWS_HTTP2_SETTINGS_COUNT = Int(AwsHTTP.HTTP2_SETTINGS_END_RANGE - AwsHTTP.HTTP2_SETTINGS_BEGIN_RANGE)
-const _H2_CHANNEL_SUPPORTED = AwsHTTP.H2Connection <: Reseau.Sockets.AbstractChannelHandler
+const _H2_CHANNEL_SUPPORTED = true
 
 function _normalize_alpn_list(alpn_list::Union{String, Nothing})
     alpn_list === nothing && return nothing