Skip to content

Commit

Permalink
Merge pull request #731 from MaterializeInc/clonable-threaded-producer
Browse files Browse the repository at this point in the history
make `ThreadedProducer` clonable like `BaseProducer`
  • Loading branch information
benesch authored Oct 3, 2024
2 parents c6d9a65 + dbeca93 commit 99bd070
Showing 1 changed file with 4 additions and 3 deletions.
7 changes: 4 additions & 3 deletions src/producer/base_producer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -639,13 +639,14 @@ where
/// queued events, such as delivery notifications. The thread will be
/// automatically stopped when the producer is dropped.
#[must_use = "The threaded producer will stop immediately if unused"]
#[derive(Clone)]
pub struct ThreadedProducer<C, Part: Partitioner = NoCustomPartitioner>
where
C: ProducerContext<Part> + 'static,
{
producer: Arc<BaseProducer<C, Part>>,
should_stop: Arc<AtomicBool>,
handle: Option<JoinHandle<()>>,
handle: Option<Arc<JoinHandle<()>>>,
}

impl FromClientConfig for ThreadedProducer<DefaultProducerContext, NoCustomPartitioner> {
Expand Down Expand Up @@ -687,7 +688,7 @@ where
Ok(ThreadedProducer {
producer,
should_stop,
handle: Some(thread),
handle: Some(Arc::new(thread)),
})
}
}
Expand Down Expand Up @@ -778,7 +779,7 @@ where
{
fn drop(&mut self) {
trace!("Destroy ThreadedProducer");
if let Some(handle) = self.handle.take() {
if let Some(handle) = self.handle.take().and_then(Arc::into_inner) {
trace!("Stopping polling");
self.should_stop.store(true, Ordering::Relaxed);
trace!("Waiting for polling thread termination");
Expand Down

0 comments on commit 99bd070

Please sign in to comment.