[Из песочницы] WebSockets и Spring WebFlux

15 ИЮНЯ 2022

------------------------------------------------------------------------

Наш телеграм-канал: t.me/@mailsgun_online

Youtube: youtube.com/@mailsgun_online

------------------------------------------------------------------------

Вступление

Все мы знаем вебсокеты, любим или не очень и можем написать их со Spring MVC.

А что на счет реактивного приложения?

В этой статье будет рассмотрено как создать вебсокеты с помощью Spring WebFlux.

Контент:

  1. Как сконфигурировать вебсокет.
  2. Как прочитать сообщение?
  3. Как отправить сообщение?
  4. Демо проект.
  5. Рекомендации как обезопасить вебсокеты.

Для создания проекта использовалась страница start.spring.io с зависимостью Reactive Web.

Как сконфигурировать вебсокет

Начнем с конфигурации, чтобы наглядно продемонтрировать почему нам нужен тот или иной компонет.

Первое что нужно сделать это написать конфигурацию для реактивного вебсокета.

Для этого нужно два бина: HandlerMapping и HamdlerAdapter.

HandlerAdapter создается довольно таки легко:

    @Bean
    public HandlerAdapter wsHandlerAdapter() {
        return new WebSocketHandlerAdapter();
    }

Зачем нужен этот бин?

Адаптер внутри себя инициализирует вебсокет сервис и использует (еще не написанную нами) имплементацию WebSocketHandler. Другими словами, этот бин включает вебсокет в Spring WebFlux.

Следующий бин это HandlerMapping. С помощью него мы можем настроить url и отвечающий за него WebSocketHandler, а так же порядок инициализации бина. Спринг будет знать как настоить нужный url и как его обрабатывать благодяря этой конфигурации. Наш url будет выглядить так http://localhost:8080/push, так как ниже мы настраиваем путь как /push.

    @Bean
    public HandlerMapping handlerMapping() {
    // url
        String path = "/push";
    // here webSocketHandler not defined yet
        Map<String, WebSocketHandler> map = Map.of(path, webSocketHandler);
    // -1 is order
        return new SimpleUrlHandlerMapping(map, -1);
    }

Конечно можно создавать сколько угодно url и WebSocketHandler.

Теперь можно создать имплементацию WebSocketHandler. С этим классом мы можем обрабатываь объект WebSocketSession, другими словами читать и отправлять сообщения.

На данный момент мы оставим Mono.empty(), но не волнуйтесь, ниже будет полноценная имплементация.

@Component
public class DefaultWebSocketHandler implements WebSocketHandler {
    @Override
    public Mono<Void> handle(WebSocketSession session) {
    return Mono.empty();
    }
}

Полный код конфигурации.

@Configuration
public class WebSocketConfig {
    private WebSocketHandler webSocketHandler;
    @Autowired
    public WebSocketConfig(WebSocketHandler webSocketHandler) {
        this.webSocketHandler = webSocketHandler;
    }
    @Bean
    public HandlerMapping handlerMapping() {
        String path = "/push";
        Map<String, WebSocketHandler> map = Map.of(path, webSocketHandler);
        return new SimpleUrlHandlerMapping(map, -1);
    }
    @Bean
    public HandlerAdapter wsHandlerAdapter() {
        return new WebSocketHandlerAdapter();
    }
}

Как прочитать сообщение?

Логика чтения может быть реализована в имплементации интерфейса WebSocketHandler.

Нам нужно использовать объект WebSocketSession и просто вызывать метод receive(), чтобы получить стрим и обработать его.

@Component
public class DefaultWebSocketHandler implements WebSocketHandler {
    @Override
    public Mono<Void> handle(WebSocketSession session) {
    session.receive().subscribe(message => {
        // process message here
    });
    return Mono.empty();
    }
}

Как отправить сообщение?

Перед тем как продолжить изменять WebSocketHandler нужно создать какуй-то простенький дата класс, который мы будем слать клиенту.

public class Event {
    private String name;
    private int count;
    public Event() {}
    public Event(String name, int count) {
        this.name = name;
        this.count = count;
    }
        // Getters, Setters, etc.
}

Так же нужно создать сервис, который сможет возвращать ивенты в виде реактивного стрима (как требует WebSocketSession).

Главное: нужно добавлсять новые ивенты в стрим.

Так что теперь может создать наш интерфейс:

public interface EventUnicastService {
    /**
     * Add message to stream
     * @param next - message which will be added to stream
     */
    void onNext(Event next);
    Flux<Event> getMessages();
}

Имплементация будет использовать EmitterProcessor, который подходит под все наши требования. Этот процессор может потреблять наши ивенты и раздавать подписчикам.

@Service
public class EventUnicastServiceImpl implements EventUnicastService {
    private EmitterProcessor<Event> processor = EmitterProcessor.create();
    @Override
    public void onNext(Event next) {
        processor.onNext(next);
    }
    @Override
    public Flux<Event> getMessages() {
        return processor.publish().autoConnect();
    }
}

Здесь:

  • publish() создает ConnectableFlux, который позволяет подписаться большому количеству объектов на Flux;
  • autoConnect() соеденяется с ConnectableFlux, когда кто-то выщывает метод subscribe у стрима.

Так же можно использовать метод replay(int), который кэширует указанное количество елементов и возвращает новым подписчикам.

Отправка сообщений будет выглядить вот так:


    @Override
    public Mono<Void> handle(WebSocketSession session) {
    Flux<WebSocketMessage> messages = unicastService.getMessages()
        .flatMap(o -> {
                    try {
                        return Mono.just(objectMapper.writeValueAsString(o)); // <- convert object to json
                    } catch (JsonProcessingException e) {
                        return Mono.error(e);
                    }
                })
            .map(session::textMessage);
    return session.send(messages);
    }

Как видно из кода выше, перед тем как отправить объект клиенту, его нужно сконвертировать в json. Spring WebFlux не предоставляет механизм для десериализации объекта в json для вебсокетов.

В примере используется Jackson, а именно ObjectMapper#writeValueAsString, так что на выходе мы имеем json строку.

Теперь можно объеденить чтение и отправку в DefaultWebSocketHandler.

@Component
public class DefaultWebSocketHandler implements WebSocketHandler {
    private EventUnicastService eventUnicastService;
    private ObjectMapper objectMapper;
    @Autowired
    public DefaultWebSocketHandler(EventUnicastService eventUnicastService, ObjectMapper objectMapper) {
        this.eventUnicastService = eventUnicastService;
        this.objectMapper = objectMapper;
    }
    @Override
    public Mono<Void> handle(WebSocketSession session) {
        Flux<WebSocketMessage> messages = session.receive()
                // .doOnNext(message -> { read message here or in the block below })
                .flatMap(message -> {
                    // or read message here
                    return eventUnicastService.getMessages();
                })
                .flatMap(o -> {
                    try {
                        return Mono.just(objectMapper.writeValueAsString(o));
                    } catch (JsonProcessingException e) {
                        return Mono.error(e);
                    }
                }).map(session::textMessage);
        return session.send(messages);
    }
}

Демо

Event generator

Для демо будет создан простой ивент генератор, который будет просто инкременитировать счетчик каждую секунду и отправлять ивент.

@Component
public class EventGenerator {
    private AtomicInteger counter = new AtomicInteger(0);
    private EventUnicastService eventUnicastService;
    @Autowired
    public EventGenerator(EventUnicastService eventUnicastService) {
        this.eventUnicastService = eventUnicastService;
    }
    @Scheduled(initialDelay = 1000, fixedDelay = 1000)
    public void generateEvent() {
        int count = counter.getAndIncrement();
        Event event = new Event("event", count);
        eventUnicastService.onNext(event);
    }
}

Не забываем включить Scheduler в приложении.

@SpringBootApplication
@EnableScheduling // <- enable scheduling!!!
public class WebfluxwebsocketsApplication

Клиент

Для клиента будет использоваться Angular.

Что нужно сделать:

Создать проект.

ng new projectname

Изменить app.component.ts.

import {Component, OnInit} from '@angular/core';
import {webSocket} from 'rxjs/webSocket';
@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit {
  title = 'websocket-angular';
  messages: string[] = [];
    // create WebSocket subject
  private subject = webSocket('ws://localhost:8080/push');
  ngOnInit(): void {
    this.subject.next({message: 'message'}); // <- ping first message
    this.subject.subscribe(message => {       // <- listen messages from server
      const event = message as Event
      this.messages.push(event.name + ' #' + event.count);
    });
  }
}

Здесь мы используем rxjs websockets.

Чтобы увидеть наши сообщения надо обновить app.component.html.

<div style="text-align:center">
  <h1>
    Welcome to {{ title }}!
  </h1>
</div>
<ul>
  <li *ngFor="let m of messages">{{m}}</li>
</ul>

Теперь можно запустить приложение и открыть в браузере localhost:4200.

Результат:

Рекомендации как обезопасить вебсокеты

  1. Используйте вебсокеты через SSL/TLS (wss:// протокол).
  2. Используйте валидацию данных клиета и сервера.
  3. Используйте вебсокет подключение после авторизации.
  4. Генерируйте уникальный ключ который можно получить по http и использовать в подключении к вебсокетам.
  5. Используйте Origin заголовок.

Выводы

В статье показана базовые кофигурация, чтение и отправка сообщений используя реактивные вебсокеты, которое было рассмотрено на примере демо проекта. Так же можно увидеть, что реализация упирается в знание Project Reactor.

Примеры кода можно найти в GitHub https://github.com/MaxNeutrino/snippets/tree/master/webflux-websocket

Теги:
Хабы:


Source: habr1

Ну и напоследок напоминаю вам о нашем новом проекте vkserfing bot.