diff options
| -rw-r--r-- | src/main.rs | 90 | ||||
| -rw-r--r-- | src/tw/mod.rs | 289 |
2 files changed, 198 insertions, 181 deletions
diff --git a/src/main.rs b/src/main.rs index b9bf27d..e45a799 100644 --- a/src/main.rs +++ b/src/main.rs @@ -210,11 +210,11 @@ fn main() { tweeter.display_info.status("Cache loaded".to_owned()); - let (twete_tx, mut twete_rx) = chan::sync::<Vec<u8>>(0); - let (coordination_tx, mut coordination_rx) = chan::sync::<TwitterConnectionState>(0); + let (twete_tx, twete_rx) = chan::sync::<(u8, Vec<u8>)>(0); + let (coordination_tx, coordination_rx) = chan::sync::<(u8, TwitterConnectionState)>(0); for (ref profile_name, ref profile) in &tweeter.profiles { - connect_twitter_stream(tweeter.app_key.clone(), profile.creds.clone(), twete_tx.clone(), coordination_tx.clone(), get_id()); + connect_twitter_stream(tweeter.app_key.clone(), profile_name.to_string(), profile.creds.clone(), twete_tx.clone(), coordination_tx.clone(), get_id()); } std::thread::spawn(move || { @@ -249,18 +249,8 @@ fn main() { Err(e) => println!("{}", e) // TODO: we got here because writing to stdout failed. what to do now? }; - loop { - match do_ui(ui_rx, twete_rx, &twete_tx, coordination_rx, &coordination_tx, &mut tweeter, &mut queryer) { - Some((new_ui_rx, new_twete_rx, new_coordination_rx)) => { - ui_rx = new_ui_rx; - twete_rx = new_twete_rx; - coordination_rx = new_coordination_rx; - }, - None => { - break; - } - } - } + do_ui(ui_rx, twete_rx, &twete_tx, coordination_rx, &coordination_tx, &mut tweeter, &mut queryer); + tcsetattr(0, TCSANOW, &termios); } @@ -294,7 +284,7 @@ fn handle_input(event: termion::event::Event, tweeter: &mut tw::TwitterCache, qu tweeter.display_info.mode = Some(display::DisplayMode::Reply(twid, "".to_owned())); } } - } + } Event::Key(Key::Ctrl('n')) => { match tweeter.display_info.mode.clone() { Some(display::DisplayMode::Compose(msg)) => { @@ -359,12 +349,12 @@ fn handle_input(event: termion::event::Event, tweeter: &mut tw::TwitterCache, qu } } -fn handle_twitter_line(line: Vec<u8>, mut tweeter: &mut tw::TwitterCache, mut queryer: &mut ::Queryer) { +fn handle_twitter_line(conn_id: u8, line: Vec<u8>, mut tweeter: &mut tw::TwitterCache, mut queryer: &mut ::Queryer) { let jsonstr = std::str::from_utf8(&line).unwrap().trim(); /* TODO: replace from_str with from_slice? */ match serde_json::from_str(&jsonstr) { Ok(json) => { - tw::handle_message(json, &mut tweeter, &mut queryer); + tw::handle_message(conn_id, json, &mut tweeter, &mut queryer); if tweeter.needs_save && tweeter.caching_permitted { tweeter.store_cache(); } @@ -375,33 +365,46 @@ fn handle_twitter_line(line: Vec<u8>, mut tweeter: &mut tw::TwitterCache, mut qu } fn do_ui( - ui_rx_orig: chan::Receiver<Result<termion::event::Event, std::io::Error>>, - twete_rx: chan::Receiver<Vec<u8>>, - twete_tx: &chan::Sender<Vec<u8>>, - coordination_rx: chan::Receiver<TwitterConnectionState>, - coordination_tx: &chan::Sender<TwitterConnectionState>, + ui_rx: chan::Receiver<Result<termion::event::Event, std::io::Error>>, + twete_rx: chan::Receiver<(u8, Vec<u8>)>, + twete_tx: &chan::Sender<(u8, Vec<u8>)>, + coordination_rx: chan::Receiver<(u8, TwitterConnectionState)>, + coordination_tx: &chan::Sender<(u8, TwitterConnectionState)>, mut tweeter: &mut tw::TwitterCache, mut queryer: &mut ::Queryer -) -> Option<(chan::Receiver<Result<termion::event::Event, std::io::Error>>, chan::Receiver<Vec<u8>>, chan::Receiver<TwitterConnectionState>)> { +) { loop { - let ui_rx_a = &ui_rx_orig; - let ui_rx_b = &ui_rx_orig; chan_select! { coordination_rx.recv() -> coordination => { - tweeter.display_info.status(format!("{:?}", coordination)); + match coordination { + Some((conn_id, coordination)) => { + match coordination { + TwitterConnectionState::Connecting(profile_name) => { + tweeter.connection_map.insert(conn_id, profile_name); + }, + TwitterConnectionState::Connected => { + tweeter.display_info.status(format!("Stream connected for profile \"{}\"", tweeter.connection_map[&conn_id])); + }, + TwitterConnectionState::Closed => { + tweeter.connection_map.remove(&conn_id); + } + } + }, + None => { /* if this stream closes something is terribly wrong... */ panic!("Coordination tx/rx closed!"); } + } }, twete_rx.recv() -> twete => match twete { - Some(line) => handle_twitter_line(line, tweeter, queryer), + Some((conn_id, line)) => handle_twitter_line(conn_id, line, tweeter, queryer), None => { tweeter.display_info.status("Twitter stream hung up...".to_owned()); display::paint(tweeter).unwrap(); - return None // if the twitter channel died, something real bad happeneed? + return; // if the twitter channel died, something real bad happeneed? } }, - ui_rx_a.recv() -> user_input => match user_input { + ui_rx.recv() -> user_input => match user_input { Some(Ok(event)) => handle_input(event, tweeter, queryer), Some(Err(_)) => (), /* stdin closed? */ - None => return None // UI ded + None => return // UI ded } } @@ -443,7 +446,7 @@ fn do_ui( tweeter.state = tw::AppState::View; match tweeter.profiles.get(&profile_name).map(|profile| profile.creds.to_owned()) { Some(user_creds) => { - connect_twitter_stream(tweeter.app_key.clone(), user_creds, twete_tx.clone(), coordination_tx.clone(), get_id()) + connect_twitter_stream(tweeter.app_key.clone(), profile_name, user_creds, twete_tx.clone(), coordination_tx.clone(), get_id()) }, None => { tweeter.display_info.status(format!("No profile named {}", profile_name)); @@ -456,9 +459,9 @@ fn do_ui( tweeter.store_cache(); tweeter.display_info.status("Bye bye!".to_owned()); display::paint(tweeter).unwrap(); - return None + return; }, - _ => () + tw::AppState::View | tw::AppState::Compose => { /* nothing special to do */ } }; } } @@ -498,19 +501,20 @@ fn url_encode(s: &str) -> String { // let (twete_tx, twete_rx) = chan::sync::<Vec<u8>>(0); #[derive(Debug)] enum TwitterConnectionState { - Connecting(u8), - Connected(u8), - Closed(u8) + Connecting(String), + Connected, + Closed } fn connect_twitter_stream( app_cred: tw::Credential, + profile_name: String, user_cred: tw::Credential, - twete_tx: chan::Sender<Vec<u8>>, - coordination_tx: chan::Sender<TwitterConnectionState>, + twete_tx: chan::Sender<(u8, Vec<u8>)>, + coordination_tx: chan::Sender<(u8, TwitterConnectionState)>, conn_id: u8 ) { std::thread::spawn(move || { - coordination_tx.send(TwitterConnectionState::Connecting(conn_id)); + coordination_tx.send((conn_id, TwitterConnectionState::Connecting(profile_name))); let mut core = Core::new().unwrap(); let connector = HttpsConnector::new(1, &core.handle()).unwrap(); @@ -527,13 +531,13 @@ fn connect_twitter_stream( println!("Twitter stream connect was abnormal: {}", status); println!("result: {:?}", res); } - coordination_tx.send(TwitterConnectionState::Connected(conn_id)); + coordination_tx.send((conn_id, TwitterConnectionState::Connected)); LineStream::new(res.body() .map(|chunk| futures::stream::iter_ok(chunk.into_iter())) .flatten()) .for_each(|s| { if s.len() != 1 { - twete_tx.send(s); + twete_tx.send((conn_id, s)); }; Ok(()) }) @@ -544,6 +548,6 @@ fn connect_twitter_stream( Ok(_good) => (), Err(e) => println!("Error in setting up: {}", e) } - coordination_tx.send(TwitterConnectionState::Closed(conn_id)); + coordination_tx.send((conn_id, TwitterConnectionState::Closed)); }); } diff --git a/src/tw/mod.rs b/src/tw/mod.rs index b81fa52..3d3f08a 100644 --- a/src/tw/mod.rs +++ b/src/tw/mod.rs @@ -111,7 +111,9 @@ pub struct TwitterCache { #[serde(skip)] pub display_info: display::DisplayInfo, #[serde(skip)] - pub state: AppState + pub state: AppState, + #[serde(skip)] + pub connection_map: HashMap<u8, String> } // Internally, a monotonically increasin i64 is always the id used. @@ -289,6 +291,94 @@ impl TwitterProfile { follower_history: HashMap::new() } } + pub fn get_settings(&self, queryer: &mut ::Queryer, app_key: &Credential) -> Result<serde_json::Value, String> { + queryer.do_api_get(::ACCOUNT_SETTINGS_URL, app_key, &self.creds) + } + pub fn get_followers(&self, queryer: &mut ::Queryer, app_key: &Credential) -> Result<serde_json::Value, String> { + queryer.do_api_get(::GET_FOLLOWER_IDS_URL, app_key, &self.creds) + } + pub fn set_following(&mut self, user_ids: Vec<String>) -> (Vec<String>, Vec<String>) { + let uid_set = user_ids.into_iter().collect::<HashSet<String>>(); + let mut new_following: Vec<String> = vec![]; + let mut lost_following: Vec<String> = vec![]; + + let new_uids = &uid_set - &self.following; + for user in new_uids { + self.add_following(&user); + new_following.push(user); + } + + let lost_uids = &self.following - &uid_set; + for user in lost_uids { + self.remove_following(&user); + lost_following.push(user); + } + (new_following, lost_following) + } + pub fn set_followers(&mut self, user_ids: Vec<String>) -> (Vec<String>, Vec<String>) { + let uid_set = user_ids.into_iter().collect::<HashSet<String>>(); + let mut new_follower: Vec<String> = vec![]; + let mut lost_follower: Vec<String> = vec![]; + + let new_uids = &uid_set - &self.followers; + for user in new_uids { + self.add_follower(&user); + new_follower.push(user); + } + + let lost_uids = &self.followers - &uid_set; + for user in lost_uids { + self.remove_follower(&user); + lost_follower.push(user); + } + (new_follower, lost_follower) + } + /* + * Returns: "did this change?" + * + * but currently pessimistically always returns true right now + * TODO: check if this should return false! (that's probably a cache bug though.) + */ + pub fn add_following(&mut self, user_id: &String) -> bool { + self.following.insert(user_id.to_owned()); + self.following_history.insert(user_id.to_owned(), ("following".to_string(), Utc::now().timestamp())); + true + } + /* + * Returns: "did this change?" + * + * but currently pessimistically always returns true right now + * TODO: check if this should return false! (that's probably a cache bug though.) + */ + pub fn remove_following(&mut self, user_id: &String) -> bool { + self.following.remove(user_id); + self.following_history.insert(user_id.to_owned(), ("unfollowing".to_string(), Utc::now().timestamp())); + true + } + /* + * Returns: "did this change?" + * + * but currently pessimistically always returns true right now + * TODO: check if this should return false! (that's probably a cache bug though.) + */ + pub fn add_follower(&mut self, user_id: &String) -> bool { + self.followers.insert(user_id.to_owned()); + self.lost_followers.remove(user_id); + self.follower_history.insert(user_id.to_owned(), ("follow".to_string(), Utc::now().timestamp())); + true + } + /* + * Returns: "did this change?" + * + * but currently pessimistically always returns true right now + * TODO: check if this should return false! (that's probably a cache bug though.) + */ + pub fn remove_follower(&mut self, user_id: &String) -> bool { + self.followers.remove(user_id); + self.lost_followers.insert(user_id.to_owned()); + self.follower_history.insert(user_id.to_owned(), ("unfollow".to_string(), Utc::now().timestamp())); + true + } } impl TwitterCache { @@ -316,10 +406,15 @@ impl TwitterCache { threads: HashMap::new(), id_conversions: IdConversions::default(), display_info: display::DisplayInfo::default(), - state: AppState::View + state: AppState::View, + connection_map: HashMap::new() } } + pub fn mut_profile_for_connection(&mut self, conn_id: u8) -> &mut TwitterProfile { + self.profiles.get_mut(&self.connection_map[&conn_id]).unwrap() + } + pub fn current_profile(&self) -> Option<&TwitterProfile> { match &self.curr_profile { &Some(ref profile_name) => self.profiles.get(profile_name), @@ -486,7 +581,7 @@ impl TwitterCache { self.cache_user(user); } } - pub fn cache_api_event(&mut self, json: serde_json::Map<String, serde_json::Value>, mut queryer: &mut ::Queryer) { + pub fn cache_api_event(&mut self, conn_id: u8, json: serde_json::Map<String, serde_json::Value>, mut queryer: &mut ::Queryer) { /* don't really care to hold on to who fav, unfav, ... when, just pick targets out. */ match json.get("event").and_then(|x| x.as_str()) { Some("quoted_tweet") => { @@ -522,24 +617,11 @@ impl TwitterCache { let followed = json["target"]["id_str"].as_str().unwrap().to_string(); self.cache_api_user(json["target"].clone()); self.cache_api_user(json["source"].clone()); - match self.current_profile().map(|profile| profile.to_owned()) { - Some(profile) => { - // for now assume single client? - // TODO: see note below - profile origin needs to be tracked. - // - if follower == profile.user.handle { - // self.add_follow( - } else { - self.add_follower(&follower); - } - }, - None => { - // TODO: this isn't really reachable - we'd have to be connected on some - // ... this will break. - // - // events need to include what profile they're associated with so we can - // note that for the tweet and event if applicable. - } + let profile = self.mut_profile_for_connection(conn_id); + if follower == profile.user.handle { + // self.add_follow( + } else { + profile.add_follower(&follower); } }, Some("unfollow") => { @@ -547,24 +629,11 @@ impl TwitterCache { let followed = json["target"]["id_str"].as_str().unwrap().to_string(); self.cache_api_user(json["target"].clone()); self.cache_api_user(json["source"].clone()); - match self.current_profile().map(|profile| profile.to_owned()) { - Some(profile) => { - // for now assume single client? - // TODO: see note below - profile origin needs to be tracked. - // - if follower == profile.user.handle { - // self.add_follow( - } else { - self.add_follower(&follower); - } - }, - None => { - // TODO: this isn't really reachable - we'd have to be connected on some - // ... this will break. - // - // events need to include what profile they're associated with so we can - // note that for the tweet and event if applicable. - } + let profile = self.mut_profile_for_connection(conn_id); + if follower == profile.user.handle { + // self.add_follow( + } else { + profile.add_follower(&follower); } }, Some(_) => () /* an uninteresting event */, @@ -631,70 +700,6 @@ impl TwitterCache { } self.users.get(user_id) } - pub fn set_following(&mut self, user_ids: Vec<String>) { - self.current_profile().map(|profile| profile.to_owned()).map(|mut profile| { - let uid_set = user_ids.into_iter().collect::<HashSet<String>>(); - - let new_uids = &uid_set - &profile.following; - for user in &new_uids { - self.display_info.status(format!("New following! {}", user)); - self.add_following(user); - } - - let lost_uids = &profile.following - &uid_set; - for user in &lost_uids { - self.display_info.status(format!("Bye, friend! {}", user)); - self.remove_following(user); - } - }); - } - pub fn set_followers(&mut self, user_ids: Vec<String>) { - self.current_profile().map(|profile| profile.to_owned()).map(|mut profile| { - let uid_set = user_ids.into_iter().collect::<HashSet<String>>(); - - let new_uids = &uid_set - &profile.followers; - for user in &new_uids { - self.display_info.status(format!("New follower! {}", user)); - self.add_follower(user); - } - - let lost_uids = &profile.followers - &uid_set; - for user in &lost_uids { - self.display_info.status(format!("Bye, friend! {}", user)); - self.remove_follower(user); - } - }); - } - pub fn add_following(&mut self, user_id: &String) { - self.current_profile().map(|profile| profile.to_owned()).map(|mut profile| { - self.needs_save = true; - profile.following.insert(user_id.to_owned()); - profile.following_history.insert(user_id.to_owned(), ("following".to_string(), Utc::now().timestamp())); - }); - } - pub fn remove_following(&mut self, user_id: &String) { - self.current_profile().map(|profile| profile.to_owned()).map(|mut profile| { - self.needs_save = true; - profile.following.remove(user_id); - profile.following_history.insert(user_id.to_owned(), ("unfollowing".to_string(), Utc::now().timestamp())); - }); - } - pub fn add_follower(&mut self, user_id: &String) { - self.current_profile().map(|profile| profile.to_owned()).map(|mut profile| { - self.needs_save = true; - profile.followers.insert(user_id.to_owned()); - profile.lost_followers.remove(user_id); - profile.follower_history.insert(user_id.to_owned(), ("follow".to_string(), Utc::now().timestamp())); - }); - } - pub fn remove_follower(&mut self, user_id: &String) { - self.current_profile().map(|profile| profile.to_owned()).map(|mut profile| { - self.needs_save = true; - profile.followers.remove(user_id); - profile.lost_followers.insert(user_id.to_owned()); - profile.follower_history.insert(user_id.to_owned(), ("unfollow".to_string(), Utc::now().timestamp())); - }); - } fn look_up_user(&mut self, id: &str, queryer: &mut ::Queryer) -> Result<serde_json::Value, String> { let url = &format!("{}?user_id={}", ::USER_LOOKUP_URL, id); @@ -712,20 +717,6 @@ impl TwitterCache { } } - pub fn get_settings(&self, queryer: &mut ::Queryer) -> Result<serde_json::Value, String> { - match self.current_profile() { - Some(ref user_profile) => queryer.do_api_get(::ACCOUNT_SETTINGS_URL, &self.app_key, &user_profile.creds), - None => Err("No authorized user to request settings".to_owned()) - } - } - - pub fn get_followers(&self, queryer: &mut ::Queryer) -> Result<serde_json::Value, String> { - match self.current_profile() { - Some(ref user_profile) => queryer.do_api_get(::GET_FOLLOWER_IDS_URL, &self.app_key, &user_profile.creds), - None => Err("No authorized user to request followers".to_owned()) - } - } - pub fn set_thread(&mut self, name: String, last_id: u64) -> bool { self.threads.insert(name, last_id); true @@ -756,10 +747,11 @@ impl TwitterCache { } fn handle_twitter_event( + conn_id: u8, structure: serde_json::Map<String, serde_json::Value>, tweeter: &mut TwitterCache, mut queryer: &mut ::Queryer) { - tweeter.cache_api_event(structure.clone(), &mut queryer); + tweeter.cache_api_event(conn_id, structure.clone(), &mut queryer); match events::Event::from_json(structure) { Ok(event) => { tweeter.display_info.recv(display::Infos::Event(event)); @@ -771,6 +763,7 @@ fn handle_twitter_event( } fn handle_twitter_delete( + conn_id: u8, structure: serde_json::Map<String, serde_json::Value>, tweeter: &mut TwitterCache, _queryer: &mut ::Queryer) { @@ -785,6 +778,7 @@ fn handle_twitter_delete( } fn handle_twitter_twete( + conn_id: u8, structure: serde_json::Map<String, serde_json::Value>, tweeter: &mut TwitterCache, _queryer: &mut ::Queryer) { @@ -795,6 +789,7 @@ fn handle_twitter_twete( } fn handle_twitter_dm( + conn_id: u8, structure: serde_json::Map<String, serde_json::Value>, tweeter: &mut TwitterCache, _queryer: &mut ::Queryer) { @@ -803,25 +798,42 @@ fn handle_twitter_dm( } fn handle_twitter_welcome( + conn_id: u8, structure: serde_json::Map<String, serde_json::Value>, tweeter: &mut TwitterCache, queryer: &mut ::Queryer) { - let user_id_nums = structure["friends"].as_array().unwrap(); - let user_id_strs = user_id_nums.into_iter().map(|x| x.as_u64().unwrap().to_string()); - tweeter.set_following(user_id_strs.collect()); - let settings = tweeter.get_settings(queryer).unwrap(); - let maybe_my_name = settings["screen_name"].as_str(); - if let Some(my_name) = maybe_my_name { - // TODO: come back to this when custom profile names are supported? -// tweeter.curr_profile = Some(my_name.to_owned()); - tweeter.display_info.status(format!("You are {}", my_name)) - } else { - tweeter.display_info.status("Unable to make API call to figure out who you are...".to_string()); - } - match tweeter.get_followers(queryer) { - Ok(followers) => { + let app_key = tweeter.app_key.clone(); + let followers_changes = { + let mut profile = tweeter.mut_profile_for_connection(conn_id); + + let settings = profile.get_settings(queryer, &app_key).unwrap(); + + let user_id_nums = structure["friends"].as_array().unwrap(); + let user_id_strs = user_id_nums.into_iter().map(|x| x.as_u64().unwrap().to_string()); + let (new_following, lost_following) = profile.set_following(user_id_strs.collect()); + + let maybe_my_name = settings["screen_name"].as_str(); + + profile.get_followers(queryer, &app_key).map(|followers| { let id_arr: Vec<String> = followers["ids"].as_array().unwrap().iter().map(|x| x.as_str().unwrap().to_owned()).collect(); - tweeter.set_followers(id_arr); + (maybe_my_name.unwrap().to_owned(), new_following, lost_following, profile.set_followers(id_arr)) + }) + }; + + match followers_changes { + Ok((my_name, new_following, lost_following, (new_followers, lost_followers))) => { + for user in new_following { + tweeter.display_info.status(format!("New following! {}", user)); + } + for user in lost_following { + tweeter.display_info.status(format!("Not following {} anymore", user)); + } + for user in new_followers { + tweeter.display_info.status(format!("New follower! {}", user)); + } + for user in lost_followers { + tweeter.display_info.status(format!("{} isn't following anymore", user)); + } }, Err(e) => { tweeter.display_info.status(e); @@ -830,6 +842,7 @@ fn handle_twitter_welcome( } pub fn handle_message( + conn_id: u8, twete: serde_json::Value, tweeter: &mut TwitterCache, queryer: &mut ::Queryer @@ -837,15 +850,15 @@ pub fn handle_message( match twete { serde_json::Value::Object(objmap) => { if objmap.contains_key("event") { - handle_twitter_event(objmap, tweeter, queryer); + handle_twitter_event(conn_id, objmap, tweeter, queryer); } else if objmap.contains_key("friends") { - handle_twitter_welcome(objmap, tweeter, queryer); + handle_twitter_welcome(conn_id, objmap, tweeter, queryer); } else if objmap.contains_key("delete") { - handle_twitter_delete(objmap, tweeter, queryer); + handle_twitter_delete(conn_id, objmap, tweeter, queryer); } else if objmap.contains_key("user") && objmap.contains_key("id") { - handle_twitter_twete(objmap, tweeter, queryer); + handle_twitter_twete(conn_id, objmap, tweeter, queryer); } else if objmap.contains_key("direct_message") { - handle_twitter_dm(objmap, tweeter, queryer); + handle_twitter_dm(conn_id, objmap, tweeter, queryer); } else { tweeter.display_info.status(format!("Unknown json: {:?}", objmap)); } |