Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Improve ergonomics of Listening API #73

Draft
wants to merge 7 commits into
base: master
Choose a base branch
from
Draft

Improve ergonomics of Listening API #73

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
f688f85
Passing database instance to listen callback and remove Sync
abdolence Feb 6, 2023
23accc9
Change clone to ref for listen_changes
abdolence Feb 6, 2023
ce2aac5
Revert back to clone for now
abdolence Feb 6, 2023
80e4da4
Merge branch 'master' into feature/improve-listen-ergonomics
abdolence Aug 11, 2023
2ddc9d1
FirestoreTargetManager support
abdolence Aug 11, 2023
b14b8ba
FirestoreTargetManager store updates
abdolence Aug 11, 2023
91168f6
Clippy fixes
abdolence Aug 11, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion examples/listen-changes.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
.add_target(TEST_TARGET_ID_BY_DOC_IDS, &mut listener)?;

listener
.start(|event| async move {
.start(|event, _edb| async move {
match event {
FirestoreListenEvent::DocumentChange(ref doc_change) => {
println!("Doc changed: {doc_change:?}");
Expand All @@ -87,6 +87,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
}
}
_ => {
//storage.remove_target(&TEST_TARGET_ID_BY_DOC_IDS).ok();
println!("Received a listen response event to handle: {event:?}");
}
}
Expand Down
159 changes: 99 additions & 60 deletions src/db/listen_changes.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,6 +232,11 @@ impl FirestoreDb {

pub type FirestoreListenEvent = listen_response::ResponseType;

pub trait FirestoreTargetManager {
fn add_target(&mut self, target: FirestoreListenerTargetParams) -> FirestoreResult<()>;
fn remove_target(&mut self, target: &FirestoreListenerTarget) -> FirestoreResult<()>;
}

#[derive(Debug, Clone, Builder)]
pub struct FirestoreListenerParams {
pub retry_delay: Option<std::time::Duration>,
Expand All @@ -243,9 +248,9 @@ where
S: FirestoreResumeStateStorage,
{
db: D,
storage: S,
initial_targets_storage: FirestoreTargetManagerStorage,
resume_state_storage: S,
listener_params: FirestoreListenerParams,
targets: Vec<FirestoreListenerTargetParams>,
shutdown_flag: Arc<AtomicBool>,
shutdown_handle: Option<JoinHandle<()>>,
shutdown_writer: Option<Arc<UnboundedSender<i8>>>,
Expand All @@ -258,44 +263,35 @@ where
{
pub async fn new(
db: D,
storage: S,
resume_state_storage: S,
listener_params: FirestoreListenerParams,
) -> FirestoreResult<FirestoreListener<D, S>> {
Ok(FirestoreListener {
db,
storage,
initial_targets_storage: FirestoreTargetManagerStorage::new(false),
resume_state_storage,
listener_params,
targets: vec![],
shutdown_flag: Arc::new(AtomicBool::new(false)),
shutdown_handle: None,
shutdown_writer: None,
})
}

pub fn add_target(
&mut self,
target_params: FirestoreListenerTargetParams,
) -> FirestoreResult<()> {
target_params.validate()?;
self.targets.push(target_params);
Ok(())
}

pub async fn start<FN, F>(&mut self, cb: FN) -> FirestoreResult<()>
where
FN: Fn(FirestoreListenEvent) -> F + Send + Sync + 'static,
FN: Fn(FirestoreListenEvent, D) -> F + Send + Sync + 'static,
F: Future<Output = AnyBoxedErrResult<()>> + Send + 'static,
{
info!(
"Starting a Firestore listener for targets: {:?}...",
&self.targets.len()
"Starting a Firestore listener for initial targets: {:?}...",
&self.initial_targets_storage.targets.len()
);

let mut initial_states: HashMap<FirestoreListenerTarget, FirestoreListenerTargetParams> =
HashMap::new();
for target_params in &self.targets {
let initial_state = self
.storage
let mut targets_storage = self.initial_targets_storage.clone();

for target_params in &mut targets_storage.targets.values_mut() {
let maybe_initial_token = self
.resume_state_storage
.read_resume_state(&target_params.target)
.map_err(|err| {
FirestoreError::SystemError(FirestoreSystemError::new(
Expand All @@ -305,10 +301,7 @@ where
})
.await?;

initial_states.insert(
target_params.target.clone(),
target_params.clone().opt_resume_type(initial_state),
);
target_params.mopt_resume_type(maybe_initial_token);
}

let (tx, rx): (UnboundedSender<i8>, UnboundedReceiver<i8>) =
Expand All @@ -317,9 +310,9 @@ where
self.shutdown_writer = Some(Arc::new(tx));
self.shutdown_handle = Some(tokio::spawn(Self::listener_loop(
self.db.clone(),
self.storage.clone(),
self.resume_state_storage.clone(),
self.shutdown_flag.clone(),
initial_states,
targets_storage,
self.listener_params.clone(),
rx,
cb,
Expand All @@ -346,24 +339,27 @@ where
db: D,
storage: S,
shutdown_flag: Arc<AtomicBool>,
mut targets_state: HashMap<FirestoreListenerTarget, FirestoreListenerTargetParams>,
mut targets_state: FirestoreTargetManagerStorage,
listener_params: FirestoreListenerParams,
mut shutdown_receiver: UnboundedReceiver<i8>,
cb: FN,
) where
D: FirestoreListenSupport + Clone + Send + Sync,
FN: Fn(FirestoreListenEvent) -> F + Send + Sync,
FN: Fn(FirestoreListenEvent, D) -> F + Send + Sync,
F: Future<Output = AnyBoxedErrResult<()>> + Send,
{
let effective_delay = listener_params
.retry_delay
.unwrap_or_else(|| std::time::Duration::from_secs(5));

while !shutdown_flag.load(Ordering::Relaxed) {
debug!("Start listening on {} targets ... ", targets_state.len());
debug!(
"Start listening on {} targets ... ",
targets_state.targets.len()
);

match db
.listen_doc_changes(targets_state.values().cloned().collect())
.listen_doc_changes(targets_state.targets.values().cloned().collect())
.await
{
Err(err) => {
Expand All @@ -374,7 +370,7 @@ where
Ok(mut listen_stream) => loop {
tokio::select! {
_ = shutdown_receiver.recv() => {
debug!("Exiting from listener on {} targets...", targets_state.len());
debug!("Exiting from listener on {} targets...", targets_state.targets.len());
shutdown_receiver.close();
break;
}
Expand All @@ -386,40 +382,36 @@ where
match tried {
Ok(Some(event)) => {
trace!("Received a listen response event to handle: {:?}", event);
match event.response_type {
Some(listen_response::ResponseType::TargetChange(ref target_change))
if !target_change.resume_token.is_empty() =>
{
for target_id_num in &target_change.target_ids {
match FirestoreListenerTarget::try_from(*target_id_num) {
Ok(target_id) => {
if let Some(target) = targets_state.get_mut(&target_id) {
let new_token: FirestoreListenerToken = target_change.resume_token.clone().into();

if let Err(err) = storage.update_resume_token(&target.target, new_token.clone()).await {
error!("Listener token storage error occurred {:?}.", err);
break;
}
else {
target.resume_type = Some(FirestoreListenerTargetResumeType::Token(new_token))
if let Some(response_type) = event.response_type {
if let listen_response::ResponseType::TargetChange(ref target_change) = &response_type {
if !target_change.resume_token.is_empty() {
for target_id_num in &target_change.target_ids {
match FirestoreListenerTarget::try_from(*target_id_num) {
Ok(target_id) => {
if let Some(target) = targets_state.targets.get_mut(&target_id) {
let new_token: FirestoreListenerToken = target_change.resume_token.clone().into();

if let Err(err) = storage.update_resume_token(&target.target, new_token.clone()).await {
error!("Listener token storage error occurred {:?}.", err);
break;
}
else {
target.resume_type = Some(FirestoreListenerTargetResumeType::Token(new_token))
}
}
},
Err(err) => {
error!("Listener system error - unexpected target ID: {} {:?}.", target_id_num, err);
break;
}
},
Err(err) => {
error!("Listener system error - unexpected target ID: {} {:?}.", target_id_num, err);
break;
}
}
}

}
Some(response_type) => {
if let Err(err) = cb(response_type).await {
error!("Listener callback function error occurred {:?}.", err);
break;
}
if let Err(err) = cb(response_type, db.clone()).await {
error!("Listener callback function error occurred {:?}.", err);
break;
}
None => {}
}
}
Ok(None) => break,
Expand Down Expand Up @@ -469,3 +461,50 @@ where
}
}
}

impl<D, S> FirestoreTargetManager for FirestoreListener<D, S>
where
D: FirestoreListenSupport + Clone + Send + Sync + 'static,
S: FirestoreResumeStateStorage + Clone + Send + Sync + 'static,
{
fn add_target(&mut self, target_params: FirestoreListenerTargetParams) -> FirestoreResult<()> {
self.initial_targets_storage.add_target(target_params)
}

fn remove_target(&mut self, target: &FirestoreListenerTarget) -> FirestoreResult<()> {
self.initial_targets_storage.remove_target(target)
}
}

#[derive(Clone)]
pub struct FirestoreTargetManagerStorage {
targets: HashMap<FirestoreListenerTarget, FirestoreListenerTargetParams>,
change_log_mode: bool,
to_remove: Vec<FirestoreListenerTarget>,
}

impl FirestoreTargetManagerStorage {
fn new(change_log_mode: bool) -> Self {
Self {
targets: HashMap::new(),
change_log_mode,
to_remove: Vec::new(),
}
}
}

impl FirestoreTargetManager for FirestoreTargetManagerStorage {
fn add_target(&mut self, target_params: FirestoreListenerTargetParams) -> FirestoreResult<()> {
target_params.validate()?;
self.targets
.insert(target_params.target.clone(), target_params);
Ok(())
}

fn remove_target(&mut self, target: &FirestoreListenerTarget) -> FirestoreResult<()> {
if self.targets.remove(target).is_none() && self.change_log_mode {
self.to_remove.push(target.clone());
}
Ok(())
}
}
15 changes: 7 additions & 8 deletions src/fluent_api/select_builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,10 @@ use crate::select_filter_builder::FirestoreQueryFilterBuilder;
use crate::{
FirestoreAggregatedQueryParams, FirestoreAggregatedQuerySupport, FirestoreAggregation,
FirestoreCollectionDocuments, FirestoreGetByIdSupport, FirestoreListenSupport,
FirestoreListener, FirestoreListenerParams, FirestoreListenerTarget,
FirestoreListenerTargetParams, FirestorePartition, FirestorePartitionQueryParams,
FirestoreQueryCollection, FirestoreQueryCursor, FirestoreQueryFilter, FirestoreQueryOrder,
FirestoreQueryParams, FirestoreQuerySupport, FirestoreResult, FirestoreResumeStateStorage,
FirestoreTargetType,
FirestoreListenerParams, FirestoreListenerTarget, FirestoreListenerTargetParams,
FirestorePartition, FirestorePartitionQueryParams, FirestoreQueryCollection,
FirestoreQueryCursor, FirestoreQueryFilter, FirestoreQueryOrder, FirestoreQueryParams,
FirestoreQuerySupport, FirestoreResult, FirestoreTargetManager, FirestoreTargetType,
};
use futures::stream::BoxStream;
use gcloud_sdk::google::firestore::v1::Document;
Expand Down Expand Up @@ -780,13 +779,13 @@ where
}

#[inline]
pub fn add_target<S>(
pub fn add_target<TM>(
self,
target: FirestoreListenerTarget,
listener: &mut FirestoreListener<D, S>,
listener: &mut TM,
) -> FirestoreResult<()>
where
S: FirestoreResumeStateStorage + Send + Sync + Clone + 'static,
TM: FirestoreTargetManager,
{
listener.add_target(FirestoreListenerTargetParams::new(
target,
Expand Down