From a43408d2cc573bdc663502f07c532fad22638a6c Mon Sep 17 00:00:00 2001
From: Felipe Cardozo <fraifelipe@gmail.com>
Date: Fri, 6 Sep 2024 15:50:05 -0300
Subject: [PATCH] chore: remove rwlock upgradable
---
crates/fluvio-stream-dispatcher/src/metadata/local.rs | 8 +++++---
1 file changed, 5 insertions(+), 3 deletions(-)
diff --git a/crates/fluvio-stream-dispatcher/src/metadata/local.rs b/crates/fluvio-stream-dispatcher/src/metadata/local.rs
index 35c2891877..685e4bbeb8 100644
--- a/crates/fluvio-stream-dispatcher/src/metadata/local.rs
+++ b/crates/fluvio-stream-dispatcher/src/metadata/local.rs
@@ -102,7 +102,7 @@ cfg_if::cfg_if! {
use anyhow::{Result, anyhow, Context};
use async_channel::{Sender, Receiver, bounded};
- use async_lock::{RwLock, RwLockUpgradableReadGuard};
+ use async_lock::RwLock;
use futures_util::{stream::BoxStream, StreamExt};
use serde::{de::DeserializeOwned};
use tracing::{warn, debug, trace};
@@ -311,13 +311,15 @@ cfg_if::cfg_if! {
async fn get_store<S: Spec + DeserializeOwned>(&self) -> Result<Arc<SpecStore>> {
let key = S::LABEL;
- let read = self.stores.upgradable_read().await;
+ let read = self.stores.read().await;
Ok(match read.get(key) {
Some(store) => store.clone(),
None => {
- let mut write = RwLockUpgradableReadGuard::upgrade(read).await;
+ drop(read);
+ let mut write = self.stores.write().await;
let store = Arc::new(SpecStore::load::<S, _>(self.path.join(key)).await?);
write.insert(key, store.clone());
+ drop(write);
store
}
})