commit e6ba2f4a0df9b95e118edd43b96478305c956ead
parent 86d1e49a2171cca5c545436ba30b49b15a7c0120
Author: Jonas Platte <jplatte+git@posteo.de>
Date: Sun, 4 Aug 2019 20:42:02 +0200
Implement basic event loop
Diffstat:
10 files changed, 180 insertions(+), 145 deletions(-)
diff --git a/.rustfmt.toml b/.rustfmt.toml
@@ -0,0 +1 @@
+merge_imports = true
diff --git a/Cargo.toml b/Cargo.toml
@@ -8,17 +8,24 @@ edition = "2018"
anymap = "0.12.1"
failure = "0.1.5"
futures-preview = "0.3.0-alpha.17"
-js_int = "0.1.2"
hyper = { git = "https://github.com/hyperium/hyper" }
hyper-tls = { git = "https://github.com/hyperium/hyper-tls" }
ruma-client = { git = "https://github.com/ruma/ruma-client", branch = "async_await" }
+tokio = { git = "https://github.com/tokio-rs/tokio" }
url = "2.0.0"
[dependencies.ruma-bot-macros]
path = "macros"
-[dev-dependencies.tokio]
-git = "https://github.com/tokio-rs/tokio"
+[patch.crates-io.ruma-events]
+git = "https://github.com/ruma/ruma-events"
+rev = "1b0be0d0e7fa1040fc51b748256b290e24677b19"
+
+[patch.'https://github.com/ruma/ruma-client-api'.ruma-client-api]
+# Cargo doesn't allow using a different branch of the same repository. Trick it into thinking that
+# we're using a different repository.
+git = "https://github.com/ruma/ruma-client-api.git"
+branch = "request-ctors"
[workspace]
members = ["macros"]
diff --git a/examples/echo.rs b/examples/echo.rs
@@ -0,0 +1,41 @@
+#![feature(async_await)]
+
+use std::{
+ collections::HashMap,
+ sync::{Arc, Mutex},
+};
+
+use failure::Fallible;
+use ruma_bot::{command_handler, Bot, BotBuilder, MsgContent, State};
+
+type AppState = Arc<Mutex<HashMap<String, String>>>;
+
+#[command_handler]
+async fn help(bot: Bot, msg_content: MsgContent) -> Fallible<()> {
+ println!("help called!");
+
+ Ok(())
+}
+
+//#[command_handler(commands = ["x", "y", "z"])]
+//async fn test(state: State<AppState>) -> Fallible<()> {
+// Ok(())
+//}
+
+//#[command_handler(command = "fetch {id}")]
+//#[command_handler(command = regex("a(B|CD) (\w+)"))]
+
+//#[reaction_handler]
+
+#[tokio::main]
+async fn main() {
+ let bot = BotBuilder::new()
+ //.state(Arc::new(Mutex::new(HashMap::<String, String>::new())))
+ .register(help)
+ .build()
+ .unwrap();
+
+ if let Err(e) = bot.run().await {
+ eprintln!("{}", e);
+ }
+}
diff --git a/macros/Cargo.toml b/macros/Cargo.toml
@@ -13,4 +13,4 @@ quote = "0.6.13"
[dependencies.syn]
version = "0.15.42"
-features = ["full", "extra-traits"]
-\ No newline at end of file
+features = ["full", "extra-traits"]
diff --git a/macros/src/lib.rs b/macros/src/lib.rs
@@ -29,7 +29,8 @@ pub fn command_handler(args: TokenStream, input: TokenStream) -> TokenStream {
let fn_ident = Ident::new(&format!("_{}_impl", handler_fn.ident), Span::call_site());
let ident = mem::replace(&mut handler_fn.ident, fn_ident.clone());
- let get_calls = (0..handler_fn.decl.inputs.len()).map(|_| quote!(param_matcher.get()));
+ // TODO: Error handling (required for State parameters)
+ let get_calls = (0..handler_fn.decl.inputs.len()).map(|_| quote!(param_matcher.get().unwrap()));
let mut commands = Vec::new();
for arg in macro_args.0 {
@@ -60,11 +61,18 @@ pub fn command_handler(args: TokenStream, input: TokenStream) -> TokenStream {
fn handle(
&self,
bot: &ruma_bot::Bot,
- ) -> Box<dyn futures::Future<Output = Result<(), failure::Error>>> {
+ msg_content: &str,
+ ) -> std::pin::Pin<Box<dyn futures::Future<Output = ()> + Send>> {
use ruma_bot::GetParam;
- let param_matcher = ruma_bot::HandlerParamMatcher { bot };
-
- Box::new(#fn_ident(#(#get_calls),*))
+ let param_matcher = ruma_bot::HandlerParamMatcher { bot, msg_content };
+ let fut = #fn_ident(#(#get_calls),*);
+
+ Box::pin(async move {
+ let res = fut.await;
+ if let Err(e) = res {
+ eprintln!("{}", e);
+ }
+ })
}
}
})
diff --git a/src/lib.rs b/src/lib.rs
@@ -6,14 +6,13 @@
//! [matrix]: https://matrix.org/
#![feature(async_await)]
-#![warn(missing_debug_implementations, missing_docs)]
+//#![warn(missing_debug_implementations, missing_docs)]
-use std::{collections::HashMap, sync::Arc};
+use std::{collections::HashMap, pin::Pin, sync::Arc};
use anymap::any::CloneAny;
use failure::{err_msg, Fallible};
use futures::{Future, TryStreamExt};
-use js_int::UInt;
use ruma_client::{api, HttpsClient as MatrixClient};
use url::Url;
@@ -23,12 +22,12 @@ type HandlerFnMap = HashMap<&'static str, Box<dyn CommandHandler>>;
pub use ruma_bot_macros::command_handler;
-mod state;
mod util;
+mod wrap;
-pub use state::State;
#[doc(hidden)]
pub use util::{GetParam, HandlerParamMatcher};
+pub use wrap::{MsgContent, State};
fn env_var(name: &'static str) -> Option<String> {
std::env::var(name)
@@ -48,10 +47,10 @@ pub trait CommandHandler: Send + Sync {
Self: Sized;
/// Used to call the command handler with the necessary application state and command context
- // TODO: Additional parameter(s) for command data
- fn handle(&self, _: &Bot) -> Box<dyn Future<Output = Result<(), failure::Error>>>;
+ fn handle(&self, _: &Bot, msg_content: &str) -> Pin<Box<dyn Future<Output = ()> + Send>>;
}
+#[derive(Clone)]
struct ConnectionDetails {
//homeserver_url: Url,
username: String,
@@ -134,6 +133,7 @@ impl BotBuilder {
}
}
+#[derive(Clone)]
pub struct Bot {
client: MatrixClient,
handlers: Arc<HandlerFnMap>,
@@ -143,93 +143,76 @@ pub struct Bot {
impl Bot {
pub async fn run(mut self) -> Result<(), ruma_client::Error> {
+ use api::r0::{
+ filter::{Filter, FilterDefinition, RoomEventFilter, RoomFilter},
+ sync::sync_events::Filter as SyncFilter,
+ };
+ use ruma_client::events::{
+ collections::all::RoomEvent,
+ room::message::{MessageEvent, MessageEventContent, TextMessageEventContent},
+ };
+
// A Bot is always instantiated with connection_details = Some(...) and this method (that
// can only be called once) is the only one method that uses connection_details. Thus, this
// .unwrap() will never fail.
let cd = self.connection_details.take().unwrap();
+ print!("logging in... ");
+
self.client.log_in(cd.username, cd.password, None).await?;
- use api::r0::{
- filter::{Filter, FilterDefinition, RoomEventFilter, RoomFilter},
- sync::sync_events::Filter as SyncFilter,
- };
+ println!("done!");
let mut sync0 = Box::pin(self.client.sync(
- Some(SyncFilter::FilterDefinition(FilterDefinition {
- event_fields: Vec::new(),
- event_format: None,
- account_data: Some(Filter {
- limit: Some(UInt::from(0u32)),
- //..Filter::default()
- senders: Vec::new(),
- not_senders: Vec::new(),
- types: Vec::new(),
- not_types: Vec::new(),
- }),
- room: Some(RoomFilter {
- account_data: Some(RoomEventFilter {
- limit: Some(UInt::from(0u32)),
- //..RoomEventFilter::default()
- rooms: Vec::new(),
- not_rooms: Vec::new(),
- senders: Vec::new(),
- not_senders: Vec::new(),
- types: Vec::new(),
- not_types: Vec::new(),
- }),
- timeline: Some(RoomEventFilter {
- limit: Some(UInt::from(0u32)),
- //..RoomEventFilter::default()
- rooms: Vec::new(),
- not_rooms: Vec::new(),
- senders: Vec::new(),
- not_senders: Vec::new(),
- types: Vec::new(),
- not_types: Vec::new(),
- }),
- ephemeral: Some(RoomEventFilter {
- limit: Some(UInt::from(0u32)),
- //..RoomEventFilter::default()
- rooms: Vec::new(),
- not_rooms: Vec::new(),
- senders: Vec::new(),
- not_senders: Vec::new(),
- types: Vec::new(),
- not_types: Vec::new(),
- }),
- state: Some(RoomEventFilter {
- limit: Some(UInt::from(0u32)),
- //..RoomEventFilter::default()
- rooms: Vec::new(),
- not_rooms: Vec::new(),
- senders: Vec::new(),
- not_senders: Vec::new(),
- types: Vec::new(),
- not_types: Vec::new(),
- }),
- //..RoomFilter::default()
- include_leave: None,
- rooms: Vec::new(),
- not_rooms: Vec::new(),
- }),
- presence: Some(Filter {
- limit: Some(UInt::from(0u32)),
- //..Filter::default()
- senders: Vec::new(),
- not_senders: Vec::new(),
- types: Vec::new(),
- not_types: Vec::new(),
- }),
- })),
+ Some(SyncFilter::FilterDefinition(FilterDefinition::ignore_all())),
None,
false,
));
let sync_start = sync0.try_next().await?.expect("sync response").next_batch;
- let mut sync_stream = Box::pin(self.client.sync(None, Some(sync_start), true));
+ let mut sync_stream = Box::pin(self.client.sync(
+ Some(SyncFilter::FilterDefinition(FilterDefinition {
+ account_data: Some(Filter::ignore_all()),
+ room: Some(RoomFilter {
+ account_data: Some(RoomEventFilter::ignore_all()),
+ ephemeral: Some(RoomEventFilter::ignore_all()),
+ state: Some(RoomEventFilter::ignore_all()),
+ ..Default::default()
+ }),
+ presence: Some(Filter::ignore_all()),
+ ..Default::default()
+ })),
+ Some(sync_start),
+ true,
+ ));
- while let Some(res) = sync_stream.try_next().await? {}
+ println!("initial sync finished!");
+
+ while let Some(res) = sync_stream.try_next().await? {
+ for (room_id, room) in res.rooms.join {
+ for event in room.timeline.events {
+ // Filter out the text messages
+ if let RoomEvent::RoomMessage(MessageEvent {
+ content: MessageEventContent::Text(TextMessageEventContent { body, .. }),
+ sender,
+ ..
+ }) = event
+ {
+ println!("received message `{}`", body);
+
+ if body.starts_with('!') {
+ if let Some(idx) = body.find(char::is_whitespace) {
+ let command = &body[1..idx];
+
+ if let Some(handler) = self.handlers.get(command) {
+ tokio::spawn(handler.handle(&self, &body[idx + 1..]));
+ }
+ }
+ }
+ }
+ }
+ }
+ }
Ok(())
}
diff --git a/src/state.rs b/src/state.rs
@@ -1,29 +0,0 @@
-use std::{
- fmt::{self, Debug},
- ops::Deref,
-};
-
-#[derive(Clone, Copy)]
-pub struct State<T: Send + Sync> {
- inner: T,
-}
-
-impl<T> Debug for State<T>
-where
- T: Debug + Send + Sync,
-{
- fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
- write!(f, "{:?}", self.inner)
- }
-}
-
-impl<T> Deref for State<T>
-where
- T: Send + Sync,
-{
- type Target = T;
-
- fn deref(&self) -> &T {
- &self.inner
- }
-}
diff --git a/src/util.rs b/src/util.rs
@@ -1,24 +1,31 @@
-use crate::{Bot, State};
+use crate::{Bot, MsgContent, State};
// Currently, GetParam could also be implemented direclty on &'a Bot, but in the future,
// event-specific data will be added to this struct.
pub struct HandlerParamMatcher<'a> {
pub bot: &'a Bot,
+ pub msg_content: &'a str,
}
pub trait GetParam<T> {
fn get(&self) -> Option<T>;
}
-impl<'a> GetParam<&'a Bot> for HandlerParamMatcher<'a> {
- fn get(&self) -> Option<&'a Bot> {
- Some(self.bot)
+impl GetParam<Bot> for HandlerParamMatcher<'_> {
+ fn get(&self) -> Option<Bot> {
+ Some(self.bot.clone())
}
}
// TODO: There should be a way of replacing 'static with 'a here without getting an error
-impl<'a, T: Clone + Send + Sync + 'static> GetParam<State<T>> for HandlerParamMatcher<'a> {
+impl<T: Clone + Send + Sync + 'static> GetParam<State<T>> for HandlerParamMatcher<'_> {
fn get(&self) -> Option<State<T>> {
self.bot.state.get().cloned()
}
}
+
+impl GetParam<MsgContent> for HandlerParamMatcher<'_> {
+ fn get(&self) -> Option<MsgContent> {
+ Some(MsgContent(self.msg_content.to_owned()))
+ }
+}
diff --git a/src/wrap.rs b/src/wrap.rs
@@ -0,0 +1,40 @@
+use std::{
+ fmt::{self, Debug},
+ ops::Deref,
+};
+
+#[derive(Clone, Copy)]
+pub struct State<T: Send + Sync> {
+ inner: T,
+}
+
+impl<T> Debug for State<T>
+where
+ T: Debug + Send + Sync,
+{
+ fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+ write!(f, "{:?}", self.inner)
+ }
+}
+
+impl<T> Deref for State<T>
+where
+ T: Send + Sync,
+{
+ type Target = T;
+
+ fn deref(&self) -> &T {
+ &self.inner
+ }
+}
+
+#[derive(Clone)]
+pub struct MsgContent(pub String);
+
+impl Deref for MsgContent {
+ type Target = str;
+
+ fn deref(&self) -> &str {
+ &self.0
+ }
+}
diff --git a/tests/echo.rs b/tests/echo.rs
@@ -1,22 +0,0 @@
-#![feature(async_await)]
-
-use failure::Fallible;
-use ruma_bot::{command_handler, BotBuilder};
-
-#[command_handler]
-async fn help() -> Fallible<()> {
- Ok(())
-}
-
-#[command_handler(commands = ["x", "y", "z"])]
-async fn test() -> Fallible<()> {
- Ok(())
-}
-
-//#[command_handler(command = "fetch {id}")]
-//#[command_handler(command = regex("a(B|CD) (\w+)"))]
-
-#[tokio::main]
-async fn main() {
- let bot = BotBuilder::new().register(help).build();
-}
