Deadlock when many exchanges are crawled

[stepan-romankov]

I guess that the problem leaves in

pub(crate) fn create_conversion_thread(
    exchange: String,
    msg_type: MessageType,
    market_type: MarketType,
    tx: Sender<Message>,
) -> Sender<String> {
    let (tx_raw, rx_raw) = std::sync::mpsc::channel();
    tokio::task::spawn(async move {
        for json in rx_raw {
            let msg = Message::new(exchange.clone(), market_type, msg_type, json);
            if tx.send(msg).is_err() {
                break; // break the loop if there is no receiver
            }
        }
    });
    tx_raw
}

And exactly in

    tokio::task::spawn(async move {

I think that's because std::sync::mpsc::channel block the thread until data is available and official tokio documentation tells that this is not allowed (https://tokio.rs/tokio/tutorial/channels).

I see two options to fix this:

• To use 'tokio::task::spawn_blocking(move || {'
• change from usage of std::sync::mpsc::channel to tokio::sync::mpsc

[soulmachine]

Hi Stepan, thanks you for the work, I'll find time to review the code this weekend.

[soulmachine]

Merged the PR, closing this issue now.