minor improvements

Signed-off-by: Heinz N. Gies <[email protected]>

src/connectors.rs

@@ -322,7 +322,7 @@ pub(crate) trait Context: Display + Clone {
 #[derive(Clone)]
 pub(crate) struct ConnectorContext {
     /// alias of the connector instance
-    pub(crate) alias: alias::Connector,
+    alias: alias::Connector,
     /// type of the connector
     connector_type: ConnectorType,
     /// The Quiescence Beacon

src/connectors/impls/discord.rs

@@ -42,6 +42,7 @@ mod utils;
 
 use crate::channel::{bounded, Receiver, Sender};
 use crate::connectors::prelude::*;
+use crate::connectors::Context as ContextTrait;
 use handler::Handler;
 use serenity::prelude::*;
 use tokio::task::JoinHandle;

src/connectors/impls/metrics.rs

@@ -206,7 +206,7 @@ impl Connector for MetricsConnector {
         ctx: SourceContext,
         builder: SourceManagerBuilder,
     ) -> Result<Option<SourceAddr>> {
-        let source = MetricsSource::new(ctx.app_ctx.metrics.rx());
+        let source = MetricsSource::new(ctx.app_ctx().metrics.rx());
         info!("{ctx} Metrics connector id: {}", source.rx.id());
         Ok(Some(builder.spawn(source, ctx)))
     }

src/connectors/impls/tcp/server.rs

@@ -257,7 +257,7 @@ impl Source for TcpServerSource {
                             let tls_reader = TcpReader::tls_server(
                                 tls_read_stream,
                                 vec![0; buf_size],
-                                ctx.alias.clone(),
+                                ctx.alias().clone(),
                                 origin_uri.clone(),
                                 meta,
                                 reader_runtime,
@@ -293,7 +293,7 @@ impl Source for TcpServerSource {
                             let tcp_reader = TcpReader::new(
                                 read_stream,
                                 vec![0; buf_size],
-                                ctx.alias.clone(),
+                                ctx.alias().clone(),
                                 origin_uri.clone(),
                                 meta,
                                 reader_runtime,

src/connectors/sink.rs

@@ -251,23 +251,23 @@ pub(crate) trait SinkRuntime: Send + Sync {
 #[derive(Clone)]
 pub(crate) struct SinkContextInner {
     /// the connector unique identifier
-    pub(crate) uid: SinkUId,
+    uid: SinkUId,
     /// the connector alias
-    pub(crate) alias: alias::Connector,
+    alias: alias::Connector,
     /// the connector type
-    pub(crate) connector_type: ConnectorType,
+    connector_type: ConnectorType,
 
     /// check if we are paused or should stop reading/writing
-    pub(crate) quiescence_beacon: QuiescenceBeacon,
+    quiescence_beacon: QuiescenceBeacon,
 
     /// notifier the connector runtime if we lost a connection
-    pub(crate) notifier: ConnectionLostNotifier,
+    notifier: ConnectionLostNotifier,
 
     /// sender for raft requests
     /// Application Context
-    pub(crate) app_ctx: AppContext,
+    app_ctx: AppContext,
 
-    pub(crate) killswitch: KillSwitch,
+    killswitch: KillSwitch,
 }
 #[derive(Clone)]
 pub(crate) struct SinkContext(Arc<SinkContextInner>);
@@ -285,18 +285,15 @@ impl SinkContext {
             KillSwitch::dummy(),
         )
     }
+
     pub(crate) fn killswitch(&self) -> KillSwitch {
         self.0.killswitch.clone()
     }
+
     pub(crate) fn uid(&self) -> SinkUId {
         self.0.uid
     }
-    pub(crate) fn notifier(&self) -> &ConnectionLostNotifier {
-        &self.0.notifier
-    }
-    pub(crate) fn app_ctx(&self) -> &AppContext {
-        &self.0.app_ctx
-    }
+
     pub(crate) fn new(
         uid: SinkUId,
         alias: alias::Connector,

src/connectors/source.rs

@@ -274,26 +274,26 @@ pub(crate) trait StreamReader: Send {
 #[derive(Clone)]
 pub(crate) struct SourceContext {
     /// connector uid
-    pub(crate) uid: SourceUId,
+    uid: SourceUId,
     /// connector alias
-    pub(crate) alias: alias::Connector,
+    alias: alias::Connector,
 
     /// connector type
-    pub(crate) connector_type: ConnectorType,
+    connector_type: ConnectorType,
     /// The Quiescence Beacon
-    pub(crate) quiescence_beacon: QuiescenceBeacon,
+    quiescence_beacon: QuiescenceBeacon,
 
     /// tool to notify the connector when the connection is lost
-    pub(crate) notifier: ConnectionLostNotifier,
+    notifier: ConnectionLostNotifier,
 
     /// Application Context
-    pub(crate) app_ctx: AppContext,
+    app_ctx: AppContext,
 
     /// kill switch
-    pub(crate) killswitch: KillSwitch,
+    killswitch: KillSwitch,
 
     /// Precomputed prefix for logging
-    pub(crate) prefix: alias::SourceContext,
+    prefix: alias::SourceContext,
 }
 
 impl SourceContext {

src/connectors/tests.rs

@@ -343,7 +343,6 @@ impl ConnectorHarness {
 
     // this is only used in integration tests,
     // otherwise this throws an error when compiled for non-integration tests
-    #[allow(dead_code)]
     pub(crate) async fn signal_tick_to_sink(&self) -> Result<()> {
         self.addr
             .send_sink(SinkMsg::Signal {

src/connectors/utils/pb.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::connectors::prelude::*;
 use crate::errors::{Error, ErrorKind, Result};
 use simd_json::StaticNode;

src/pipeline.rs

@@ -235,7 +235,6 @@ pub(crate) enum MgmtMsg {
 }
 
 #[cfg(test)]
-#[allow(dead_code)]
 mod report {
     use tremor_common::ports::Port;
 

src/raft/test.rs

@@ -70,7 +70,6 @@ impl TestNode {
         })
     }
 
-    #[allow(dead_code)]
     async fn just_start(path: impl AsRef<Path>) -> ClusterResult<Self> {
         let addr = free_node_addr().await?;
         let running = Node::load_from_store(path, raft_config()?).await?;
@@ -82,7 +81,6 @@ impl TestNode {
         })
     }
 
-    #[allow(dead_code)]
     async fn join_as_learner(path: impl AsRef<Path>, join_addr: &Addr) -> ClusterResult<Self> {
         let addr = free_node_addr().await?;
         let mut node = Node::new(path, raft_config()?);