Remove uneccesary wrapper struct from kafka source.
@@ -114,26 +114,15 @@ impl std::fmt::Debug for Int {
}
}
-pub struct StreamAndMsgs<'consumer> {
- pub stream: stream_consumer::MessageStream<'consumer, LoggingConsumerContext, SmolRuntime>,
-}
-
-impl<'consumer> StreamAndMsgs<'consumer> {
- fn new(
- stream: stream_consumer::MessageStream<'consumer, LoggingConsumerContext, SmolRuntime>,
- ) -> Self {
- Self { stream }
- }
-}
-
rental! {
pub mod rentals {
- use super::{StreamAndMsgs, LoggingConsumer};
+ use super::{LoggingConsumer, LoggingMessageStream};
+ use rdkafka::consumer::stream_consumer;
#[rental(covariant)]
pub struct MessageStream {
consumer: Box<LoggingConsumer>,
- stream: StreamAndMsgs<'consumer>
+ stream: LoggingMessageStream<'consumer>
}
}
}
@@ -141,16 +130,14 @@ rental! {
#[allow(clippy::transmute_ptr_to_ptr)]
impl rentals::MessageStream {
#[allow(mutable_transmutes, clippy::mut_from_ref)]
- unsafe fn mut_suffix(
- &self,
- ) -> &mut stream_consumer::MessageStream<'static, LoggingConsumerContext, SmolRuntime> {
- transmute(&self.suffix().stream)
+ unsafe fn mut_suffix(&self) -> &mut LoggingMessageStream<'static> {
+ transmute(&self.suffix())
}
unsafe fn consumer(&mut self) -> &mut LoggingConsumer {
struct MessageStream {
consumer: Box<LoggingConsumer>,
- _stream: StreamAndMsgs<'static>,
+ _stream: LoggingMessageStream<'static>,
};
let s: &mut MessageStream = transmute(self);
&mut s.consumer
@@ -251,6 +238,8 @@ impl ConsumerContext for LoggingConsumerContext {
}
}
pub type LoggingConsumer = StreamConsumer<LoggingConsumerContext>;
+pub type LoggingMessageStream<'consumer> =
+ stream_consumer::MessageStream<'consumer, LoggingConsumerContext, SmolRuntime>;
#[async_trait::async_trait()]
impl Source for Int {
@@ -398,9 +387,7 @@ impl Source for Int {
};
let stream = rentals::MessageStream::new(Box::new(consumer), |c| {
- StreamAndMsgs::new(
- c.start_with_runtime::<SmolRuntime>(Duration::from_millis(100), false),
- )
+ c.start_with_runtime::<SmolRuntime>(Duration::from_millis(100), false)
});
self.stream = Some(stream);
