some public API cleanup

Keruspe · Keruspe · commit 9b808732c3ee · 2025-08-15T17:07:24.000+02:00
diff --git a/examples/client-certificate.rs b/examples/client-certificate.rs
@@ -1,8 +1,7 @@
 use lapin::{
-    BasicProperties, Connection, ConnectionProperties,
+    BasicProperties, Confirmation, Connection, ConnectionProperties,
     message::DeliveryResult,
     options::*,
-    publisher_confirm::Confirmation,
     tcp::{OwnedIdentity, OwnedTLSConfig},
     types::FieldTable,
 };
@@ -48,7 +47,7 @@ async fn main() {
     let channel_a = conn.create_channel().await.expect("create_channel");
     //receive channel
     let channel_b = conn.create_channel().await.expect("create_channel");
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
 
     //create the hello queue
     let queue = channel_a
@@ -59,7 +58,7 @@ async fn main() {
         )
         .await
         .expect("queue_declare");
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
     info!(?queue, "Declared queue");
 
     info!("will consume");
@@ -81,7 +80,7 @@ async fn main() {
                     .expect("basic_ack");
             }
         });
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
 
     info!("will publish");
     let payload = b"Hello world!";
@@ -98,5 +97,5 @@ async fn main() {
         .await
         .expect("publisher-confirms");
     assert_eq!(confirm, Confirmation::NotRequested);
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
 }
diff --git a/examples/connection.rs b/examples/connection.rs
@@ -1,6 +1,6 @@
 use lapin::{
-    BasicProperties, Connection, ConnectionProperties, message::DeliveryResult, options::*,
-    publisher_confirm::Confirmation, types::FieldTable,
+    BasicProperties, Confirmation, Connection, ConnectionProperties, message::DeliveryResult,
+    options::*, types::FieldTable,
 };
 use tracing::info;
 
@@ -25,7 +25,7 @@ fn main() {
             let channel_a = conn.create_channel().await.expect("create_channel");
             //receive channel
             let channel_b = conn.create_channel().await.expect("create_channel");
-            info!(state=?conn.status().state());
+            info!(state=?conn.status());
 
             //create the hello queue
             let queue = channel_a
@@ -36,7 +36,7 @@ fn main() {
                 )
                 .await
                 .expect("queue_declare");
-            info!(state=?conn.status().state());
+            info!(state=?conn.status());
             info!(?queue, "Declared queue");
 
             info!("will consume");
@@ -66,7 +66,7 @@ fn main() {
                         }
                     }
                 });
-            info!(state=?conn.status().state());
+            info!(state=?conn.status());
 
             info!("will publish");
             let payload = b"Hello world!";
@@ -83,7 +83,7 @@ fn main() {
                 .await
                 .expect("publisher-confirms");
             assert_eq!(confirm, Confirmation::NotRequested);
-            info!(state=?conn.status().state());
+            info!(state=?conn.status());
         }
 
         conn.run().expect("conn.run");
diff --git a/examples/consumer.rs b/examples/consumer.rs
@@ -20,7 +20,7 @@ async fn main() {
 
     //receive channel
     let channel = conn.create_channel().await.expect("create_channel");
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
 
     let queue = channel
         .queue_declare(
@@ -30,7 +30,7 @@ async fn main() {
         )
         .await
         .expect("queue_declare");
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
     info!(?queue, "Declared queue");
 
     info!("will consume");
@@ -43,7 +43,7 @@ async fn main() {
         )
         .await
         .expect("basic_consume");
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
 
     while let Some(delivery) = consumer.next().await {
         info!(message=?delivery, "received message");
diff --git a/examples/publisher_confirms.rs b/examples/publisher_confirms.rs
@@ -26,7 +26,7 @@ async fn tokio_main() {
     let channel_a = conn.create_channel().await.expect("create_channel");
     //receive channel
     let channel_b = conn.create_channel().await.expect("create_channel");
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
 
     //create the hello queue
     let queue = channel_a
@@ -37,14 +37,14 @@ async fn tokio_main() {
         )
         .await
         .expect("queue_declare");
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
     info!(?queue, "Declared queue");
 
     channel_a
         .confirm_select(ConfirmSelectOptions::default())
         .await
         .expect("confirm_select");
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
     info!("Enabled publisher-confirms");
 
     info!("will consume");
@@ -66,7 +66,7 @@ async fn tokio_main() {
                     .expect("basic_ack");
             }
         });
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
 
     info!("will publish");
     let payload = b"Hello world!";
@@ -84,7 +84,7 @@ async fn tokio_main() {
         .expect("publisher-confirms");
     assert!(confirm.is_ack());
     assert_eq!(confirm.take_message(), None);
-    info!(state=?conn.status().state());
+    info!(state=?conn.status());
 
     for _ in 1..=2 {
         channel_a
diff --git a/examples/pubsub.rs b/examples/pubsub.rs
@@ -1,7 +1,7 @@
 use futures_lite::stream::StreamExt;
 use lapin::{
-    BasicProperties, Connection, ConnectionProperties, Result, options::*,
-    publisher_confirm::Confirmation, types::FieldTable,
+    BasicProperties, Confirmation, Connection, ConnectionProperties, Result, options::*,
+    types::FieldTable,
 };
 use tracing::info;
 
diff --git a/examples/reconnect.rs b/examples/reconnect.rs
@@ -1,7 +1,7 @@
 use futures_lite::stream::StreamExt;
 use lapin::{
-    BasicProperties, Connection, ConnectionProperties, Result, options::*,
-    publisher_confirm::Confirmation, types::FieldTable,
+    BasicProperties, Confirmation, Connection, ConnectionProperties, Result, options::*,
+    types::FieldTable,
 };
 use tracing::info;
 
diff --git a/src/acker.rs b/src/acker.rs
@@ -89,10 +89,12 @@ impl Acker {
         Ok(true)
     }
 
+    /// True if our channel got closed or encountered an error
     pub fn poisoned(&self) -> bool {
         self.channel_killswitch.killed()
     }
 
+    /// False if poisoned or already used
     pub fn usable(&self) -> bool {
         !self.poisoned() && !self.killswitch.killed()
     }
diff --git a/src/channel.rs b/src/channel.rs
@@ -1,14 +1,12 @@
 use crate::{
-    BasicProperties, Configuration, Connection, ConnectionStatus, Error, ErrorKind, ExchangeKind,
-    Promise, PromiseResolver, Result,
+    BasicProperties, ChannelState, ChannelStatus, Configuration, Connection, ConnectionState,
+    ConnectionStatus, Error, ErrorKind, ExchangeKind, Promise, PromiseResolver, Result,
     acknowledgement::Acknowledgements,
-    auth::Credentials,
     basic_get_delivery::BasicGetDelivery,
     channel_closer::ChannelCloser,
     channel_receiver_state::DeliveryCause,
-    channel_status::{ChannelState, ChannelStatus},
     connection_closer::ConnectionCloser,
-    connection_status::{ConnectionResolver, ConnectionState, ConnectionStep},
+    connection_status::{ConnectionResolver, ConnectionStep},
     consumer::Consumer,
     consumers::Consumers,
     error_handler::ErrorHandler,
@@ -25,7 +23,10 @@ use crate::{
     topology::ChannelDefinition,
     types::*,
 };
-use amq_protocol::frame::{AMQPContentHeader, AMQPFrame};
+use amq_protocol::{
+    auth::Credentials,
+    frame::{AMQPContentHeader, AMQPFrame},
+};
 use executor_trait::FullExecutor;
 use std::{convert::TryFrom, fmt, sync::Arc};
 use tracing::{Level, error, info, level_enabled, trace};
diff --git a/src/channel_closer.rs b/src/channel_closer.rs
@@ -1,6 +1,4 @@
-use crate::{
-    channel_status::ChannelStatus, internal_rpc::InternalRPCHandle, protocol, types::ChannelId,
-};
+use crate::{ChannelStatus, internal_rpc::InternalRPCHandle, protocol, types::ChannelId};
 use std::fmt;
 
 pub(crate) struct ChannelCloser {
diff --git a/src/channel_status.rs b/src/channel_status.rs
@@ -73,10 +73,6 @@ impl ChannelStatus {
         inner.finalize_connection();
     }
 
-    pub fn state(&self) -> ChannelState {
-        self.lock_inner().state
-    }
-
     pub(crate) fn set_state(&self, state: ChannelState) {
         self.lock_inner().state = state;
     }
diff --git a/src/connection.rs b/src/connection.rs
@@ -1,10 +1,9 @@
 use crate::{
-    AsyncTcpStream, ConnectionProperties, Error, ErrorKind, Promise, Result,
+    AsyncTcpStream, ConnectionProperties, ConnectionStatus, Error, ErrorKind, Promise, Result,
     channel::Channel,
     channels::Channels,
     configuration::Configuration,
     connection_closer::ConnectionCloser,
-    connection_status::ConnectionStatus,
     frames::Frames,
     heartbeat::Heartbeat,
     internal_rpc::{InternalRPC, InternalRPCHandle},
@@ -334,12 +333,10 @@ impl Connect for &str {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::BasicProperties;
     use crate::channel_receiver_state::{ChannelReceiverState, DeliveryCause};
-    use crate::channel_status::ChannelState;
-    use crate::connection_status::ConnectionState;
     use crate::options::BasicConsumeOptions;
     use crate::types::{ChannelId, FieldTable, ShortString};
+    use crate::{BasicProperties, ChannelState, ConnectionState};
     use amq_protocol::frame::AMQPContentHeader;
     use amq_protocol::protocol::{AMQPClass, basic};
     use executor_trait::FullExecutor;
@@ -451,9 +448,7 @@ mod tests {
         {
             let body_frame = AMQPFrame::Body(channel.id(), b"{}".to_vec());
             conn.channels.handle_frame(body_frame).unwrap();
-            let channel_state = channel.status().state();
-            let expected_state = ChannelState::Connected;
-            assert_eq!(channel_state, expected_state);
+            assert!(channel.status().connected());
         }
     }
 
@@ -513,9 +508,7 @@ mod tests {
                 }),
             );
             conn.channels.handle_frame(header_frame).unwrap();
-            let channel_state = channel.status().state();
-            let expected_state = ChannelState::Connected;
-            assert_eq!(channel_state, expected_state);
+            assert!(channel.status().connected());
         }
     }
 }
diff --git a/src/connection_closer.rs b/src/connection_closer.rs
@@ -1,4 +1,4 @@
-use crate::{connection_status::ConnectionStatus, internal_rpc::InternalRPCHandle, protocol};
+use crate::{ConnectionStatus, internal_rpc::InternalRPCHandle, protocol};
 use std::sync::atomic::{AtomicBool, Ordering};
 
 pub(crate) struct ConnectionCloser {
diff --git a/src/connection_properties.rs b/src/connection_properties.rs
@@ -5,9 +5,9 @@ use crate::{
 
 #[derive(Clone)]
 pub struct ConnectionProperties {
-    pub locale: String,
-    pub client_properties: FieldTable,
-    pub recovery_config: Option<RecoveryConfig>,
+    pub(crate) locale: String,
+    pub(crate) client_properties: FieldTable,
+    pub(crate) recovery_config: Option<RecoveryConfig>,
 }
 
 impl Default for ConnectionProperties {
@@ -21,6 +21,12 @@ impl Default for ConnectionProperties {
 }
 
 impl ConnectionProperties {
+    #[must_use]
+    pub fn with_locale(mut self, locale: String) -> Self {
+        self.locale = locale;
+        self
+    }
+
     #[must_use]
     pub fn with_connection_name(mut self, connection_name: LongString) -> Self {
         self.client_properties.insert(
diff --git a/src/connection_status.rs b/src/connection_status.rs
@@ -1,8 +1,5 @@
-use crate::{
-    Connection, ConnectionProperties, Error, PromiseResolver, Result,
-    auth::{Credentials, SASLMechanism},
-    uri::AMQPUri,
-};
+use crate::{Connection, ConnectionProperties, Error, PromiseResolver, Result, uri::AMQPUri};
+use amq_protocol::auth::{Credentials, SASLMechanism};
 use std::{
     fmt,
     sync::{Arc, Mutex, MutexGuard},
@@ -19,7 +16,7 @@ impl ConnectionStatus {
         status
     }
 
-    pub fn state(&self) -> ConnectionState {
+    pub(crate) fn state(&self) -> ConnectionState {
         self.lock_inner().state
     }
 
diff --git a/src/consumer.rs b/src/consumer.rs
@@ -2,7 +2,7 @@ use crate::{
     BasicProperties, Error, Result,
     channel_closer::ChannelCloser,
     consumer_canceler::ConsumerCanceler,
-    consumer_status::{ConsumerState, ConsumerStatus},
+    consumer_status::ConsumerStatus,
     error_holder::ErrorHolder,
     internal_rpc::InternalRPCHandle,
     message::{Delivery, DeliveryResult},
@@ -213,11 +213,6 @@ impl Consumer {
         self.consumer_tag.clone()
     }
 
-    /// Gets the current state of the Consumer.
-    pub fn state(&self) -> ConsumerState {
-        self.status.state()
-    }
-
     /// Get the name of the queue we're consuming
     pub fn queue(&self) -> ShortString {
         self.queue.clone()
diff --git a/src/consumer_status.rs b/src/consumer_status.rs
@@ -34,22 +34,22 @@ impl fmt::Debug for ConsumerStatus {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let mut debug = f.debug_struct("ConsumerStatus");
         if let Some(inner) = self.try_read() {
-            debug.field("state", &inner.state());
+            debug.field("state", &inner.state);
         }
         debug.finish()
     }
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub enum ConsumerState {
+pub(crate) enum ConsumerState {
     Active,
     ActiveWithDelegate,
     Canceling,
     Canceled,
 }
 
 impl ConsumerState {
-    pub fn is_active(self) -> bool {
+    pub(crate) fn is_active(self) -> bool {
         matches!(
             self,
             ConsumerState::Active | ConsumerState::ActiveWithDelegate
diff --git a/src/error.rs b/src/error.rs
diff --git a/src/heartbeat.rs b/src/heartbeat.rs
diff --git a/src/io_loop.rs b/src/io_loop.rs
diff --git a/src/lib.rs b/src/lib.rs
diff --git a/src/socket_state.rs b/src/socket_state.rs
diff --git a/tests/connection.rs b/tests/connection.rs

Original file line number	Diff line number	Diff line change
`@@ -89,10 +89,12 @@ impl Acker {`
`89`	`89`	`Ok(true)`
`90`	`90`	`}`
`91`	`91`
	`92`	`+ /// True if our channel got closed or encountered an error`
`92`	`93`	`pub fn poisoned(&self) -> bool {`
`93`	`94`	`self.channel_killswitch.killed()`
`94`	`95`	`}`
`95`	`96`
	`97`	`+ /// False if poisoned or already used`
`96`	`98`	`pub fn usable(&self) -> bool {`
`97`	`99`	`!self.poisoned() && !self.killswitch.killed()`
`98`	`100`	`}`
Original file line number	Diff line number	Diff line change
`@@ -73,10 +73,6 @@ impl ChannelStatus {`
`73`	`73`	`inner.finalize_connection();`
`74`	`74`	`}`
`75`	`75`
`76`		`- pub fn state(&self) -> ChannelState {`
`77`		`- self.lock_inner().state`
`78`		`- }`
`79`		`-`
`80`	`76`	`pub(crate) fn set_state(&self, state: ChannelState) {`
`81`	`77`	`self.lock_inner().state = state;`
`82`	`78`	`}`