Skip to content

refactor: Move session stop reason creation into Session constructor #377

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

Merged
minghuaw merged 7 commits into from
Aug 17, 2026
Merged

refactor: Move session stop reason creation into Session constructor #377

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
4d9dbb2
propagate session and connection stop reasons to links and acceptors
minghuaw Aug 16, 2026
e3cba57
surface session stop reasons on link operations during teardown
minghuaw Aug 16, 2026
04e131b
replace IllegalConnectionState with ConnectionStopped(ConnectionStopR…
minghuaw Aug 16, 2026
08e733a
differentiate local and remote stop reasons and fix close-exchange te…
minghuaw Aug 16, 2026
7cd404c
create session stop reason cell in the builder and share via getter
minghuaw Aug 16, 2026
5a72cf3
Merge branch 'main' into fix/stop-reason-propagation
minghuaw Aug 16, 2026
92a961d
cleanup
minghuaw Aug 16, 2026
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
11 changes: 4 additions & 7 deletions fe2o3-amqp/src/acceptor/session.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -287,18 +287,15 @@ impl SessionAcceptor {
}
};

// The stop reason cells are created here so that the session object, the
// engine, and the handle share them: the engine publishes the reason on
// the session at exit, and the handle (and the links attached through
// it) can read it.
let session_stop_reason = Arc::new(OnceLock::new());

// `into_session` creates the shared stop-reason cell; the engine
// publishes the reason on the session at exit, and the handle (and the
// links attached through it) read the same cell via a clone.
let mut session = self.0.clone().into_session(
outgoing_channel,
local_state,
session_stop_reason.clone(),
connection.connection_stop_reason.clone(),
);
let session_stop_reason = session.session_stop_reason().clone();
session.on_incoming_begin(
IncomingChannel(incoming_session.channel),
incoming_session.begin,
Expand Down
66 changes: 32 additions & 34 deletions fe2o3-amqp/src/session/builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,7 @@ use tokio::sync::mpsc;
use crate::{
connection::{AllocSessionError, ConnectionHandle, ConnectionStopReason},
control::SessionControl,
endpoint::OutgoingChannel,
link::SessionStopReason,
endpoint::{OutgoingChannel, Session as _},
session::{engine::SessionEngine, SessionState},
util::Constant,
Session,
Expand Down Expand Up @@ -91,14 +90,13 @@ cfg_transaction! {
outgoing_channel: OutgoingChannel,
control_link_acceptor: ControlLinkAcceptor,
local_state: SessionState,
session_stop_reason: Arc<OnceLock<SessionStopReason>>,
connection_stop_reason: Arc<OnceLock<ConnectionStopReason>>,
) -> TxnSession<Session> {
let txn_manager = TransactionManager::new(outgoing, control_link_acceptor);
let session = Session {
// control,
outgoing_channel,
session_stop_reason,
session_stop_reason: Arc::new(OnceLock::new()),
connection_stop_reason,
local_state,
initial_outgoing_id: Constant::new(self.next_outgoing_id),
Expand Down Expand Up @@ -139,15 +137,13 @@ impl Builder {

pub(crate) fn into_session(
self,
// control: mpsc::Sender<SessionControl>,
outgoing_channel: OutgoingChannel,
local_state: SessionState,
session_stop_reason: Arc<OnceLock<SessionStopReason>>,
connection_stop_reason: Arc<OnceLock<ConnectionStopReason>>,
) -> Session {
Session {
outgoing_channel,
session_stop_reason,
session_stop_reason: Arc::new(OnceLock::new()),
connection_stop_reason,
local_state,
initial_outgoing_id: Constant::new(self.next_outgoing_id),
Expand Down Expand Up @@ -271,7 +267,6 @@ impl Builder {
mpsc::channel::<SessionControl>(DEFAULT_SESSION_CONTROL_BUFFER_SIZE);
let (incoming_tx, incoming_rx) = mpsc::channel(self.buffer_size);
let (outgoing_tx, outgoing_rx) = mpsc::channel(self.buffer_size);
let session_stop_reason = Arc::new(OnceLock::new());

// create session in connection::Engine
let outgoing_channel = match connection.allocate_session(incoming_tx).await {
Expand All @@ -286,27 +281,28 @@ impl Builder {
};

#[cfg(not(all(feature = "transaction", feature = "acceptor")))]
let (engine_handle, outcome) = {
let (engine_handle, outcome, session_stop_reason) = {
let session = self.into_session(
outgoing_channel,
local_state,
session_stop_reason.clone(),
connection.connection_stop_reason.clone(),
);
let engine = SessionEngine::begin_client_session(
let session_stop_reason = session.session_stop_reason().clone();
let (handle, outcome) = SessionEngine::begin_client_session(
connection.control.clone(),
session,
session_control_rx,
incoming_rx,
connection.outgoing.clone(),
outgoing_rx,
)
.await?;
engine.spawn()
.await?
.spawn();
(handle, outcome, session_stop_reason)
};

#[cfg(all(feature = "transaction", feature = "acceptor"))]
let (engine_handle, outcome) = {
let (engine_handle, outcome, session_stop_reason) = {
let mut this = self;
match this.control_link_acceptor.take() {
Some(control_link_acceptor) => {
Expand All @@ -316,37 +312,39 @@ impl Builder {
outgoing_channel,
control_link_acceptor,
local_state,
session_stop_reason.clone(),
connection.connection_stop_reason.clone(),
);
let engine = SessionEngine::begin_client_session(
let session_stop_reason = session.session_stop_reason().clone();
let (handle, outcome) = SessionEngine::begin_client_session(
connection.control.clone(),
session,
session_control_rx,
incoming_rx,
connection.outgoing.clone(),
outgoing_rx,
)
.await?;
engine.spawn()
.await?
.spawn();
(handle, outcome, session_stop_reason)
}
None => {
let session = this.into_session(
outgoing_channel,
local_state,
session_stop_reason.clone(),
connection.connection_stop_reason.clone(),
);
let engine = SessionEngine::begin_client_session(
let session_stop_reason = session.session_stop_reason().clone();
let (handle, outcome) = SessionEngine::begin_client_session(
connection.control.clone(),
session,
session_control_rx,
incoming_rx,
connection.outgoing.clone(),
outgoing_rx,
)
.await?;
engine.spawn()
.await?
.spawn();
(handle, outcome, session_stop_reason)
}
}
};
Expand Down Expand Up @@ -386,7 +384,6 @@ impl Builder {
mpsc::channel::<SessionControl>(DEFAULT_SESSION_CONTROL_BUFFER_SIZE);
let (incoming_tx, incoming_rx) = mpsc::channel(self.buffer_size);
let (outgoing_tx, outgoing_rx) = mpsc::channel(self.buffer_size);
let session_stop_reason = Arc::new(OnceLock::new());

// create session in connection::Engine
let outgoing_channel = match connection.allocate_session(incoming_tx).await {
Expand All @@ -400,23 +397,24 @@ impl Builder {
},
};

let (engine_handle, outcome) = {
let (engine_handle, outcome, session_stop_reason) = {
let session = self.into_session(
outgoing_channel,
local_state,
session_stop_reason.clone(),
connection.connection_stop_reason.clone(),
);
let engine = SessionEngine::begin_client_session(
let session_stop_reason = session.session_stop_reason().clone();
let (handle, outcome) = SessionEngine::begin_client_session(
connection.control.clone(),
session,
session_control_rx,
incoming_rx,
connection.outgoing.clone(),
outgoing_rx,
)
.await?;
engine.spawn_on_local_set(local_set)
.await?
.spawn_on_local_set(local_set);
(handle, outcome, session_stop_reason)
};

let handle = SessionHandle {
Expand Down Expand Up @@ -453,7 +451,6 @@ impl Builder {
mpsc::channel::<SessionControl>(DEFAULT_SESSION_CONTROL_BUFFER_SIZE);
let (incoming_tx, incoming_rx) = mpsc::channel(self.buffer_size);
let (outgoing_tx, outgoing_rx) = mpsc::channel(self.buffer_size);
let session_stop_reason = Arc::new(OnceLock::new());

// create session in connection::Engine
let outgoing_channel = match connection.allocate_session(incoming_tx).await {
Expand All @@ -467,23 +464,24 @@ impl Builder {
},
};

let (engine_handle, outcome) = {
let (engine_handle, outcome, session_stop_reason) = {
let session = self.into_session(
outgoing_channel,
local_state,
session_stop_reason.clone(),
connection.connection_stop_reason.clone(),
);
let engine = SessionEngine::begin_client_session(
let session_stop_reason = session.session_stop_reason().clone();
let (handle, outcome) = SessionEngine::begin_client_session(
connection.control.clone(),
session,
session_control_rx,
incoming_rx,
connection.outgoing.clone(),
outgoing_rx,
)
.await?;
engine.spawn_local()
.await?
.spawn_local();
(handle, outcome, session_stop_reason)
};

let handle = SessionHandle {
Expand Down
Loading