post and user lookups

src/state/mod.rs

@@ -70,24 +70,29 @@ impl State
 	))
     }
 
-    /// Insert a new post
-    pub async fn insert_new_post(&self, post: Post) -> SharedPost
-    {
-	todo!("Rewrite");
-	let mut onee = self.0.oneesan.write().await;
-	let id = *post.post_id();
-
-	if let Some(&owner) = post.owner_id() {
-	    //TODO: Add `index` to `posts_user_map`. (how do we get `index` before this)
-	}
-	
-	let post = Arc::new(RwLock::new(post));
-	//TODO: If post exists with the same ID, remove it first.
-	let index = onee.posts.insert(Arc::clone(&post));
-	onee.posts_map.insert(id, index);
 
-	
-	post
+    /// Get a shared reference to a user by this ID, if it exists.
+    ///
+    /// # Locks
+    /// This functions holds the state read lock while performing lookups.
+    /// # Panics
+    /// If the internal ID mappings are invalid
+    pub async fn get_user_by_id(&self, id: UserID) -> Option<SharedUser>
+    {
+	let read = self.0.oneesan.read().await;
+	read.users_map.get(&id).map(|&idx| read.users.get(idx).unwrap().clone())
+    }
+    
+    /// Get a shared reference to a post by this ID, if it exists.
+    ///
+    /// # Locks
+    /// This functions holds the state read lock while performing lookups.
+    /// # Panics
+    /// If the internal ID mappings are invalid
+    pub async fn get_post_by_id(&self, id: PostID) -> Option<SharedPost>
+    {
+	let read = self.0.oneesan.read().await;
+	read.posts_map.get(&id).map(|&idx| read.posts.get(idx).unwrap().clone())
     }
     
     /// Consume into a stream that yields all users lazily.
@@ -108,6 +113,31 @@ impl State
 	});
 	rx
     }
+
+    
+    /// Consume into a stream that yields all posts created by this user lazily.
+    ///
+    /// # Locks
+    /// The stream producer holds the Oneesan *read* lock until the stream is consumed or dropped.
+    ///
+    /// # Panics
+    /// The background task will panic and drop the producer if the internal ID mappings are invalid
+    pub fn all_posts_by_user_stream(self: Arc<Self>, user: UserID) -> impl Stream<Item=SharedPost>
+    {
+	let (mut tx, rx) = mpsc_channel!(defaults::STATE_STREAM_BUFFER_SIZE);
+	tokio::spawn(async move {
+	    let onee = self.0.oneesan.read().await;
+	    if let Some(map) = onee.posts_user_map.get(&user) {
+		for &post in map.iter()
+		{
+		    if tx.send(onee.posts.get(post).unwrap().clone()).await.is_err() {
+			break;
+		    }
+		}
+	    }
+	});
+	rx
+    }
     
     /// Consume into a stream that yields all posts lazily.
     ///