Изучаю Rust: Как я UDP чат сделал c Azul

Azul

Краткий обзор остальных фреймворков

• Orbtk — Почти что то же самое, что и Azul и так же в глубокой альфе
• Conrod — Видео Можно создавать кросс платформенные десктопные приложения.
• Yew — WebAssembly и похож на React. Для веб разработки.

Клиент

Структура, в которой группируются вспомогательные функции для чтения и записи в сокет

struct ChatService {}

impl ChatService {
    //1 
    fn read_data(socket: &Option<UdpSocket>) -> Option<String> {
        //2 
        let mut buf = [0u8; 4096];
        match socket {
            Some(s) => {
                //3 
                match s.recv(&mut buf) {
                    //4 
                    Ok(count) => Some(String::from_utf8(buf[..count].into())
                        .expect("can't parse to String")),
                    Err(e) => {
                        //5
                        println!("Error {}", e);
                        None
                    }
                }
            }
            _ => None,
        }
    }
    //6
    fn send_to_socket(message: String, socket: &Option<UdpSocket>) {
        match socket {
            //7 
            Some(s) => { s.send(message.as_bytes()).expect("can't send"); }
            _ => return,
        }
    }
}
 

1. Читаем денные из сокета
2. Буфер для данных, которые будем считывать из сокета.
3. Блокирующий вызов. Здесь поток выполнения останавливается до тех пор, пока не будут считанны данные или произойдет таймаут.
4. Получаем строку из массива байт в кодировке UTF8.
5. Сюда мы попадаем, если соединение оборвалось по таймауту или произошла другая ошибка.
6. Отправляет строку в сокет.
7. Преобразуем строку в байты в кодировке UTF8 и отправляем данные в сокет. Запись данных в сокет не блокирующая, т.е. поток выполнения продолжит свою работу. Если отправить данные не удалось, то прерываем работу программы с сообщением «can't send».

Структура, которая группирует функции для обработки событий, поступающих от пользователя, и изменения нашей DataModel

struct Controller {}
//1 
const TIMEOUT_IN_MILLIS: u64 = 2000;

impl Controller {
    //2 
    fn send_pressed(app_state: &mut azul::prelude::AppState<ChatDataModel>, _event: azul::prelude::WindowEvent<ChatDataModel>) -> azul::prelude::UpdateScreen {
        //3 
        let data = app_state.data.lock().unwrap();
        //4
        let message = data.messaging_model.text_input_state.text.clone();
         data.messaging_model.text_input_state.text  =  "".into();
        //5 
        ChatService::send_to_socket(message, &data.messaging_model.socket);
        //6
        azul::prelude::UpdateScreen::Redraw
    }
    //7 
    fn login_pressed(app_state: &mut azul::prelude::AppState<ChatDataModel>, _event: azul::prelude::WindowEvent<ChatDataModel>) -> azul::prelude::UpdateScreen {
        //8
        use std::time::Duration;
        //9 
        if let Some(ref _s) = app_state.data.clone().lock().unwrap().messaging_model.socket {
            return azul::prelude::UpdateScreen::DontRedraw;
        }
        //10
        app_state.add_task(Controller::read_from_socket_async, &[]);
        //11        app_state.add_daemon(azul::prelude::Daemon::unique(azul::prelude::DaemonCallback(Controller::redraw_daemon)));
        //12 
        let mut data = app_state.data.lock().unwrap();
        //13 
        let local_address = format!("127.0.0.1:{}", data.login_model.port_input.text.clone().trim());
        //14 
        let socket = UdpSocket::bind(&local_address)
            .expect(format!("can't bind socket to {}", local_address).as_str());
        //15 
        let remote_address = data.login_model.address_input.text.clone().trim().to_string();
        //16 
        socket.connect(&remote_address)
            .expect(format!("can't connect to {}", &remote_address).as_str());
        //17 
        socket.set_read_timeout(Some(Duration::from_millis(TIMEOUT_IN_MILLIS)))
            .expect("can't set time out to read");
        // 18
        data.logged_in = true;
        // 19
        data.messaging_model.socket = Option::Some(socket);
        //20 
        azul::prelude::UpdateScreen::Redraw
    }
    //21 
    fn read_from_socket_async(app_data: Arc<Mutex<ChatDataModel>>, _: Arc<()>) {
        //22 
        let socket = Controller::get_socket(app_data.clone());
        loop {
            //23 
            if let Some(message) = ChatService::read_data(&socket) {
                //24 
                app_data.modify(|state| {
                    //25 
                    state.messaging_model.has_new_message = true;
                    //26
                    state.messaging_model.messages.push(message);
                });
            }
        }
    }
    //27
    fn redraw_daemon(state: &mut ChatDataModel, _repres: &mut azul::prelude::Apprepres) -> (azul::prelude::UpdateScreen, azul::prelude::TerminateDaemon) {
        //28 
        if state.messaging_model.has_new_message {
            state.messaging_model.has_new_message = false;
            (azul::prelude::UpdateScreen::Redraw, azul::prelude::TerminateDaemon::Continue)
        } else {
            (azul::prelude::UpdateScreen::DontRedraw, azul::prelude::TerminateDaemon::Continue)
        }
    }
    //29 
    fn get_socket(app_data: Arc<Mutex<ChatDataModel>>) -> Option<UdpSocket> {
        //30 
        let ref_model = &(app_data.lock().unwrap().messaging_model.socket);
        //31 
        match ref_model {
            Some(s) => Some(s.try_clone().unwrap()),
            _ => None
        }
    }
}

1. Таймаут в миллисекундах, после которого будет прервана блокирующая операция чтения из сокета.
2. Функция отрабатывает, когда пользователь хочет оправить новое сообщение на сервер.
3. Получаем во владение мютекс с нашей моделью данных. Это блокирует поток перерисовки интерфейса до тех пор, пока мютекс не будет освобожден.
4. Делаем копию введенного пользователем текста, чтобы передать его дальше и очищаем поле для ввода текста.
5. Отправляем сообщение.
6. Сообщаем Фреймворку, что после обработки этого события нужно перерисовать интерфейс.
7. Функция отрабатывает, когда пользователь хочет подключиться к серверу.
8. Подключаем структуру для представления отрезка времени из стандартной библиотеки.
9. Если мы уже подключены к серверу, то прерываем выполнение функции сообщаем Фреймворку, что нет необходимости перерисовывать интерфейс.
10. Добавляем задачу, которая будет выполняться асинхронно в потоке из пула потоков Фреймворка Azul. Обращение к мютексу с моделью данных блокирует обновление UI до тех пор, пока мютекс не освободится.
11. Добавляем повторяющуюся задачу, которая выполняется в основном потоке. Любые длительные вычисления в этом демоне блокирует обновление интерфейса.
12. Получаем во владение мютекс.
13. Считываем введенный пользователем порт и создаем на основе него локальный адрес, будем прослушивать.
14. Создаем UDP сокет, который считывает пакеты, приходящие на локальный адрес.
15. Считываем введенный пользователем адрес сервера.
16. Говорим нашему UDP сокету читать пакеты только от этого сервера.
17. Устанавливаем таймаут для операции чтения из сокета. Запись в сокет происходит без ожидания, т. е. мы просто пишем данные и не ждем ничего, а операция чтения из сокета блокирует поток и ждет пока не придут данные, которые можно считать. Если не установить таймаут, то операция чтения из сокета будет ждать бесконечно.
18. Устанавливаем флаг, указывающий на то, что пользователь уже подключился к серверу.
19. Передаем в модель данных созданный сокет.
20. Сообщаем Фреймворку, что после обработки этого события нужно перерисовать интерфейс.
21. Асинхронная операция, выполняющаяся в пуле потоков Фреймворка Azul.
22. Получаем копию сокета из нашей модели данных.
23. Пытаемся прочитать данные из сокета. Если не сделать копию сокета и напрямую ждать тут, пока придёт сообщение из сокета, который в мютекс в нашей модели денных, то весь интерфейс перестанет обновляться до тех пор, пока мы не освободим мютекс.
24. Если нам пришло какое-то сообщение, то изменяем нашу модель данных modify делает то же, что и lock().unwrap() с передачей результата в лямбду и освобождением мютекс после того, как закончится код лямбды.
25. Устанавливаем флаг, обозначающий, что у нас новое сообщение.
26. Добавляем сообщение в массив всех сообщения чата.
27. Повторяющаяся синхронная операция работающая в основном потоке.
28. Если у нас есть новое сообщение, то сообщаем Фреймворку, что нужно перерисовать интерфейс с нуля и продолжить работу этого демона иначе не рисуем интерфейс с начала, но все равно вызываем этот Функция в следующем цикле.
29. Создает копию нашего сокета для того, чтобы не держать заблокированным мютекс с нашей моделью данных.
30. Получаем во владение мютекс и получаем ссылку на сокет.
31. Создаем копию сокета. Мютекс освободится автоматически при выходе из Функцияа.

Асинхронная обработка данных и демоны в Azul

// Problem - blocks UI :(
fn start_connection(app_state: &mut AppState<MyDataModel>, _event: WindowEvent<MyDataModel>) -> UpdateScreen {
    //Добавляем асинхронную задачу
    app_state.add_task(start_async_task, &[]);
    //Добавляем демон
    app_state.add_daemon(Daemon::unique(DaemonCallback(start_daemon)));
    UpdateScreen::Redraw
}

fn start_daemon(state: &mut MyDataModel, _repres: &mut Apprepres) -> (UpdateScreen, TerminateDaemon) {
    //Блокирует UI на десять секунд
    thread::sleep(Duration::from_secs(10));
        state.counter += 10000;
        (UpdateScreen::Redraw, TerminateDaemon::Continue)
}

fn start_async_task(app_data: Arc<Mutex<MyDataModel>>, _: Arc<()>) {
     // simulate slow load
    app_data.modify(|state| {
        //Блокирует UI на десять секунд
        thread::sleep(Duration::from_secs(10));
        state.counter += 10000;
    });
}

fn start_async_task(app_data: Arc<Mutex<MyDataModel>>, _: Arc<()>) {
    //Не блокируем UI. Ожидаем асинхронно.
    thread::sleep(Duration::from_secs(10));
    app_data.modify(|state| {       
        state.counter += 10000;
    });
}

Наши стили

const CUSTOM_CSS: &str = "
.row { height: 50px; }
.orange {
    background: linear-gradient(to bottom, #f69135, #f37335);
    font-color: white;
    border-bottom: 1px solid #8d8d8d;
}";

Собственно, Функции для создания нашего DOM для отображения его пользователю

impl azul::prelude::Layout for ChatDataModel {
    //1 
    fn layout(&self, info: azul::prelude::WindowInfo<Self>) -> azul::prelude::Dom<Self> {
        //2 
        if self.logged_in {
            self.chat_form(info)
        } else {
            self.login_form(info)
        }
    }
}

impl ChatDataModel {
    //3 
    fn login_form(&self, info: azul::prelude::WindowInfo<Self>) -> azul::prelude::Dom<Self> {
        //4 
        let button = azul::widgets::button::Button::with_label("Login")
            //5 
            .dom()
            //6 
            .with_class("row")
            //7 
            .with_class("orange")
            //8 
            .with_callback(
                azul::prelude::On::MouseUp,
                azul::prelude::Callback(Controller::login_pressed));

        //9 
        let port_label = azul::widgets::label::Label::new("Enter port to listen:")
            .dom()
            .with_class("row");
        //10 
        let port = azul::widgets::text_input::TextInput::new()
            //11 
            .bind(info.window, &self.login_model.port_input, &self)
            .dom(&self.login_model.port_input)
            .with_class("row");

        // 9
        let address_label = azul::widgets::label::Label::new("Enter server address:")
            .dom()
            .with_class("row");

        //10
        let address = azul::widgets::text_input::TextInput::new()
             //11
            .bind(info.window, &self.login_model.address_input, &self)
            .dom(&self.login_model.address_input)
            .with_class("row");

        //12 
        azul::prelude::Dom::new(azul::prelude::NodeType::Div)
            .with_child(port_label)
            .with_child(port)
            .with_child(address_label)
            .with_child(address)
            .with_child(button)
    }

    //13 
    fn chat_form(&self, info: azul::prelude::WindowInfo<Self>) -> azul::prelude::Dom<Self> {
        //14 
        let button = azul::widgets::button::Button::with_label("Send")
            .dom()
            .with_class("row")
            .with_class("orange")
            .with_callback(azul::prelude::On::MouseUp, azul::prelude::Callback(Controller::send_pressed));
        //15 
        let text = azul::widgets::text_input::TextInput::new()
            .bind(info.window, &self.messaging_model.text_input_state, &self)
            .dom(&self.messaging_model.text_input_state)
            .with_class("row");
        //12
        let mut dom = azul::prelude::Dom::new(azul::prelude::NodeType::Div)
            .with_child(text)
            .with_child(button);
        //16
        for i in &self.messaging_model.messages {
            dom.add_child(azul::widgets::label::Label::new(i.clone()).dom().with_class("row"));
        }
        dom
    }
}

1. Функция, которая создает конечный DOM, и вызывается каждый раз, когда нужно перерисовать интерфейс.
2. Если мы уже подключены к серверу, то показываем форму для отправки и чтения сообщений, иначе отображаем форму для подключения к серверу.
3. Создает форму для ввода данных необходимых для подключения к серверу.
4. Создаем кнопку с текстовой надписью Login.
5. Преобразуем ее в объект DOM.
6. Добавляем ей класс row.
7. Добавляем ей css класс orange.
8. Добавляем обработчик события для нажатия на кнопку.
9. Создаем текстовую метку с текстом для отображения пользователю и css классом row.
10. Создаем текстовое поле для ввода текста с текстом из свойства нашей модели и css классом row.
11. Привязываем текстовое поле к свойству нашей DataModel. Это двухсторонняя привязка. Теперь редактирование TextInput автоматически изменяет текст в свойстве нашей модели и обратное тоже верно. Если мы изменим текст в нашей модели, то изменится текст в TextInput.
12. Создаем корневой DOM элемент, в который помещаем наши UI элементы.
13. Создает форму для отправки и чтения сообщений.
14. Создаем кнопку с текстом «Send» и css классами «row», «orange» и обработчиком события при ее нажатии.
15. Создаем поле для ввода текста с двухсторонней привязкой с свойству модели self.messaging_model.text_input_state и css классом «row».
16. Добавляем текстовые метки, которые отображают сообщения, которые были написаны в чате.

Наша модель, которая хранит состояние нашего интерфейса

//1 
#[derive(Debug)]
//2 
struct ChatDataModel {
    //3 
    logged_in: bool,
    //4 
    messaging_model: MessagingDataModel,
    //5 
    login_model: LoginDataModel,
}

#[derive(Debug, Default)]
struct LoginDataModel {
    //6 
    port_input: azul::widgets::text_input::TextInputState,
    //7 
    address_input: azul::widgets::text_input::TextInputState,
}

#[derive(Debug)]
struct MessagingDataModel {
    //8
    text_input_state: azul::widgets::text_input::TextInputState,
    //9 
    messages: Vec<String>,
    //10 
    socket: Option<UdpSocket>,
    //11 
    has_new_message: bool,
}

1. Это позволит отображать нашу структуру в виде строки в шаблоне вида {:?}
2. Наша модель данных. Для того, чтобы ее можно было использовать в Azul. Она обязательно должна реализовать трейт Layout.
3. Флаг для проверки того, подключен ли пользователь к серверу или нет.
4. Модель для отображения формы для отправки сообщений на сервер и сохранения полученных с сервера сообщений.
5. Модель для отображения формы для подключения к серверу.
6. Порт, который ввел пользователь. Мы будем его прослушивать нашим сокетом.
7. Адрес сервера, который ввел пользователь. Мы будем к нему подключаться.
8. Сообщение пользователя. Мы его отправим на сервер.
9. Массив сообщений, который пришел с сервера.
10. Сокет, через который мы общаемся с сервером.
11. Флаг для проверки того, пришло ли нам новое сообщение от сервера.

И, наконец, главная точка входа в приложение. Запускает цикл от рисовки GUI и обработки ввода пользователя

pub fn run() {
    //1
    let app = azul::prelude::App::new(ChatDataModel {
        logged_in: false,
        messaging_model: MessagingDataModel {
            text_input_state: azul::widgets::text_input::TextInputState::new(""),
            messages: Vec::new(),
            socket: None,
            has_new_message: false,
        },
        login_model: LoginDataModel::default(),
    }, azul::prelude::AppConfig::default());
    // 2
    let mut style = azul::prelude::css::native();
    //3 
    style.merge(azul::prelude::css::from_str(CUSTOM_CSS).unwrap());
    //4 
    let window = azul::prelude::Window::new(azul::prelude::WindowCreateOptions::default(), style).unwrap();
    //5 
    app.run(window).unwrap();
}

1. Создаем приложение со стартовыми данными.
2. Стили, используемые приложением по умолчанию.
3. Добавляем к ним наши собственные стили.
4. Создаем окно, в котором будет отображать наше приложение.
5. Запускаем приложение в этом окне.

Сервер

Главная точка входа в приложение

pub fn run() {
    //1 
    let socket = create_socket();
    //2 
    let (sx, rx) = mpsc::channel();
    //3 
    start_sender_thread(rx, socket.try_clone().unwrap());
    loop {
        //4 
        sx.send(read_data(&socket)).unwrap();
    }
}

1. Создаем сокет.
2. Создаем односторонний канал с одним отправителем сообщений sx и множеством получателей rx.
3. Запускаем рассылку сообщений всем получателям в отдельном потоке.
4. Читаем данные из сокета и оправляем их в поток, занимающийся рассылкой сообщений клиентам, подключенным к серверу.

Функция для создания потока для рассылки сообщений клиентам

fn start_sender_thread(rx: mpsc::Receiver<(Vec<u8>, SocketAddr)>, socket: UdpSocket) {
    //1 
    thread::spawn(move || {
        //2 
        let mut addresses = Vec::<SocketAddr>::new();
        //3 
        loop {
            //4 
            let (bytes, pre) = rx.recv().unwrap();
            // 5
            if !addresses.contains(&pre) {
                println!(" {} connected to server", pre);
                addresses.push(pre.clone());
            }
            //6 
            let result = String::from_utf8(bytes)
                .expect("can't parse to String")
                .trim()
                .to_string();
            println!("received {} from {}", result, pre);
            //7 
            let message = format!("FROM: {} MESSAGE: {}", pre, result);
            let data_to_send = message.as_bytes();
            //8 
            addresses
                .iter()
                .for_each(|s| {
                    //9 
                    socket.send_to(data_to_send, s)
                          //10 
                        .expect(format!("can't send to {}", pre).as_str());
                });
        }
    });
}

1. Запускаем новый поток. move значит, что переменные переходят во владение лямбды и потока, соответственно. Конкретнее, наш новый поток «поглотит» переменные rx и socket.
2. Коллекция адресов, подключенных к нам клиентов. Всем им мы будем рассылать наши сообщения. Вообще, в реальном проекте надо бы сделать обработку отключения от нас клиента и удаления его адреса из этого массива.
3. Запускаем бесконечный цикл.
4. Читаем данные из канала. Тут поток будет заблокирован до тех пор, пока не придут новые данные.
5. Если такого адреса нет в нашем массиве, то добавляем его туда.
6. Декодируем UTF8 строку из массива байт.
7. Создаем массив байт, которые собираемся отправить всем нашим клиентам.
8. Проходим по коллекции адресов и отправляем данные каждому.
9. Операция записи в UDP сокет неблокирующая, поэтому здесь Функция не будет ждать пока сообщение придёт к получателю и выполнится почти мгновенно.
10. expect в случае ошибки сделает экстренный выход из программы с заданным сообщением.

Функция создает сокет на основе данных введенных пользователем

const TIMEOUT_IN_MILLIS: u64 = 2000;
 
fn create_socket() -> UdpSocket {
    println!("Enter port to listen");
    //1 
    let local_port: String = read!("{}\n");
    let local_address = format!("127.0.0.1:{}", local_port.trim());
    println!("server address {}", &local_address);
    //2 
    let socket = UdpSocket::bind(&local_address.trim())
        .expect(format!("can't bind socket to {}", &local_address).as_str());
    //3 
    socket.set_read_timeout(Some(Duration::from_millis(TIMEOUT_IN_MILLIS)))
        .expect("can't set time out to read");
    //4 
    socket
}

1. Считываем порт, который будет слушать наш сервер, и создаем на его основе локальный адрес сервера.
2. Создаем UDP сокет, прослушивающий этот адрес.
3. Устанавливаем таймаут для операции чтения. Операция чтения блокирующая и она заблокирует поток до тех пор, пока не придут новые данные или не наступит таймаут.
4. Возвращаем из Функции созданные сокет.
5. Функция читает данные из сокета и возвращает их вместе с адресом отправителя.

Функция для чтения данных из сокета

fn read_data(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
    //1 
    let mut buf = [0u8; 4096];
    //2 
    loop {
        match socket.recv_from(&mut buf) {
            //3 
            Ok((count, address)) => {
                //4 
                return (buf[..count].into(), address);
            }
            //5 
            Err(e) => {
                println!("Error {}", e);
                continue;
            }
        };
    }
}

1. Буфер — место, куда будем считывать данные.
2. Запускает цикл, который будет выполняться до тех пор, пока не будут считаны валидные данные.
3. Получаем количество считанных байт и адрес отправителя.
4. Делаем срез массива от его начала до количеств, считанных байт и преобразуем его в вектор байт.
5. Если произошёл таймаут или другая ошибка, то переходим к следующей итерации цикла.

Про слои в приложении

class User
{
    public string FirstName { get; set; }
    public string LastName { get; set; }
    public string FullName
    {
        get
        {
            return FirstName + " " + LastName;
        }
    }

    public bool Equal(User user)
    {
        return this.FullName == user.FullName;
    }
}

using System;
using System.Collections.Generic;
using System.Text;

//Entities 
class User
{
    public int Id { get; set; }
    public string Name { get; set; }
}

//Data Access Layer 

   class UserRepository
{
    private readonly Dictionary<int, User> _db;

    public UserRepository()
    {
        _db = new Dictionary<int, User>();
    }

    public User Get(int id)
    {
        return _db[id];
    }

    public void Save(User user)
    {
        _db[user.Id] = user;
    }
}

//Business Logic Layer 

class UserService
{
    private readonly UserRepository _repo;
    private int _currentId = 0;
    public UserService()
    {
        _repo = new UserRepository();
    }

    public void AddNew()
    {
        _currentId++;
        var user = new User
        {
            Id = _currentId,
            Name = _currentId.ToString()
        };
        _repo.Save(user);
    }

    public string GetAll()
    {
        StringBuilder sb = new StringBuilder();

        for (int i = 1; i <= _currentId; i++)
        {
            sb.AppendLine($"Id: {i} Name: {_repo.Get(i).Name}");
        }
        return sb.ToString();
    }
}

//presentation Layer aka Application Layer

class UserController
{
    private readonly UserService _service;
    public UserController()
    {
        _service = new UserService();
    }

    public string RunExample()
    {
        _service.AddNew();
        _service.AddNew();
        return _service.GetAll();
    }
}

namespace ConsoleApp1
{
    class Program
    {
        static void Main(string[] args)
        {
            var controller = new UserController();
            Console.WriteLine(controller.RunExample());
            Console.ReadLine();
        }
    }
}

P.S.