WebSockety – Część 1

Zgodnie z wynikami ankiety na naszej grupie (jeśli jeszcze do niej nie dołączyłeś zapraszam Cię do tego -> https://www.facebook.com/groups/devopsi . Będziesz miał realny wpływ na to o czym będziemy pisać.) dzisiejszy wpis poświęcę Websocketom. Czym są WebSockety? Kiedy i w jaki sposób z nich korzystać?

Wpis o websocketach zostanie podzielony na 3 mniejsze wpisy:

  1. Dzisiejszy: trochę teorii + Java;
  2. Pojawi się pod koniec lutego – dotyczyć będzie frontendu;
  3. Pojawi się na początku marca – połączymy Java i frontend oraz uruchomimy naszą aplikację.

Websockety – czym są?

Klienci (m.in. przeglądarki internetowe) i serwery (m.in. aplikacje backendowe) komunikują się między sobą poprzez protokół HTTP/S który to wspiera żądania (np. z przeglądarki) i odpowiedzi na nie (np. z aplikacji).

Tyle teorii. Jak to wygląda w praktyce?

Wpisujemy w przeglądarce adres naszej aplikacji.

Aplikacja (klient) wysyła żądanie do serwera.

Następnie nawiązywane jest połączenie pomiędzy klientem a serwerem.

Odpowiedź jest wysyłana przez serwer do klienta.

Połączenie zostaje zakończone.

Połączenie HTTP/S spełniło swoją rolę – skomunikowało Klienta i Serwer, więc może zostać zakończone. A co w przypadku kiedy chcielibyśmy, aby połączenie trwało? Przykładowo piszemy aplikację, która ma pokazywać notowania w czasie rzeczywistym. Idąc tropem komunikacji HTTP/S możemy natrafić na pojęcie HTTP Pooling, czyli odpytywanie HTTP. W praktyce będzie wyglądać to następująco:

Klient w 1s: Czy notowania Apple się zmieniły?

Serwer w 1s: Nie.

Klient w 2s: Czy notowania Apple się zmieniły?

Serwer w 2s: Nie.

i tak dalej co sekundę.

Jak można się domyślić rozwiązanie to jest bardzo mało wydajne. Przy takiej komunikacji zużywamy dużo zasobów a dodatkowym problemem jest liczba nieudanych żądań. W tym momencie z odsieczą przychodzi nam technika Long Pooling. Long Pooling utrzymuje połączenie po nawiązaniu go, ale jednak jest bardzo zasobożerny. Dlatego wracamy do naszych Websocketów.

WebSocket nie wymaga wysłania żądania w celu udzielenia odpowiedzi. Nawiązuje połączenie z serwerem i nasłuchuje danych. Serwer jak tylko dostanie dane przekaże je klientowi w ramach nawiązanego połączenia.

Jak to wygląda w praktyce?

Klient wysyła żądanie HTTP do serwera.

Nawiązywane jest połączenie HTTP klient-serwer.

Jeśli serwer obsługuje protokół WebSocket następuje aktualizacja połączenia.

Następuje zmiana połączenia HTTP na WebSocket.

Tyle teorii.

Websockety w Java

Do naszej implementacji użyjemy protokołu STOMP.

Zaczynamy od wygenerowania projektu w Spring Initializr

W moim wypadku wygląda to tak:

Oczywiście nie narzucam wyboru Mavena czy Javy 11, ale WebSocket i Spring Web jako Dependencies są obowiązkowe 🙂

Następnie musimy dodać dodatkowe dependency:

<dependency>
            <groupId>org.springdoc</groupId>
            <artifactId>springdoc-openapi-ui</artifactId>
            <version>${springdoc.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springdoc</groupId>
            <artifactId>springdoc-openapi-data-rest</artifactId>
            <version>${springdoc.version}</version>
        </dependency>
        <dependency>
            <groupId>com.h2database</groupId>
            <artifactId>h2</artifactId>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

Następnie musimy utworzyć nasz model. Przyjmijmy, że będziemy się zajmować notowaniami giełdowymi, więc nasz model będzie miał 4 elementy:

  • name
  • purchaseLimit
  • saleLimit
  • rate
import lombok.AllArgsConstructor;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.Setter;

@AllArgsConstructor
@Getter
@Setter
@EqualsAndHashCode
public class Stock {
    public String name;
    public Double purchaseLimit;
    public Double saleLimit;
    public Double rate;
}

W kolejnym kroku tworzymy nasz serwis. Zamockujemy sobie w nim dane, z których będziemy korzystać inicjalnie, oraz dodamy metodę odpowiadającą za randomowe wypychanie danych na frontend.

package pl.stompexample.websockets.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.messaging.SessionDisconnectEvent;
import pl.stompexample.websockets.model.Stock;

import javax.annotation.PostConstruct;
import java.util.*;

@Service
@Slf4j
public class StockService {
    private final Set<String> listenerClients = new HashSet<>();
    private final List<Stock> stock=new LinkedList<>();

    @PostConstruct
    private void putData(){
        stock.add(new Stock("Alior",38.80,16.6650,16.6300));
        stock.add(new Stock("CCC",368.80,16.6650,17.6300));
        stock.add(new Stock("CDPROJECT",368.80,16.6650,145.6300));
        stock.add(new Stock("CyfrPLSat",368.80,16.6650,1644.6300));
        stock.add(new Stock("DINOPL",68.80,13.6650,16.6300));
        stock.add(new Stock("JSW",332.80,16.6650,10.6300));
        stock.add(new Stock("KGHM",368.80,16.6650,16.6300));
        stock.add(new Stock("LOTOS",38.80,14.6650,16.6300));
        stock.add(new Stock("LPP",368.80,16.6650,16.6300));
        stock.add(new Stock("MBANK",368.80,16.6650,16.6300));
        stock.add(new Stock("ORANGEPL",368.80,16.6650,1.6300));
        stock.add(new Stock("PEKAO",368.80,16.6650,156.6300));
        stock.add(new Stock("PGE",368.80,16.6650,23.6300));
        stock.add(new Stock("PGNIG",368.80,16.6650,36.6300));
        stock.add(new Stock("PKNORLEN",368.80,16.6650,16.6300));
        stock.add(new Stock("Sample",368.80,16.6650,16.6300));
        stock.add(new Stock("Sample 2",368.80,16.6650,16.6300));
        stock.add(new Stock("Sample 3",368.80,16.6650,16.6300));
        stock.add(new Stock("Sample 4",368.80,16.6650,16.6300));
        stock.add(new Stock("Sample 5",368.80,16.6650,16.6300));
    }

    @EventListener
    public void sessionDisconnectionHandler(SessionDisconnectEvent event) {
        String sessionId = event.getSessionId();
        log.info("Disconnecting " + sessionId + "!");
        removeSession(sessionId);
    }

    public List<Stock> getStocks(){
        return stock;
    }

    public void addSession(String sessionId) {
        log.info("Added session "+sessionId);
        this.listenerClients.add(sessionId);
    }

    public void removeSession(String sessionId) {
        log.info("Removed session "+sessionId);
        this.listenerClients.remove(sessionId);
    }

    public Set<String> getListenerClients() {
        return listenerClients;
    }

    public Stock getRandomChangedStock() {
        Random rand= new Random();
        int index = rand.nextInt(stock.size());
        Stock stockObj=stock.get(index);
        int randChoice=rand.nextInt(3)+1;
        double randDouble=Math.round((rand.nextDouble()*400)+1);
        switch(randChoice){
            case 1:
                stockObj.setPurchaseLimit(randDouble);
                break;
            case 2:
                stockObj.setSaleLimit(randDouble);
                break;
            case 3:
                stockObj.setRate(randDouble);
                break;
        }
        return stockObj;
    }
}

Następnie tworzymy nasz kontroler:

package pl.stompexample.websockets.controller;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import pl.stompexample.websockets.model.Stock;
import pl.stompexample.websockets.service.StockService;

import java.util.List;

@Controller
@Slf4j
@RequestMapping("/stocks")
@AllArgsConstructor
class StockController {
    private final StockService stockService;

    @GetMapping
    @ResponseBody
    List<Stock> getAllStocks(){
        return stockService.getStocks();
    }

    //stomp
    @MessageMapping("/start")
    public void start(StompHeaderAccessor stompHeaderAccessor) {
        log.info(stompHeaderAccessor.getSessionId());
        stockService.addSession(stompHeaderAccessor.getSessionId());
    }
    @MessageMapping("/stop")
    public void stop(StompHeaderAccessor stompHeaderAccessor) {
        stockService.removeSession(stompHeaderAccessor.getSessionId());
    }

}

Pozostało nam utworzenie konfiguracji dla Websocketów jak i dla Schedulera, który będzie odpowiedzialny za losowe wysyłanie randomowych danych na frontend, czyli wywoływanie getRandomChangedStock.

Konfiguracja Websocketów:

package pl.stompexample.websockets.config;

import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.*;

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/topic");
        registry.setApplicationDestinationPrefixes("/app");
    }
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws/stocks")
                .setAllowedOrigins("*")
                .withSockJS();
    }
}

oraz konfiguracja Schedulera:

package pl.stompexample.websockets.config;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.EventListener;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.messaging.SessionDisconnectEvent;
import pl.stompexample.websockets.model.Stock;
import pl.stompexample.websockets.service.StockService;

import javax.annotation.PostConstruct;
import javax.management.Notification;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

@Configuration
@RequiredArgsConstructor
@EnableScheduling
@Slf4j
public class SchedulerConfig {
    private final SimpMessagingTemplate template;
    private final StockService stockService;

    @Scheduled(fixedDelay = 6000)
    public void dispatch(){
        for (String listener : stockService.getListenerClients()) {
            SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE);
            headerAccessor.setSessionId(listener);
            headerAccessor.setLeaveMutable(true);
            template.convertAndSendToUser(
                    listener,
                    "/topic/stocks",
                    stockService.getRandomChangedStock(),
                    headerAccessor.getMessageHeaders());
        }
    }
}

I to tak naprawdę tyle jeśli chodzi o backend. Konfiguracją CORS zajmiemy się w części 3. Jeśli masz jakieś pytania – zachęcam do zadawania ich bezpośrednio na naszej -> grupie <- . Tam też w ciągu kilku dni udostępnimy repozytorium z powyższym kodem 🙂

Przypominam również o zapisach na Newsletter, aby nie ominął Cię żaden wpis: