Using blocking push events functions about slack-morphism-rust

[oren0e]

wow you're awesome - that solved it!
So now with this architecture I end up opening up a new session each time a message is sent. Is that normal? I assume the session gets closed at the end of the scope right?
I also left the two kinds of event subscriptions - it does not bother me that a mention message gets matched twice as long as I see one reply, which I do now.
Last question: If I'll make eval_code (and as a consequence execute() and create_share_link()) to be async, will I gain speed? And if so, then I still need to use tokio::spawn::(async move ...) or is it tokio::task::spawn_blocking() now?
This is the first project in Rust that I do with async so that's why I'm not sure about a lot of things

[oren0e]

Thank you very much for all the help! I really appreciate it.

[abdolence]

Not sure if I follow the exactly your particular issue, but you have always an ability to call blocking function inside async function (the opposite is more complicated though).
The recommended way to that would be something like this for Tokio:

fn my_blocking_function(
    event: SlackPushEventCallback,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    println!("{:#?}", event);
    Ok(())
}

async fn test_push_events_sm_function(
    event: SlackPushEventCallback,
    _client: Arc<SlackHyperClient>,
    _states: Arc<SlackClientEventsUserState>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    tokio::task::spawn_blocking(move || my_blocking_function(event)).await?
}

https://docs.rs/tokio/latest/tokio/task/fn.spawn_blocking.html

However, usually you will need to communicate with Slack API back using client and async methods again, so I don't think it is convenient option to work with the API.

[abdolence]

If you want to have some kind of synchronous queue to handle all incoming requests one by one then you can use Rust
https://docs.rs/tokio/latest/tokio/sync/mpsc/index.html
to have one consumer to handle it in some order.

[abdolence]

This example demonstrates how to work with mpsc:

async fn test_push_events_sm_function(
    event: SlackPushEventCallback,
    client: Arc<SlackHyperClient>,
    states: Arc<SlackClientEventsUserState>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let sender = {
        let state = states.read().unwrap();
        state
            .get_user_state::<MyEnvironment>()
            .unwrap()
            .sender
            .clone()
    };

    Ok(sender
        .send(MyEvent {
            client: client.clone(),
            event,
        })
        .map_err(|e| Box::new(e))?)
}

#[derive(Debug)]
pub struct MyEvent {
    pub client: Arc<SlackHyperClient>,
    pub event: SlackPushEventCallback,
}

pub struct MyEnvironment {
    pub sender: Arc<UnboundedSender<MyEvent>>,
}

async fn test_client_with_socket_mode() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let client = Arc::new(SlackClient::new(SlackClientHyperConnector::new()));

    let socket_mode_callbacks = SlackSocketModeListenerCallbacks::new()
        .with_command_events(test_command_events_function)
        .with_interaction_events(test_interaction_events_function)
        .with_push_events(test_push_events_sm_function);

    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<MyEvent>();

    tokio::spawn(async move {
        while let Some(message) = rx.recv().await {
            println!("Your message event: {:?}", message); // This is where you handle everything one by one now
        }
    });

    let listener_environment = Arc::new(
        SlackClientEventsListenerEnvironment::new(client.clone())
            .with_error_handler(test_error_handler)
            .with_user_state(MyEnvironment {
                sender: Arc::new(tx),
            }),
    );

[abdolence]

One more option would be blocking state which is RWMutex:

let state = states.write().unwrap();
// And do something in blocked state
...

[oren0e]

@abdolence Thank you for the detailed replies.
My problem is that I use your crate together with a simple reqwest Client that should communicate with playground (it's for a slack bot that can run Rust code). I suspect this client being async is a problem (although I am not sure) because when I issue a request I get multiple same replies back, in a weird timing - a bunch of replies followed by another batch after 5-10 seconds. I only ever want to get a single reply. I suspect this is due to async, but I'm not sure.

[abdolence]

@oren0e Ah, I probably know one of the reason.
For socket mode there is a retry from Slack if you don't return ack quickly and 5-10 seconds is definitely an issue. You have to release the callback function faster to avoid this. If you have some more time-consuming function you should call it on a separate "thread" using tokio spawn for example. The library is sending Acks to Slack as soon as you release callbacks, so this is why it is important.

So something like:

async fn test_push_events_sm_function(
    event: SlackPushEventCallback,
    _client: Arc<SlackHyperClient>,
    _states: Arc<SlackClientEventsUserState>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        tokio::spawn(async move {
           my_time_consuming_function(...)
        });
Ok(()) 
}

In this case, you are returning Ok quickly (and that's Ack underneath to Slack API). Generally speaking even if you use Events API it is good to return something to Slack no more than 2-3 seconds - otherwise Slack users may see some errors.

[oren0e]

@abdolence yeah I just happened to stumble across this stack overflow thread which says the same:
https://stackoverflow.com/questions/50715387/slack-events-api-triggers-multiple-times-by-one-message

I've tried what you suggested and it does not seem to work - it still send batches of identical replies. The code is here: https://github.com/oren0e/rusty-slackbot/blob/Refactor/Use_Slack_Morphism/src/bot.rs
The "problematic" function is eval_code which calls functions that interact with playground, and yes, this is the bottleneck of the flow. I also tried to make it async (consequently using reqwest async Client) - but the results were the same :/
I suspect that either:

1. I need to manually set a header with X-Slack-No-Retry: 1 like it says here https://api.slack.com/apis/connections/events-api#graceful_retries (is this possible in slack-morphism?)
2. or maybe it has to do with the fact that my app has 2 bot event subscriptions - app_mention and message.channels - I don't know if the last one is the proper one for listening to messages, but maybe it somehow catches the responses as well (as they are posted in the channel by the bot, but on the other hand - they shouldn't match, because they are not rust code)

[abdolence]

Regarding

app_mention and message.channels

The last one should give you all of the messages including with mentions, so you will receive sometimes 2 messages using both and you should handle it or stop using one of them.

I'll look at your code a bit later.

[abdolence]

https://github.com/oren0e/rusty-slackbot/blob/eb2ffe921c21b71f35ed0516b847f934f2158375/src/bot.rs#L32

Here you still blocking on_message callback. In your case I would recommend to relocate whole body of on_message - to another function (let's say it process_message) and just call it with spawning (not spawn_block/wait):

async fn on_message(
    event: SlackPushEventCallback,
    _client: Arc<SlackHyperClient>,
    _states: Arc<SlackClientEventsUserState>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        tokio::spawn(async move {
           process_message(...)
        });
Ok(()) 
}

[oren0e]

I moved it like you suggested:

pub async fn on_message(
    event: SlackPushEventCallback,
    client: Arc<SlackHyperClient>,
    _states: Arc<SlackClientEventsUserState>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    tokio::spawn(async move { process_message(client, event) });
    Ok(())
}

async fn process_message(
    client: Arc<SlackHyperClient>,
    event: SlackPushEventCallback,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let token_value =
        SlackApiTokenValue(env::var("SLACK_BOT_TOKEN").expect("SLACK_BOT_TOKEN env var not found"));
    let token = SlackApiToken::new(token_value);
    let session = client.open_session(&token);

    match event.event {...

But now there are no responses at all, not for mention and not for other messages. (eval_code is not an async function, maybe that matters?)

[abdolence]

But now there are no responses at all, not for mention and not for other messages. (eval_code is not an async function, maybe that matters?)

You also need to await on process_message - let's fix it first:

pub async fn on_message(
    event: SlackPushEventCallback,
    client: Arc<SlackHyperClient>,
    _states: Arc<SlackClientEventsUserState>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    tokio::spawn(async move { process_message(client, event).await; });
    Ok(())
}

Updated: eval_code doesn't matter for now

[abdolence]

So now with this architecture I end up opening up a new session each time a message is sent. Is that normal? I assume the session gets closed at the end of the scope right?

Those sessions is nothing more than just convenient way to have client and token reference to call multiple Slack API methods. Don't worry about them, you shouldn't be worried to create millions of them, they cost almost nothing and doesn't have any state/network calls/resources.

f I'll make eval_code (and as a consequence execute() and create_share_link()) to be async, will I gain speed? And if so, then I still need to use tokio::spawn::(async move ...) or is it tokio::task::spawn_blocking() now?

I don't think in your case it is about speed, more like blocking resources and threads which wasn't necessary. It feels weird that you actually have an http client that able to work with async as well (reqwest). It will remove the complexity with tokio::task::spawn_blocking() and will clean and simpilify the code if you start using the http client in async way as well :)

[abdolence]

One more thing, @oren0e I think it is worth to mention that it is quite important to have error handler defined:

• so you can return Ack even if you have errors happening (look at the example for socket mode for that)
• to see any issues with incoming messages in your logs

[oren0e]

@abdolence - what do you mean by error handler? I have this https://github.com/oren0e/rusty-slackbot/blob/Refactor/Use_Slack_Morphism/src/error.rs (needs to be edited a little bit). I actually wanted to have this custom error to be the return error in process_message().
If you refer to this https://slack-rust.abdolence.dev/socket-mode.html example then I don't see any error handling examples there.
I do plan to integrate the tracing crate for logs.

[abdolence]

This one:

slack-morphism-rust/src/hyper/examples/socket_mode.rs

Line 46 in d3cda52

fn test_error_handler(

[abdolence]

And this is how it is enabled:

slack-morphism-rust/src/hyper/examples/socket_mode.rs

Line 69 in d3cda52

.with_error_handler(test_error_handler),

[oren0e]

This one:

slack-morphism-rust/src/hyper/examples/socket_mode.rs

Line 46 in d3cda52

fn test_error_handler(

So if this should always return an Ack then this function is mostly for recording the errors to logs right? Something like

error!("There was an error of bla bla bla");

And for the solved problem - so if I understood correctly, separating the "heavy" tasks into another thread let the thread of the callbacks to release the callbacks more quickly while the other thread is handling the heavy tasks, thus allowing the library to send Ack faster to Slack which then prevented it from retrying the same request. Is that correct?

[abdolence]

Yes, all correct. 👍🏻